MQTT在SpringBoot中的配置

1. 完整示例

  1. <!--第一步：就是在Maven中的需要添加的依赖-->
  2.  <dependencies>
  3.         <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
  4.         <dependency>
  5.             <groupId>com.alibaba</groupId>
  6.             <artifactId>fastjson</artifactId>
  7.             <version>1.2.68</version>
  8.         </dependency>
  10.         <!--mqtt依赖-->
  11.         <dependency>
  12.             <groupId>org.springframework.boot</groupId>
  13.             <artifactId>spring-boot-starter-integration</artifactId>
  14.         </dependency>
  15.         <dependency>
  16.             <groupId>org.springframework.integration</groupId>
  17.             <artifactId>spring-integration-stream</artifactId>
  18.         </dependency>
  19.         <dependency>
  20.             <groupId>org.springframework.integration</groupId>
  21.             <artifactId>spring-integration-mqtt</artifactId>
  22.         </dependency>
  24.         <!--lombok-->
  25.         <dependency>
  26.             <groupId>org.projectlombok</groupId>
  27.             <artifactId>lombok</artifactId>
  28.             <optional>true</optional>
  29.         </dependency>
  31.         <dependency>
  32.             <groupId>org.springframework.boot</groupId>
  33.             <artifactId>spring-boot-starter-web</artifactId>
  34.         </dependency>
  36.         <dependency>
  37.             <groupId>org.springframework.boot</groupId>
  38.             <artifactId>spring-boot-starter-test</artifactId>
  39.             <scope>test</scope>
  40.         </dependency>
  41.     </dependencies>
  1. # 第二步： 在application.yml中，添加关于mqtt的配置 （如果mqtt的IP、port、user、password在库中则需要在实现时候查表进行） #
  3. server:
  4.   port: 8023
  5. spring:
  6.   #给项目来个名字
  7.   application:
  8.     name: mqtt
  9.   #MQTT-用户名
  10.   mqtt:
  11.     username: admin
  12.     #MQTT-密码password: password
  13.     password: public
  14.     #MQTT-服务器连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:1883，tcp://192.168.2.133:1883
  15.     url: tcp://127.0.0.1:1883
  16.     #MQTT-连接服务器默认客户端ID
  17.     client:
  18.       id: mqttId
  19.     #MQTT-默认的消息推送主题，实际可在调用接口时指定
  20.     default:
  21.       topic: topic
  22.       yihonWQM: YIHON_WQ_M
  23.     #连接超时
  24.     completionTimeout: 3000
  1. /**
  2.  * 第三步： MqttSenderAndReceiveConfig推送接受消息类
  3.  */
  4. import cn.hutool.core.util.IdUtil;
  5. import com.zxk.mqtt.MqttQosEnum;
  6. import com.zxk.mqtt.handle.MqttReceiveHandle;
  7. import com.zxk.camera.pojo.dto.EmqInfoDTO;
  8. import com.zxk.camera.service.ConfigInfoService;
  9. import lombok.extern.slf4j.Slf4j;
  10. import org.apache.commons.lang3.StringUtils;
  11. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  12. import org.springframework.beans.factory.annotation.Value;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.context.annotation.Configuration;
  15. import org.springframework.integration.annotation.IntegrationComponentScan;
  16. import org.springframework.integration.annotation.ServiceActivator;
  17. import org.springframework.integration.channel.DirectChannel;
  18. import org.springframework.integration.core.MessageProducer;
  19. import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
  20. import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
  21. import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
  22. import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
  23. import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
  24. import org.springframework.messaging.MessageChannel;
  25. import org.springframework.messaging.MessageHandler;
  27. import java.util.Arrays;
  28. import java.util.List;
  30. @Slf4j
  31. @Configuration
  32. @IntegrationComponentScan
  33. public class MqttConfiguration {
  34.     private final ConfigInfoService configInfoService;
  35.     private final MqttReceiveHandle mqttReceiveHandle;
  36.     /**
  37.      * 消息驱动
  38.      */
  39.     private MqttPahoMessageDrivenChannelAdapter adapter;
  41.    @Value("${spring.mqtt.username}")
  42.     private String username;
  44.     @Value("${spring.mqtt.password}")
  45.     private String password;
  47.     @Value("${spring.mqtt.url}")
  48.     private String hostUrl;
  50.     @Value("${spring.mqtt.client.id}")
  51.     private String clientId;
  53.     @Value("${spring.mqtt.default.topic}")
  54.     private String defaultTopic;
  55.     //水质设备主题
  56.     @Value("${spring.mqtt.default.yihonWQM}")
  57.     private String yihonWQM;
  59.     @Value("${spring.mqtt.completionTimeout}")
  60.     private int completionTimeout;   //连接超时
  63.     public MqttConfiguration(ConfigInfoService configInfoService, MqttReceiveHandle mqttReceiveHandle) {
  64.         this.configInfoService = configInfoService;
  65.         this.mqttReceiveHandle = mqttReceiveHandle;
  66.     }
  68.     /**
  69.      * mqtt 连接配置属性
  70.      */
  72.     @Bean(value = "getMqttConnectOptions")
  73.     public MqttConnectOptions getMqttConnectOptions1(){
  74.         MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
  75.         // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
  76.         mqttConnectOptions.setCleanSession(true);
  77.         // 设置超时时间 单位为秒
  78.         mqttConnectOptions.setConnectionTimeout(10);
  79.         mqttConnectOptions.setAutomaticReconnect(true);
  80.         mqttConnectOptions.setUserName(username);
  81.         mqttConnectOptions.setPassword(password.toCharArray());
  82.         mqttConnectOptions.setServerURIs(new String[]{hostUrl});
  83.         // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
  84.         mqttConnectOptions.setKeepAliveInterval(10);
  85.         // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
  86.         //mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
  87.         return mqttConnectOptions;
  90.     /**
  91.      * MQTT工厂
  92.      */
  93.     @Bean
  94.     public MqttPahoClientFactory mqttClientFactory() {
  95.         DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
  96.         factory.setConnectionOptions(getMqttConnectOptions1());
  97.         return factory;
  98.     }
  100.     /**
  101.      * MQTT信息通道（生产者）
  102.      */
  103.     @Bean
  104.     public MessageChannel mqttOutboundChannel() {
  105.         return new DirectChannel();
  106.     }
  108.     /**
  109.      * MQTT消息处理器（生产者）
  110.      */
  111.     @Bean
  112.     @ServiceActivator(inputChannel = "mqttOutboundChannel")
  113.     public MessageHandler mqttOutbound() {
  114.         MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(IdUtil.simpleUUID(), mqttClientFactory());
  115.         messageHandler.setAsync(true);
  116.         messageHandler.setDefaultTopic(defaultTopic);
  117.         messageHandler.setDefaultQos(2);
  118.         return messageHandler;
  119.     }
  121.     @Bean
  122.     public MessageChannel mqttInputChannel() {
  123.         return new DirectChannel();
  124.     }
  126.     /**
  127.      * 配置client,监听的topic
  128.      * MQTT消息订阅绑定（消费者）
  129.      */
  130.     @Bean
  131.     public MessageProducer inbound() {
  132.         if (adapter == null) {
  133.             adapter = new MqttPahoMessageDrivenChannelAdapter(IdUtil.simpleUUID() + "_inbound1", mqttClientFactory(), defaultTopic, yihonWQM);
  134.         }
  135.         String[] topics = listenTopic.split(",");
  136.         for (String topic : topics) {
  137.             if (!StringUtils.isEmpty(topic)) {
  138.                 adapter.addTopic(topic, 2);
  139.             }
  140.         }
  141.         adapter.setCompletionTimeout(completionTimeout);
  142.         DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter(2,false,"UTF-8");
  143.         defaultPahoMessageConverter.setPayloadAsBytes(true);
  144.         adapter.setConverter(defaultPahoMessageConverter);
  145.         // 设置服务质量
  146.         // 0 最多一次，数据可能丢失;
  147.         // 1 至少一次，数据可能重复;
  148.         // 2 只有一次，有且只有一次;最耗性能
  149.         adapter.setQos(1);
  150.         adapter.setOutputChannel(mqttInputChannel());
  151.         return adapter;
  152.     }
  154.     /**
  155.      * 增加监听的topic
  156.      *
  157.      * @param topicArr topic数组
  158.      */
  159.     public List<String> addListenTopic(String[] topicArr) {
  160.         if (adapter == null) {
  161.             adapter = new MqttPahoMessageDrivenChannelAdapter(IdUtil.simpleUUID() + "_inbound1", mqttClientFactory(),
  162.                     defaultTopic);
  163.         }
  164.         List<String> listTopic = Arrays.asList(adapter.getTopic());
  165.         for (String topic : topicArr) {
  166.             if (!StringUtils.isEmpty(topic)) {
  167.                 if (!listTopic.contains(topic)) {
  168.                     adapter.addTopic(topic, 2);
  169.                 }
  170.             }
  171.         }
  172.         return Arrays.asList(adapter.getTopic());
  173.     }
  175.     /**
  176.      * 移除一个监听的topic
  177.      */
  178.     public List<String> removeListenTopic(String topic) {
  179.         if (adapter == null) {
  180.             adapter = new MqttPahoMessageDrivenChannelAdapter(IdUtil.simpleUUID() + "_inbound1", mqttClientFactory(),
  181.                     defaultTopic);
  182.         }
  183.         List<String> listTopic = Arrays.asList(adapter.getTopic());
  184.         if (listTopic.contains(topic)) {
  185.             adapter.removeTopic(topic);
  186.         }
  187.         return Arrays.asList(adapter.getTopic());
  188.     }
  189.      /**
  190.      * MQTT消息处理器（消费者）
  191.      * **/
  193.     @Bean
  194.     @ServiceActivator(inputChannel = "mqttInputChannel")
  195.     public MessageHandler handler() {
  196.         return mqttReceiveHandle::handle;
  197.     }
  198. }
  1. /**
  2.  * 第四步： MqttGateway发送消息类
  3.  */
  4. import org.springframework.integration.annotation.MessagingGateway;
  5. import org.springframework.integration.mqtt.support.MqttHeaders;
  6. import org.springframework.messaging.handler.annotation.Header;
  8. /**
  9.  * mqtt发送消息
  10.  * （defaultRequestChannel = "mqttOutboundChannel" 对应config配置）
  11.  * **/
  12. @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
  13. public interface MqttGateway {
  15.     /**
  16.      * 发送信息到MQTT服务器
  17.      *
  18.      * @param data 发送的文本
  19.      */
  20.     void sendToMqtt(String data);
  22.     /**
  23.      * 发送信息到MQTT服务器
  24.      *
  25.      * @param topic 主题
  26.      * @param payload 消息主体
  27.      */
  28.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
  29.                     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. }
  1. /**
  2.  * 第五步： mqtt客户端消息处理
  3.  */
  4. import com.alibaba.fastjson.JSON;
  5. import com.alibaba.fastjson.JSONObject;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.springframework.integration.mqtt.support.MqttHeaders;
  8. import org.springframework.messaging.Message;
  9. import org.springframework.stereotype.Component;
  11. /**
  12.  * mqtt客户端消息处理类
  13.  * **/
  14. @Slf4j
  15. @Component
  16. public class MqttReceiveHandle {
  18.     public void handle(Message<?> message){
  19.         log.info("主题：{}，QOS:{}，消息接收到的数据：{}", message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC), message.getHeaders().get(MqttHeaders.RECEIVED_QOS), message.getPayload());
  20.         byte[] payLoadByte = (byte[]) message.getPayload();
  21.         String payLoad = new String(payLoadByte);
  22.         System.out.println("mqtt返回的信息有：" +JSON.toJSONString(payLoad));
  23.         JSONObject jsonMessage = JSONObject.parseObject(payLoad);
  25.         // todo: 当获取到返回信息后，继续往下执行不同返回的不同逻辑
  26.     }
  27. }

在对具体的回调函数中，继续处理业务逻辑的时候：使用注解形式来获取对象和Service（如@Autowired、@Service），直接使用注解会报java.lang.NullException错误

解决方案： 通过上下文的方式，获取进行业务处理的service！

  1. /**
  2.  * 第六步： 写一个方法，用来获取上下文
  3.  */
  5. import org.springframework.beans.BeansException;
  6. import org.springframework.context.ApplicationContext;
  7. import org.springframework.context.ApplicationContextAware;
  8. import org.springframework.stereotype.Component;
  10. @Component
  11. public class GetUserDefinedServiceUtil implements ApplicationContextAware{
  12.     private static ApplicationContext applicationContext = null;
  14.     @Override
  15.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  16.         if (null == GetUserDefinedServiceUtil.applicationContext) {
  17.             GetUserDefinedServiceUtil.applicationContext = applicationContext;
  18.         }
  19.     }
  21.     //获取applicationContext
  22.     public static ApplicationContext getApplicationContext() {
  23.         return applicationContext;
  24.     }
  27.     //通过class获取Bean.
  28.     public static <T> T getBean(Class<T> clazz){
  29.         return getApplicationContext().getBean(clazz);
  30.     }
  32.     //通过name,以及Clazz返回指定的Bean
  33.     public static <T> T getBean(String name,Class<T> clazz){
  34.         return getApplicationContext().getBean(name, clazz);
  35.     }
  36. }
  1. /**
  2.  * 在业务中，取出service对象
  3.  */
  4. xxxxService xxxxService = GetUserDefinedServiceUtil.getBean(xxxxService.class);