MQTT协议因低延迟、效率高在工业物联网领域使用的频率特别高，前面两篇文档分别对MQTT内容和MQTT服务器做了简单介绍，今天本文从实战的角度阐述如何用代码实现发送MQTT消息。
1.引入相关的依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-integration</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency>
2.在application.yml配置MQTT服务器信息
server: port: 9090 mqtt: host: tcp://127.0.0.1:1883 clientinid: mqttinId clientoutid: mqttoutid topic: virus qoslevel: 1 #MQTT 认证 username: xxx password: yyy # 10s timeout: 10000 #20s keepalive: 20
3.配置MQTT消息推送配置
@Configuration
@IntegrationComponentScan
public class MqttSenderConfig {
@Value("${mqtt.username}")
private String username;
@Value("${mqtt.password}")
private String password;
@Value("${mqtt.host}")
private String hostUrl;
@Value("${mqtt.clientinid}")
private String clientId;
@Value("${mqtt.topic}")
private String defaultTopic;
@Value("${mqtt.timeout}")
private int completionTimeout;
@Bean
public MqttConnectOptions getMqttConnectOptions(){
MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
mqttConnectOptions.setCleanSession(true);
mqttConnectOptions.setConnectionTimeout(10);
mqttConnectOptions.setKeepAliveInterval(90);
mqttConnectOptions.setAutomaticReconnect(true);
mqttConnectOptions.setUserName(username);
mqttConnectOptions.setPassword(password.toCharArray());
mqttConnectOptions.setServerURIs(new String[]{hostUrl});
mqttConnectOptions.setKeepAliveInterval(2);
return mqttConnectOptions;
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions());
return factory;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(defaultTopic);
return messageHandler;
}
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
}
4.MQTT消息推送接口
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
}
5.MQTT消息推送API
@RestController
public class MessageController {
@Autowired
MqttGateway mqttGateway;
@RequestMapping("/sendMqttMessage")
public String sendMqttMessage(String message, String topic) {
mqttGateway.sendToMqtt(message, topic);
return "ok";
}
}
测试
接下来就可以在POSTMAN中进行测试了，输入消息内容和主题，就可以在相应的频道发送消息了。如果使用其它的消息客户端进行测试的话，可以接受到消息
