一、安装配置

1、依赖配置

编辑项目的 pom.xml 文件，添加如下依赖：

  1. <dependency>
  2.     <groupId>org.springframework.integration</groupId>
  3.     <artifactId>spring-integration-stream</artifactId>
  4. </dependency>
  5. <dependency>
  6.     <groupId>org.springframework.integration</groupId>
  7.     <artifactId>spring-integration-mqtt</artifactId>
  8. </dependency>

2、配置 MQTT 服务器基本信息

编辑项目的 application.properties 文件，增加 MQTT 服务器配置信息：

  1. server.port=10001
  2. # 用户名（这里为空）
  3. mqtt.username=
  4. # 密码（这里为空）
  5. mqtt.password=
  6. # 推送信息的连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:1883,tcp://x.x.x.x:1883
  7. mqtt.host=tcp://x.x.x.x:1883
  9. # 生产者连接服务器默认客户端ID
  10. mqtt.sender.clientId=mqttProducer
  11. # 默认的推送主题，实际可在调用接口时指定
  12. mqtt.sender.defaultTopic=hangge
  14. # 消费者连接服务器默认客户端ID（这里使用随机数）
  15. mqtt.receiver.clientId=${random.value}
  16. # 默认的接收主题，实际可在调用接口时指定
  17. mqtt.receiver.defaultTopic=hangge

二、实现 MQTT 消息的发送

基础连接配置

  1. package com.gyz.mqtt.config;
  3. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  4. import org.springframework.beans.factory.annotation.Value;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  7. import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
  8. import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
  10. /**
  11.  * 配置连接
  12.  *
  13.  * @author gong_yuzhuo
  14.  */
  15. @Configuration
  16. public class MqttBaseConfig {
  18.     /**
  19.      * 发布的bean名称
  20.      */
  21.     public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
  23.     /** 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息 */
  24.     private static final byte[] WILL_DATA;
  26.     static {
  27.         WILL_DATA = "offline".getBytes();
  28.     }
  30.     @Value("${mqtt.username}")
  31.     private String username;
  33.     @Value("${mqtt.password}")
  34.     private String password;
  36.     @Value("${mqtt.host}")
  37.     private String url;
  39.     @Value("${mqtt.sender.clientId}")
  40.     private String clientId;
  42.     @Value("${mqtt.sender.defaultTopic}")
  43.     private String defaultTopic;
  45.     @Bean
  46.     public MqttPahoClientFactory factory() {
  47.         DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
  48.         MqttConnectOptions options = new MqttConnectOptions();
  49.         // 设置连接的用户名
  50.         if (!username.trim().equals("")) {
  51.             options.setUserName(username);
  52.         }
  53.         // 设置连接的密码
  54.         options.setPassword(password.toCharArray());
  55.         // 设置连接的地址
  56.         options.setServerURIs(new String[]{url});
  57.         // 设置超时时间 单位为秒
  58.         options.setConnectionTimeout(10);
  59.         // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
  60.         // 但这个方法并没有重连的机制
  61.         options.setKeepAliveInterval(20);
  62.         // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
  63.         options.setWill("willTopic", WILL_DATA, 2, false);
  64.         factory.setConnectionOptions(options);
  65.         return factory;
  66.     }
  67. }

MqttSenderConfig.java（MQTT 消息推送配置类）

  1. package com.gyz.mqtt.config;
  3. import org.springframework.beans.factory.annotation.Value;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.integration.annotation.ServiceActivator;
  7. import org.springframework.integration.channel.DirectChannel;
  8. import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
  9. import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
  10. import org.springframework.messaging.MessageChannel;
  11. import org.springframework.messaging.MessageHandler;
  13. /**
  14.  * MQTT配置，生产者
  15.  */
  16. @Configuration
  17. public class MqttSenderConfig {
  19.     /**
  20.      * 发布的bean名称
  21.      */
  22.     public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
  24.     /** 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息 */
  25.     private static final byte[] WILL_DATA;
  27.     static {
  28.         WILL_DATA = "offline".getBytes();
  29.     }
  31.     @Value("${mqtt.username}")
  32.     private String username;
  34.     @Value("${mqtt.password}")
  35.     private String password;
  37.     @Value("${mqtt.host}")
  38.     private String url;
  40.     @Value("${mqtt.sender.clientId}")
  41.     private String clientId;
  43.     @Value("${mqtt.sender.defaultTopic}")
  44.     private String defaultTopic;
  47.     /**
  48.      * MQTT信息通道（生产者）
  49.      */
  50.     @Bean(name = CHANNEL_NAME_OUT)
  51.     public MessageChannel mqttOutboundChannel() {
  52.         return new DirectChannel();
  53.     }
  55.     /**
  56.      * MQTT消息处理器（生产者）
  57.      */
  58.     @Bean
  59.     @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
  60.     public MessageHandler mqttOutbound(MqttPahoClientFactory factory) {
  61.         MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
  62.                 clientId,
  63.                 factory);
  64.         messageHandler.setAsync(true);
  65.         messageHandler.setDefaultTopic(defaultTopic);
  66.         return messageHandler;
  67.     }
  69. }

2、IMqttSender（消息推送接口类）

  1. package com.gyz.mqtt.service;
  3. import com.gyz.mqtt.config.MqttSenderConfig;
  4. import org.springframework.integration.annotation.MessagingGateway;
  5. import org.springframework.integration.mqtt.support.MqttHeaders;
  6. import org.springframework.messaging.handler.annotation.Header;
  7. import org.springframework.stereotype.Component;
  9. /**
  10.  * @author gong_yuzhuo
  11.  */
  12. @Component
  13. @MessagingGateway(defaultRequestChannel = MqttSenderConfig.CHANNEL_NAME_OUT)
  14. public interface IMqttSender {
  16.     /**
  17.      * 发送信息到MQTT服务器
  18.      *
  19.      * @param data 发送的文本
  20.      */
  21.     void sendToMqtt(String data);
  23.     /**
  24.      * 发送信息到MQTT服务器
  25.      *
  26.      * @param topic     主题
  27.      * @param payload   消息主体
  28.      */
  29.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
  31.     /**
  32.      * 发送信息到MQTT服务器
  33.      *
  34.      * @param topic     主题
  35.      * @param QOS       对消息处理的几种机制：
  36.      *                      0 表示的是订阅者没收到消息不会再次发送，消息会丢失。
  37.      *                      1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。
  38.      *                      2 多了一次去重的动作，确保订阅者收到的消息有一次。
  39.      * @param payload   消息主体
  40.      */
  41.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
  42.                     @Header(MqttHeaders.QOS) int QOS,
  43.                     String payload);
  44. }

3、HelloController.java（测试类）

  1. package com.gyz.mqtt.controller;
  3. import com.gyz.mqtt.service.IMqttSender;
  5. import org.springframework.web.bind.annotation.PathVariable;
  6. import org.springframework.web.bind.annotation.RequestMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  9. import javax.annotation.Resource;
  11. /**
  12.  * 测试发送消息
  13.  *
  14.  * @author gong_yuzhuo
  15.  */
  16. @RestController
  17. public class HelloController {
  19.     /**
  20.      * 注入发送MQTT的Bean
  21.      */
  22.     @Resource
  23.     private IMqttSender imqttSender;
  25.     /**
  26.      * 发送自定义消息内容（使用默认主题）
  27.      * @param data
  28.      */
  29.     @RequestMapping("/test1/{data}")
  30.     public void test1(@PathVariable("data") String data) {
  31.         imqttSender.sendToMqtt(data);
  32.     }
  34.     /**
  35.      * 发送自定义消息内容，且指定主题
  36.      * @param topic
  37.      * @param data
  38.      */
  39.     @RequestMapping("/test2/{topic}/{data}")
  40.     public void test2(@PathVariable("topic") String topic, @PathVariable("data") String data) {
  41.         imqttSender.sendToMqtt(topic, data);
  42.     }
  44. }

4、测试运行

  1. 项目启动后，使用 MQTTBox订阅“hangge”这个主题，然后使用浏览器访问 http://localhost:10001/test1/abcd1234 可以看到 MQTTBox 这边可以成功接收到发布的消息：

MQTTBox的安装和使用

五、SpringBoot - 集成MQTT（发布消息） - 图1

  1. 接着我们访问 http://localhost:10001/test2/china/abcd1234，这次除了发送自定消息外还指定了主题（china，而不是使用默认主题），MQTTBox 这边订阅该主题并显示结果：

五、SpringBoot - 集成MQTT（发布消息） - 图2

参考文章：

https://www.hangge.com/blog/cache/detail_2610.html

https://www.hangge.com/blog/cache/detail_2611.html