如何用代码实现发送MQTT消息

MQTT协议因低延迟、效率高在工业物联网领域使用的频率特别高，前面两篇文档分别对MQTT内容和MQTT服务器做了简单介绍，今天本文从实战的角度阐述如何用代码实现发送MQTT消息。

1.引入相关的依赖

org.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-integrationorg.springframework.integrationspring-integration-mqtt

2.在application.yml配置MQTT服务器信息

server:  port: 9090mqtt:  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@IntegrationComponentScanpublic 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

@RestControllerpublic class MessageController {    @Autowired    MqttGateway mqttGateway;    @RequestMapping("/sendMqttMessage")    public String sendMqttMessage(String message, String topic) {        mqttGateway.sendToMqtt(message, topic);        return "ok";    }}

测试

接下来就可以在POSTMAN中进行测试了，输入消息内容和主题，就可以在相应的频道发送消息了。如果使用其它的消息客户端进行测试的话，可以接受到消息