devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【MQTT】SpringBoot整合MQTT（EMQX）_springboot emqx

作者：weixin_40725706 | 2024-06-23 00:25:08

赞

踩

springboot emqx

最近在做MQTT对接，然后发送消息，然后参考网上的实战文章进行了一下整理。

文章主要参考自（https://www.codetd.com/article/13550340），然后自己做了些许更改。

1、整合准备

SpringBoot：2.2.2.RELEASE

MQTT平台：EMQX4.4.1（Docker运行）

虚拟机服务器：Centos7（192.168.56.102 ）

发送端：cloud-mqtt-send8001

接收端：cloud-mqtt-accept8002

2、发送端：cloud-mqtt-send8002

导入POM依赖：

  1.     <!-- mqtt -->
  2.     <dependency>
  3.        <groupId>org.springframework.boot</groupId>
  4.        <artifactId>spring-boot-starter-integration</artifactId>
  5.     </dependency>
  6.     <dependency>
  7.        <groupId>org.springframework.integration</groupId>
  8.        <artifactId>spring-integration-stream</artifactId>
  9.     </dependency>
  10.     <dependency>
  11.        <groupId>org.springframework.integration</groupId>
  12.        <artifactId>spring-integration-mqtt</artifactId>
  13.     </dependency>
  14.     <!--配置文件报错问题-->
  15.     <dependency>
  16.         <groupId>org.springframework.boot</groupId>
  17.         <artifactId>spring-boot-configuration-processor</artifactId>
  18.         <optional>true</optional>
  19.     </dependency>
  20.     <dependency>
  21.       <groupId>org.projectlombok</groupId>
  22.       <artifactId>lombok</artifactId>
  23.       <version>1.18.22</version>
  24.       <scope>provided</scope>
  25.   </dependency>

设置YML

  1. server:
  2.   port: 8001
  3. ​
  4. spring:
  5.   application:
  6.     name: mqtt-send
  7. ​
  8. #mqtt属性配置
  9. mqtt:
  10.   hostUrl: tcp://192.168.56.102:1883
  11.   username: admin
  12.   password: public
  13.   clientid: mqtt_send_client
  14.   cleanSession: true
  15.   reconnect: true
  16.   #连接超时
  17.   timeout: 1000
  18.   #设置会话心跳时间
  19.   keepalive: 100
  20.   defaultTopic: client:report:1
  21.   isOpen: true
  22.   qos: 1

新建配置类：MqttProperties.java

  1. @Data
  2. @Component
  3. @ConfigurationProperties(prefix = "mqtt")
  4. public class MqttProperties {
  5. ​
  6.     /**
  7.      * 用户名
  8.      */
  9.     private String username;
  10. ​
  11.     /**
  12.      * 密码
  13.      */
  14.     private String password;
  15. ​
  16.     /**
  17.      * 连接地址
  18.      */
  19.     private String hostUrl;
  20. ​
  21.     /**
  22.      * 客户端Id，同一台服务器下，不允许出现重复的客户端id
  23.      */
  24.     private String clientId;
  25. ​
  26.     /**
  27.      * 默认连接主题
  28.      */
  29.     private String defaultTopic;
  30. ​
  31.     /**
  32.      * 超时时间
  33.      */
  34.     private int timeout;
  35. ​
  36.     /**
  37.      * 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端
  38.      * 发送个消息判断客户端是否在线，但这个方法并没有重连的机制
  39.      */
  40.     private int keepAlive;
  41. ​
  42.     /**
  43.      * 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连
  44.      * 接记录，这里设置为true表示每次连接到服务器都以新的身份连接
  45.      */
  46.     private Boolean cleanSession;
  47. ​
  48.     /**
  49.      * 是否断线重连
  50.      */
  51.     private Boolean reconnect;
  52. ​
  53.     /**
  54.      * 启动的时候是否关闭mqtt
  55.      */
  56.     private Boolean isOpen;
  57. ​
  58.     /**
  59.      * 连接方式
  60.      */
  61.     private Integer qos;
  62. }

添加MQTT发送客户端：MqttSendClient：

  1. @Slf4j
  2. @Component
  3. public class MqttSendClient {
  4. ​
  5. ​
  6.     @Autowired
  7.     private MqttSendCallBack mqttSendCallBack;
  8. ​
  9.     @Autowired
  10.     private MqttProperties mqttProperties;
  11. ​
  12.     private static MqttClient mqttClient;
  13. ​
  14.     private static MqttClient getClient() {
  15. ​
  16. ​
  17.         return mqttClient;
  18.     }
  19. ​
  20.     private static void setClient(MqttClient client) {
  21. ​
  22. ​
  23.         MqttSendClient.mqttClient = client;
  24.     }
  25. ​
  26.     /**
  27.      * 客户端连接
  28.      * @return
  29.      */
  30.     public void connect(){
  31.         MqttClient client = null;
  32. ​
  33.         try {
  34. ​
  35. ​
  36.             //String uuid = UUID.randomUUID().toString().replaceAll("-",""); //设置每一个客户端的id
  37.             client = new MqttClient(mqttProperties.getHostUrl(),mqttProperties.getClientId() , new MemoryPersistence());
  38.             MqttConnectOptions options = new MqttConnectOptions();
  39.             options.setUserName(mqttProperties.getUsername());
  40.             options.setPassword(mqttProperties.getPassword().toCharArray());
  41.             options.setConnectionTimeout(mqttProperties.getTimeout());
  42.             options.setKeepAliveInterval(mqttProperties.getKeepAlive());
  43.             options.setCleanSession(true);
  44.             options.setAutomaticReconnect(false);
  45. ​
  46.             MqttSendClient.setClient(client);
  47.             try {
  48. ​
  49. ​
  50.                 // 设置回调
  51.                 client.setCallback(mqttSendCallBack);
  52.                 client.connect(options);
  53.             } catch (Exception e) {
  54. ​
  55. ​
  56.                 e.printStackTrace();
  57.             }
  58.         } catch (Exception e) {
  59. ​
  60.             e.printStackTrace();
  61.         }
  62.     }
  63. ​
  64. ​
  65. ​
  66. ​
  67.     /**
  68.      * 发布，默认qos为0，非持久化
  69.      *
  70.      * @param topic 主题名
  71.      * @param pushMessage 消息
  72.      */
  73.     public void publish(String topic, String pushMessage) {
  74.         publish(0, false, topic, pushMessage);
  75.     }
  76. ​
  77.     /**
  78.      * 发布
  79.      *
  80.      * @param qos
  81.      * @param retained
  82.      * @param topic
  83.      * @param pushMessage
  84.      */
  85.     public void publish(int qos, boolean retained, String topic, String pushMessage) {
  86.         MqttMessage message = new MqttMessage();
  87.         message.setQos(qos);
  88.         message.setRetained(retained);
  89.         message.setPayload(pushMessage.getBytes());
  90.         MqttTopic mTopic = MqttSendClient.getClient().getTopic(topic);
  91.         if (null == mTopic) {
  92.             log.error("主题不存在:{}",mTopic);
  93.         }
  94.         try {
  95.             mTopic.publish(message);
  96.             log.info("消息发送成功");
  97.         } catch (Exception e) {
  98.             log.error("mqtt发送消息异常:",e);
  99.         }
  100.     }
  101. }

添加MQTT发送客户端回调类：MqttSendCallBack

  1. @Slf4j
  2. @Component
  3. public class MqttSendCallBack implements MqttCallbackExtended {
  4. ​
  5. ​
  6.     /**
  7.      * 链接EMQ服务器后触发
  8.      * @param reconnect
  9.      * @param serverURI
  10.      */
  11.     @Override
  12.     public void connectComplete(boolean reconnect, String serverURI) {
  13.         log.info("————————————————-ClientID:{}——————————————"+"链接成功");
  14.     }
  15. ​
  16.     /**
  17.      * 客户端连接断开后触发
  18.      * 这里可以做重新链接操作
  19.      */
  20.     @Override
  21.     public void connectionLost(Throwable cause) {
  22.         log.error("【MQTT-发送端】链接断开！");
  23.     }
  24. ​
  25.     @Override
  26.     public void messageArrived(String topic, MqttMessage message) throws Exception {
  27.         log.info("【MQTT-发送端】接收消息主题 : " + topic);
  28.         log.info("【MQTT-发送端】接收消息Qos : " + message.getQos());
  29.         log.info("【MQTT-发送端】接收消息内容 : " + new String(message.getPayload()));
  30.     }
  31. ​
  32.     /**
  33.      * 发送消息回调
  34.      * @param token
  35.      */
  36.     @Override
  37.     public void deliveryComplete(IMqttDeliveryToken token) {
  38. ​
  39.         String[] topics = token.getTopics();
  40. ​
  41.         if (topics!=null && topics.length>0){
  42.             for (String topic : topics) {
  43. ​
  44. ​
  45.                 log.info("【MQTT-发送端】向主题：" + topic + "发送消息成功！");
  46.             }
  47.         }
  48. ​
  49.         try {
  50. ​
  51. ​
  52.             MqttMessage message = token.getMessage();
  53.             byte[] payload = message.getPayload();
  54.             String s = new String(payload, "UTF-8");
  55.             log.info("【MQTT-发送端】消息的内容是：" + s);
  56.         } catch (MqttException e) {
  57. ​
  58. ​
  59.             e.printStackTrace();
  60.         } catch (UnsupportedEncodingException e) {
  61. ​
  62. ​
  63.             e.printStackTrace();
  64.         }
  65.     }
  66. }
  67. ​

添加：MqttCondition

  1. public class MqttCondition implements Condition {
  2.     @Override
  3.     public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
  4. ​
  5.         System.out.println("MqttCondition。。。。");
  6. ​
  7.         //1、能获取到ioc使用的beanfactory
  8.         ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
  9.         //2、获取类加载器
  10.         ClassLoader classLoader = context.getClassLoader();
  11.         //3、获取当前环境信息
  12.         Environment environment = context.getEnvironment();
  13.         String isOpen = environment.getProperty("mqtt.isOpen");
  14.         return Boolean.valueOf(isOpen);
  15.     }
  16. }

添加MQTT配置类：MqttConfig

  1. @Configuration
  2. public class MqttConfig {
  3. ​
  4.     @Autowired
  5.     private MqttSendClient mqttSendClient;
  6. ​
  7.     @Conditional(MqttCondition.class)
  8.     @Bean
  9.     public MqttSendClient getMqttSendClient(){
  10.         mqttSendClient.connect();
  11.         return mqttSendClient;
  12.     }
  13. }
  14. ​

主启动类

  1. @SpringBootApplication
  2. public class MqttSendApplication {
  3. ​
  4.     public static void main(String[] args) {
  5.         SpringApplication.run(MqttSendApplication.class, args);
  6.     }
  7. }

启动项目，链接MQTT服务器成功。

项目整体代码结构如下：

 

3、接收端：cloud-mqtt-accept8002

导入POM

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-web</artifactId>
  5.     </dependency>
  6.     <dependency>
  7.         <groupId>org.springframework.boot</groupId>
  8.         <artifactId>spring-boot-starter-actuator</artifactId>
  9.     </dependency>
  10.     <dependency>
  11.         <groupId>org.springframework.boot</groupId>
  12.         <artifactId>spring-boot-devtools</artifactId>
  13.         <scope>runtime</scope>
  14.         <optional>true</optional>
  15.     </dependency>
  16.     <dependency>
  17.         <groupId>org.projectlombok</groupId>
  18.         <artifactId>lombok</artifactId>
  19.         <optional>true</optional>
  20.     </dependency>
  21.     <dependency>
  22.         <groupId>org.springframework.boot</groupId>
  23.         <artifactId>spring-boot-starter-test</artifactId>
  24.         <scope>test</scope>
  25.     </dependency>
  26.     <!--MQTT客户端工具-->
  27.     <dependency>
  28.         <groupId>org.eclipse.paho</groupId>
  29.         <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  30.         <version>1.2.2</version>
  31.     </dependency>
  32.     <dependency>
  33.         <groupId>org.springframework.boot</groupId>
  34.         <artifactId>spring-boot-configuration-processor</artifactId>
  35.         <optional>true</optional>
  36.     </dependency>
  37. </dependencies>

设置YML

  1. server:
  2.   port: 8002
  3. ​
  4. spring:
  5.   application:
  6.     name: mqtt-accept
  7. ​
  8. ​
  9. mqtt:
  10.   hostUrl: tcp://192.168.56.102:1883
  11.   username: admin
  12.   password: public
  13.   clientid: mqtt_accept_client
  14.   cleanSession: true
  15.   reconnect: true
  16.   #连接超时
  17.   timeout: 1000
  18.   #设置会话心跳时间
  19.   keepalive: 100
  20.   defaultTopic: client:report:1
  21.   isOpen: true
  22.   qos: 1

属性配置文件：MqttProperties.java

  1. @Data
  2. @Component
  3. @ConfigurationProperties(prefix = "mqtt")
  4. public class MqttProperties {
  5. ​
  6.     /**
  7.      * 用户名
  8.      */
  9.     private String username;
  10. ​
  11.     /**
  12.      * 密码
  13.      */
  14.     private String password;
  15. ​
  16.     /**
  17.      * 连接地址
  18.      */
  19.     private String hostUrl;
  20. ​
  21.     /**
  22.      * 客户端Id，同一台服务器下，不允许出现重复的客户端id
  23.      */
  24.     private String clientId;
  25. ​
  26.     /**
  27.      * 默认连接主题
  28.      */
  29.     private String defaultTopic;
  30. ​
  31.     /**
  32.      * 超时时间
  33.      */
  34.     private int timeout;
  35. ​
  36.     /**
  37.      * 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端
  38.      * 发送个消息判断客户端是否在线，但这个方法并没有重连的机制
  39.      */
  40.     private int keepAlive;
  41. ​
  42.     /**
  43.      * 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连
  44.      * 接记录，这里设置为true表示每次连接到服务器都以新的身份连接
  45.      */
  46.     private Boolean cleanSession;
  47. ​
  48.     /**
  49.      * 是否断线重连
  50.      */
  51.     private Boolean reconnect;
  52. ​
  53.     /**
  54.      * 启动的时候是否关闭mqtt
  55.      */
  56.     private Boolean isOpen;
  57. ​
  58.     /**
  59.      * 连接方式
  60.      */
  61.     private Integer qos;
  62. }
  63. ​

添加MQTT接收客户端：MqttAcceptClient

  1. @Slf4j
  2. @Component
  3. public class MqttAcceptClient {
  4. ​
  5. ​
  6.     @Autowired
  7.     private MqttAcceptCallback mqttAcceptCallback;
  8. ​
  9.     @Autowired
  10.     private MqttProperties mqttProperties;
  11. ​
  12. ​
  13.     private static MqttClient mqttClient;
  14. ​
  15.     public static MqttClient getMqttClient() {
  16.         return mqttClient;
  17.     }
  18. ​
  19.     public static void setMqttClient(MqttClient mqttClient) {
  20.         MqttAcceptClient.mqttClient = mqttClient;
  21.     }
  22. ​
  23. ​
  24.     /**
  25.      * 客户端连接
  26.      */
  27.     public void connect() {
  28. ​
  29. ​
  30.         MqttClient client;
  31.         try {
  32. ​
  33. ​
  34.             client = new MqttClient(mqttProperties.getHostUrl(), mqttProperties.getClientId(), new MemoryPersistence());
  35.             MqttConnectOptions options = new MqttConnectOptions();
  36.             options.setUserName(mqttProperties.getUsername());
  37.             options.setPassword(mqttProperties.getPassword().toCharArray());
  38.             options.setConnectionTimeout(mqttProperties.getTimeout());
  39.             options.setKeepAliveInterval(mqttProperties.getKeepAlive());
  40.             options.setAutomaticReconnect(mqttProperties.getReconnect());
  41.             options.setCleanSession(mqttProperties.getCleanSession());
  42.             MqttAcceptClient.setMqttClient(client);
  43.             try {
  44. ​
  45. ​
  46.                 // 设置回调
  47.                 client.setCallback(mqttAcceptCallback);
  48.                 client.connect(options);
  49.             } catch (Exception e) {
  50. ​
  51. ​
  52.                 e.printStackTrace();
  53.             }
  54.         } catch (Exception e) {
  55. ​
  56. ​
  57.             e.printStackTrace();
  58.         }
  59.     }
  60. ​
  61.     /**
  62.      * 重新连接
  63.      */
  64.     public void reconnection() {
  65. ​
  66. ​
  67.         try {
  68.             mqttClient.connect();
  69.         } catch (MqttException e) {
  70. ​
  71. ​
  72.             e.printStackTrace();
  73.         }
  74.     }
  75. ​
  76.     /**
  77.      * 订阅某个主题
  78.      *
  79.      * @param topic 主题
  80.      * @param qos   连接方式
  81.      */
  82.     public void subscribe(String topic, int qos) {
  83. ​
  84. ​
  85.         log.info("==============开始订阅主题==============" + topic);
  86.         try {
  87. ​
  88. ​
  89.             mqttClient.subscribe(topic, qos);
  90.         } catch (MqttException e) {
  91. ​
  92. ​
  93.             e.printStackTrace();
  94.         }
  95.     }
  96. ​
  97.     /**
  98.      * 取消订阅某个主题
  99.      *
  100.      * @param topic
  101.      */
  102.     public void unsubscribe(String topic) {
  103. ​
  104. ​
  105.         log.info("==============开始取消订阅主题==============" + topic);
  106.         try {
  107. ​
  108. ​
  109.             mqttClient.unsubscribe(topic);
  110.         } catch (MqttException e) {
  111. ​
  112. ​
  113.             e.printStackTrace();
  114.         }
  115.     }
  116. }
  117. ​

添加mqtt接受服务的回调类:MqttAcceptCallback

  1. @Slf4j
  2. @Component
  3. public class MqttAcceptCallback implements MqttCallbackExtended {
  4.     @Autowired
  5.     private MqttAcceptClient mqttAcceptClient;
  6. ​
  7.     /**
  8.      * 客户端断开后触发
  9.      *
  10.      * @param throwable
  11.      */
  12.     @Override
  13.     public void connectionLost(Throwable throwable) {
  14. ​
  15. ​
  16.         log.info("【MQTT-消费端】连接断开，可以做重连");
  17.         if (MqttAcceptClient.getMqttClient() == null || !MqttAcceptClient.getMqttClient().isConnected()) {
  18. ​
  19. ​
  20.             log.info("【MQTT-消费端】emqx重新连接....................................................");
  21.             mqttAcceptClient.reconnection();
  22.         }
  23.     }
  24. ​
  25.     /**
  26.      * 客户端收到消息触发
  27.      *
  28.      * @param topic       主题
  29.      * @param mqttMessage 消息
  30.      */
  31.     @Override
  32.     public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
  33. ​
  34. ​
  35.         log.info("【MQTT-消费端】接收消息主题 : " + topic);
  36.         log.info("【MQTT-消费端】接收消息Qos : " + mqttMessage.getQos());
  37.         log.info("【MQTT-消费端】接收消息内容 : " + new String(mqttMessage.getPayload()));
  38. //        int i = 1/0;
  39.     }
  40. ​
  41.     /**
  42.      * 发布消息成功
  43.      *
  44.      * @param token token
  45.      */
  46.     @Override
  47.     public void deliveryComplete(IMqttDeliveryToken token) {
  48. ​
  49. ​
  50.         String[] topics = token.getTopics();
  51.         for (String topic : topics) {
  52. ​
  53. ​
  54.             log.info("【MQTT-消费端】向主题：" + topic + "发送消息成功！");
  55.         }
  56.         try {
  57. ​
  58. ​
  59.             MqttMessage message = token.getMessage();
  60.             byte[] payload = message.getPayload();
  61.             String s = new String(payload, "UTF-8");
  62.             log.info("【MQTT-消费端】消息的内容是：" + s);
  63.         } catch (MqttException e) {
  64. ​
  65. ​
  66.             e.printStackTrace();
  67.         } catch (UnsupportedEncodingException e) {
  68. ​
  69. ​
  70.             e.printStackTrace();
  71.         }
  72.     }
  73. ​
  74.     /**
  75.      * 连接emq服务器后触发
  76.      *
  77.      * @param b
  78.      * @param s
  79.      */
  80.     @Override
  81.     public void connectComplete(boolean b, String s) {
  82. ​
  83.         System.out.println("s: " + s);
  84. ​
  85.         log.info("--------------------【MQTT-消费端】连接成功！--------------------");
  86.         // 以/#结尾表示订阅所有以test开头的主题
  87.         // 订阅所有机构主题
  88.         mqttAcceptClient.subscribe("test_queue", 0);
  89.     }
  90. }

MqttCondition

  1. public class MqttCondition implements Condition {
  2.     @Override
  3.     public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
  4. ​
  5.         System.out.println("MqttCondition。。。。");
  6. ​
  7.         //1、能获取到ioc使用的beanfactory
  8.         ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
  9.         //2、获取类加载器
  10.         ClassLoader classLoader = context.getClassLoader();
  11.         //3、获取当前环境信息
  12.         Environment environment = context.getEnvironment();
  13.         String isOpen = environment.getProperty("mqtt.isOpen");
  14.         return Boolean.valueOf(isOpen);
  15.     }
  16. }

MQTT启动配置类：MqttConfig.java

  1. @Configuration
  2. public class MqttConfig {
  3. ​
  4.     @Autowired
  5.     private MqttAcceptClient mqttAcceptClient;
  6. ​
  7.     @Conditional(MqttCondition.class)
  8.     @Bean
  9.     public MqttAcceptClient getMqttAcceptClient(){
  10. ​
  11.         mqttAcceptClient.connect();
  12.         //mqttAcceptClient.subscribe("test_queue",0);
  13.         return mqttAcceptClient;
  14.     }
  15. }

主启动类

  1. package com.xlh.springcloud;
  2. ​
  3. import org.springframework.boot.SpringApplication;
  4. import org.springframework.boot.autoconfigure.SpringBootApplication;
  5. ​
  6. @SpringBootApplication
  7. public class MqttAcceptApplication {
  8. ​
  9.     public static void main(String[] args) {
  10.         SpringApplication.run(MqttAcceptApplication.class,args);
  11.     }
  12. }

启动项目，链接MQTT服务器，然后订阅主题：

 

项目整体结构如下：

 

4、项目测试

分别启动发送端和接收端以后，EMQX监控平台如下：

 

然后调用发送端测试接口：http://localhost:8001/send

发送端cloud-mqtt-send8001运行结果如下：

 

接收端cloud-mqtt-accept8002运行结果如下：

可以看到接收端成功的接收到消息

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/748088
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号