MQTT协议及其使用案例

MQTT 概述

MQTT是基于TCP/IP协议栈构建的异步通信消息协议，是一种轻量级的发布、订阅信息传输协议。 可以在不可靠的网络环境中进行扩展，适用于设备硬件存储空间或网络带宽有限的场景。 使用MQTT协议，消息发送者与接收者不受时间和空间的限制。
Docker 部署 MQTT（采用docker-compose.yml）

1. version: "3"
2. services:
3.     mqtt:
4.         image: eclipse-mosquitto
5.         container_name: mqtt
6.         privileged: true
7.         ports:
8.             - 1883:1883
9.             - 9001:9001
10.         volumes:
11.             - ./config:/mosquitto/config
12.             - ./data:/mosquitto/data
13.             - ./log:/mosquitto/log

复制代码

• 文件夹
  
  
• 创建 config/mosquitto.conf
  

1. persistence true
2. listener 1883
3. persistence_location /mosquitto/data
4. log_dest file /mosquitto/log/mosquitto.log
6. # 关闭匿名模式
7. # allow_anonymous true
8. # 指定密码文件
9. password_file /mosquitto/config/pwfile.conf

复制代码

• docker部署执行：docker compose up -d
  
• 设置访问权限(用户名：admin，密码：admin123)
  

1. docker exec -it mqtt sh
2. touch /mosquitto/config/pwfile.conf
3. chmod -R 755 /mosquitto/config/pwfile.conf
4. mosquitto_passwd -b /mosquitto/config/pwfile.conf admin admin123

复制代码

• 重启mqtt容器：docker compose restart
  

Springboot 整合

• 依赖
  

1.     <parent>
2.         <groupId>org.springframework.boot</groupId>
3.         <artifactId>spring-boot-starter-parent</artifactId>
4.         <version>2.5.5</version>
5.         <relativePath/>
6.     </parent>
7.    
8.     <dependencies>
9.                    
10.         <dependency>
11.             <groupId>org.springframework.integration</groupId>
12.             <artifactId>spring-integration-mqtt</artifactId>
13.         </dependency>
14.         
15.         <dependency>
16.             <groupId>org.projectlombok</groupId>
17.             <artifactId>lombok</artifactId>
18.         </dependency>
19.         
20.         <dependency>
21.             <groupId>org.springframework.boot</groupId>
22.             <artifactId>spring-boot-starter</artifactId>
23.         </dependency>
24.         <dependency>
25.             <groupId>org.springframework.boot</groupId>
26.             <artifactId>spring-boot-starter-web</artifactId>
27.         </dependency>
28.         <dependency>
29.             <groupId>com.alibaba</groupId>
30.             <artifactId>fastjson</artifactId>
31.             <version>1.2.62</version>
32.         </dependency>
34.         
35.         <dependency>
36.             <groupId>org.apache.httpcomponents</groupId>
37.             <artifactId>httpclient</artifactId>
38.         </dependency>
39.         
40.         <dependency>
41.             <groupId>junit</groupId>
42.             <artifactId>junit</artifactId>
43.             <version>4.12</version>
44.         </dependency>
45.         <dependency>
46.             <groupId>org.springframework</groupId>
47.             <artifactId>spring-test</artifactId>
48.         </dependency>
49.         <dependency>
50.             <groupId>org.springframework.boot</groupId>
51.             <artifactId>spring-boot-test</artifactId>
52.         </dependency>
53.     </dependencies>

复制代码

• 配置文件
  

1. mqtt.host=tcp://127.0.0.1:1883
2. mqtt.clientId=mqttx_a071ba88
3. mqtt.username=admin
4. mqtt.password=admin123
5. mqtt.topic=test_topic
6. mqtt.timeout=36000
7. mqtt.keepAlive=6000

复制代码

• 配置类
  

1. @Slf4j
2. @Configuration
3. public class MyMqttConfiguration {
4.     @Value("${mqtt.host}")
5.     String broker;
6.     @Value("${mqtt.clientId}")
7.     String clientId;
8.     @Value("${mqtt.username}")
9.     String username;
10.     @Value("${mqtt.password}")
11.     String password;
12.     @Value("${mqtt.timeout}")
13.     Integer timeout;
14.     @Value("${mqtt.keepAlive}")
15.     Integer keepAlive;
16.     @Value("${mqtt.topic}")
17.     String topic;
18.     @Autowired
19.     MyHandle myHandle;
21.     @Bean
22.     public MyMqttClient myMqttClient(){
23.         MyMqttClient mqttClient = new MyMqttClient(broker, username, password, clientId, timeout, keepAlive,myHandle);
24.         for (int i = 0; i < 10; i++) {
25.             try {
26.                 mqttClient.connect();
27.                 mqttClient.subscribe(topic,0);
28.                 return mqttClient;
29.             } catch (MqttException e) {
30.                 log.error("MQTT connect exception,connect time = " + i);
31.                 try {
32.                     Thread.sleep(2000);
33.                 } catch (InterruptedException e1) {
34.                     e1.printStackTrace();
35.                 }
36.             }
37.         }
38.         return mqttClient;
39.     }
41. }

复制代码

• 客户端
  

1. import lombok.extern.slf4j.Slf4j;
2. import org.eclipse.paho.client.mqttv3.*;
3. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
4. import org.springframework.util.ObjectUtils;
6. @Slf4j
7. public class MyMqttClient {
8.     private static MqttClient client;
9.     private String host;
10.     private String clientId;
11.     private String username;
12.     private String password;
13.     private Integer timeout;
14.     private Integer keepAlive;
15.     private MyHandle myHandle;
17.     public  MyMqttClient(){
18.         System.out.println("MyMqttClient空构造函数");
19.     }
21.     public MyMqttClient(String host, String username, String password, String clientId, Integer timeOut, Integer keepAlive,MyHandle myHandle) {
22.         System.out.println("MyMqttClient全参构造");
23.         this.host = host;
24.         this.username = username;
25.         this.password = password;
26.         this.clientId = clientId;
27.         this.timeout = timeOut;
28.         this.keepAlive = keepAlive;
29.         this.myHandle = myHandle;
30.     }
32.     public static MqttClient getClient() {
33.         return client;
34.     }
36.     public static void setClient(MqttClient client) {
37.         MyMqttClient.client = client;
38.     }
40.     /**
41.      * 设置mqtt连接参数
42.      */
43.      public MqttConnectOptions setMqttConnectOptions(String username,String password,Integer timeout, Integer keepAlive){
44.          MqttConnectOptions options = new MqttConnectOptions();
45.          options.setUserName(username);
46.          options.setPassword(password.toCharArray());
47.          options.setConnectionTimeout(timeout);
48.          options.setKeepAliveInterval(keepAlive);
49.          options.setCleanSession(true);
50.          options.setAutomaticReconnect(true);
51.          return options;
52.      }
54.     /**
55.      * 连接mqtt服务端
56.      */
57.     public void connect() throws MqttException {
58.         if(client == null){
59.             client = new MqttClient(host,clientId,new MemoryPersistence());
60.             client.setCallback(new MyMqttCallback(MyMqttClient.this,this.myHandle));
61.         }
62.         MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepAlive);
63.         if(!client.isConnected()){
64.             client.connect(mqttConnectOptions);
65.         }else{
66.             client.disconnect();
67.             client.connect(mqttConnectOptions);
68.         }
69.         log.info("MQTT connect success");
70.     }
72.     /**
73.      * 断开连接
74.      * @throws MqttException
75.      */
76.     public void disconnect()throws MqttException{
77.         if(null!=client && client.isConnected()){
78.             client.disconnect();;
79.         }
80.     }
81.     /**
82.      * 发布,qos默认为0，非持久化
83.      */
84.      public void publish(String pushMessage,String topic,int qos){
85.          publish(pushMessage, topic, qos, false);
86.      }
88.     /**
89.      * 发布消息
90.      *
91.      * @param pushMessage
92.      * @param topic
93.      * @param qos
94.      * @param retained:留存
95.      */
96.     public void publish(String pushMessage, String topic, int qos, boolean retained) {
97.         MqttMessage mqttMessage = new MqttMessage();
98.         mqttMessage.setPayload(pushMessage.getBytes());
99.         mqttMessage.setQos(qos);
100.         mqttMessage.setRetained(retained);
101.         MqttTopic mqttTopic = MyMqttClient.getClient().getTopic(topic);
102.         if(ObjectUtils.isEmpty(mqttTopic)){
103.             log.error("主题不存在");
104.         }
105.         synchronized (this){
106.             try{
107.                 MqttDeliveryToken mqttDeliveryToken = mqttTopic.publish(mqttMessage);
108.                 mqttDeliveryToken.waitForCompletion(1000L);
109.             }catch (MqttException e){
110.                 e.printStackTrace();
111.             }
112.         }
113.     }
115.     /**
116.      * 订阅
117.      *
118.      * @param topic
119.      * @param qos
120.      */
121.     public void subscribe(String topic, int qos) {
122.         try {
123.             MyMqttClient.getClient().subscribe(topic, qos);
124.             log.info("订阅主题:"+topic+"成功！");
125.         } catch (MqttException e) {
126.             log.error("订阅主题:"+topic+"失败！",e);
127.         }
128.     }
129.     /**
130.      * 取消订阅
131.      */
132.     public void cleanTopic(String topic){
133.         if(!ObjectUtils.isEmpty(client) && client.isConnected()){
134.             try{
135.                 client.unsubscribe(topic);
136.             }catch (MqttException e){
137.                 log.error("取消订阅失败！"+e);
138.             }
139.         }else{
140.             log.info("主题不存在或未连接！");
141.         }
142.     }
143. }

复制代码

• 回调类（消息发送和接收时响应）
  

1. @Slf4j
2. public class MyMqttCallback implements MqttCallbackExtended {
3.     private MyMqttClient myMqttClient;
4.     private MyHandle myHandle;
5.     public MyMqttCallback(MyMqttClient myMqttClient,MyHandle myHandle) {
6.         this.myMqttClient = myMqttClient;
7.         this.myHandle = myHandle;
8.     }
10.     /**
11.      * 连接完成
12.      * @param reconnect
13.      * @param serverURI
14.      */
15.     @Override
16.     public void connectComplete(boolean reconnect,String serverURI) {
17.         log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");
18.         //订阅主题(可以在这里订阅主题)
19.         try {
20.             MyMqttClient.getClient().subscribe("topic1");
21.         } catch (MqttException e) {
22.             log.error("主题订阅失败");
23.         }
24.     }
26.     /**
27.      * 连接丢失 进行重连操作
28.      * @param throwable
29.      */
30.     @Override
31.     public void connectionLost(Throwable throwable) {
32.         log.warn("mqtt connectionLost >>> 5S之后尝试重连: {}", throwable.getMessage());
33.         long reconnectTimes = 1;
34.         while (true){
35.             try{
36.                 Thread.sleep(5000);
37.             }catch (InterruptedException ignored){}
38.             try{
39.                 if(MyMqttClient.getClient().isConnected()){ // 已连接
40.                     return;
41.                 }
42.                 reconnectTimes+=1;
43.                 log.warn("mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
44.                 MyMqttClient.getClient().reconnect();
45.             }catch (MqttException e){
46.                 log.error("mqtt断链异常",e);
47.             }
48.         }
49.     }
51.     /**
52.      * 订阅者收到消息之后执行
53.      * @param topic
54.      * @param mqttMessage
55.      * @throws Exception
56.      */
57.     @Override
58.     public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
59.         log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
60.         myHandle.handle(topic,mqttMessage);
61.     }
63.     /**
64.      * * 消息到达后
65.      * subscribe后，执行的回调函数
66.      * publish后，配送完成后回调的方法
67.      *
68.      * @param iMqttDeliveryToken
69.      */
70.     @Override
71.     public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
72.         System.out.println("接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用");
73.         log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
74.     }
75. }

复制代码

1. import lombok.extern.slf4j.Slf4j;
2. import org.eclipse.paho.client.mqttv3.MqttMessage;
3. import org.springframework.scheduling.annotation.Async;
4. import org.springframework.stereotype.Service;
6. @Service
7. @Slf4j
8. public class MyHandle {
9.     @Async
10.     public void handle(String topic, MqttMessage message) {
11.         log.info("处理消息主题:" + topic + " 信息:" + message);
12.     }
13. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！