MQTT从入门到醒目之 MQTT 客户端编程

MQTT 客户端编程

1 在VUE中利用MQTT

详细步骤如下所示：
1、初始化vue项目

1. // 创建一个使用vite构建的前端项目
2. npm create vite@latest
4. // 进入到项目中，执行如下命令安装项目依赖
5. npm install

复制代码

2、安装element plus

1. // 安装element plus
2. npm install element-plus --save
4. // 安装mqtt.js依赖
5. npm install mqtt --save
7. // 在main.js中添加如下代码
8. import { createApp } from 'vue'
9. import App from './App.vue'
10. import ElementPlus from 'element-plus'
11. import 'element-plus/dist/index.css'
13. const app = createApp(App)
14. app.use(ElementPlus)
15. app.mount('#app')

复制代码

3、导入课程资料中的MqttDemo.vue页面到components文件夹下
4、更改App.vue页面代码如下所示

1. <script setup>
2. import MqttDemo from "./components/MqttDemo.vue";
3. </script>
5. <template>
6.   <MqttDemo/>
7. </template>
9. <style>
10. </style>

复制代码

5、创建和关闭毗连

1. <script setup>
2. import mqtt from "mqtt";
4. // 定义链接信息对象
5. const connectionInfo = ref({
6.   protocol: 'ws',
7.   host: "192.168.136.147",
8.   port: 8083,
9.   clientId: "emqx_vue3_" + Math.random().toString(16).substring(2, 8),
10.   username: "zhangsan",
11.   password: "123",
12.   clean: true,
13.   connectTimeout: 10 * 1000, // ms
14.   reconnectPeriod: 4000, // ms
15. })
17. // 创建链接对象
18. const client = ref({})
19. const clientInitData = ref({      // 链接初始化相关数据
20.   connnected: false
21. })
23. // 建立连接事件处理函数
24. const createConnection = () => {
26.   const { protocol, host, port , ...options } = connectionInfo.value;
27.   const connectUrl = `${protocol}://${host}:${port}/mqtt`;
28.   console.log(connectUrl)
29.   client.value = mqtt.connect(connectUrl , options);   // 建立连接
30.   clientInitData.value.connnected = true ;
31.   console.info("createConnection successful...")
33. }
35. // 端口链接事件处理函数
36. const closeConnection = () => {
38.   // 关闭链接
39.   client.value.end(false , () => {   // 如果设置为true，将会立即关闭套接字，并且不发送MQTT DISCONNECT包。如果设置为false（默认值），则会发送MQTT DISCONNECT包给代理，然后关闭套接字。
40.     clientInitData.value.connnected = false;
41.     console.info("closeConnection successful...")
42.   })
44. }
45. </script>
47. <template>
48.   <div class="mqtt-demo">
49.     <el-card>
50.       <h1>配置信息</h1>
51.       <el-form label-position="top" >
52.         <el-row :gutter="20">
54.           <el-col :span="8">
55.             <el-form-item prop="protocol" label="选择协议">
56.               <el-select v-model="connectionInfo.protocol">
57.                 <el-option label="ws://" value="ws"></el-option>
58.                 <el-option label="wss://" value="wss"></el-option>
59.               </el-select>
60.             </el-form-item>
61.           </el-col>
63.           <el-col :span="8">
64.             <el-form-item prop="host" label="主机地址">
65.               <el-input v-model="connectionInfo.host" ></el-input>
66.             </el-form-item>
67.           </el-col>
69.           <el-col :span="8">
70.             <el-form-item prop="port" label="端口号">
71.               <el-input type="number" v-model="connectionInfo.port" placeholder="8083/8084"></el-input>
72.             </el-form-item>
73.           </el-col>
75.           <el-col :span="8">
76.             <el-form-item prop="clientId" label="客户端ID">
77.               <el-input v-model="connectionInfo.clientId"> </el-input>
78.             </el-form-item>
79.           </el-col>
81.           <el-col :span="8">
82.             <el-form-item prop="username" label="用户名">
83.               <el-input v-model="connectionInfo.username"></el-input>
84.             </el-form-item>
85.           </el-col>
87.           <el-col :span="8">
88.             <el-form-item prop="password" label="密码">
89.               <el-input v-model="connectionInfo.password"></el-input>
90.             </el-form-item>
91.           </el-col>
93.           <el-col :span="24">
94.             <el-button type="primary" :disabled="clientInitData.connnected" @click="createConnection">建立连接</el-button>
95.             <el-button type="danger" :disabled="!clientInitData.connnected" @click="closeConnection">断开连接</el-button>
96.           </el-col>
98.         </el-row>
99.       </el-form>
100.     </el-card>
101.    </div>
102. </template>

复制代码

6、订阅和取消订阅

1. <script setup>
2.    
3. // 消息质量取值数组
4. const qosList = [0, 1, 2];
5. const receivedMessages = ref(null)
6. const subscriptionInfo = ref({     // 订阅参数数据模型
7.   topic: '' ,
8.   qos: 0
9. })
10. const subscriptionInitData = ref({        // 订阅初始化数据
11.   subscription: false
12. })
14. // 定义定义主题的事件处理函数
15. const subscriptionTopicHandler = () => {
17.   const { topic, qos } = subscriptionInfo.value
18.   console.info(qos)
19.   client.value.subscribe(topic, { qos } , (error , res) => {
20.     if(error) {
21.       console.info("subscriptionTopic Error:", error);
22.       return ;
23.     }
24.     subscriptionInitData.value.subscription = true ;
25.     console.info("subscriptionTopic successful.... ");
27.     // 订阅成功以后，监听发送消息事件
28.     client.value.on('message' , (topic , message) => {
29.       console.info("topic -----> " + topic + ", message -----> " + message)
30.       receivedMessages.value = "topic -----> " + topic + ", message -----> " + message ;
31.     })
33.   })
35. }
37. // 定义取消订阅的事件处理函数
38. const unSubscriptionTopicHandler = () => {
39.   const {topic, qos } = subscriptionInfo.value
40.   client.value.unsubscribe(topic, { qos } , (error , res) => {
41.     if(error) {
42.       console.info("unSubscriptionTopic Error:", error);
43.       return ;
44.     }
45.     subscriptionInitData.value.subscription = false ;
46.     console.info("unSubscriptionTopic successful.... ");
48.   })
49. }
51. </script>
53. <template>
55.             <el-card>
56.           <h1>订阅主题</h1>
57.           <el-form label-position="top" >
58.             <el-row :gutter="20">
59.               <el-col :span="8">
60.                 <el-form-item prop="topic" label="Topic">
61.                   <el-input v-model="subscriptionInfo.topic"></el-input>
62.                 </el-form-item>
63.               </el-col>
64.               <el-col :span="8">
65.                 <el-form-item prop="qos" label="QoS">
66.                   <el-select v-model="subscriptionInfo.qos">
67.                     <el-option
68.                         v-for="qos in qosList"
69.                         :key="qos"
70.                         :label="qos"
71.                         :value="qos"
72.                     ></el-option>
73.                   </el-select>
74.                 </el-form-item>
75.               </el-col>
76.               <el-col :span="8">
77.                 <el-button  type="primary" class="sub-btn" :disabled="subscriptionInitData.subscription" @click="subscriptionTopicHandler">订阅主题</el-button>
78.                 <el-button type="primary" class="sub-btn" :disabled="!subscriptionInitData.subscription" @click="unSubscriptionTopicHandler">取消订阅</el-button>
79.               </el-col>
80.             </el-row>
81.           </el-form>
82.         </el-card>
84. </template>

复制代码

7、发布消息

1. <script setup>
2.    
3. // 发布消息参数
4. const publishInfo = ref({
5.   topic: '' ,
6.   qos: 0,
7.   payLoad: ''
8. })
9.    
10. // 定义发布消息的事件处理函数
11. const doPublish = () => {
12.   const {topic , payLoad , qos } = publishInfo.value ;
13.   client.value.publish(topic , payLoad , { qos } , (error , res) => {
14.     if(error) {
15.       console.info("publish msg info error...." , error)
16.       return ;
17.     }
18.     console.info("publish msg info successful....")
19.   }) ;
21. }
23. </script>
24. <template>
26.         <el-card>
27.       <h1>发布消息</h1>
28.       <el-form label-position="top" >
29.         <el-row :gutter="20">
30.           <el-col :span="8">
31.             <el-form-item prop="topic" label="Topic">
32.               <el-input v-model="publishInfo.topic"></el-input>
33.             </el-form-item>
34.           </el-col>
35.           <el-col :span="8">
36.             <el-form-item prop="payload" label="Payload">
37.               <el-input v-model="publishInfo.payLoad"></el-input>
38.             </el-form-item>
39.           </el-col>
40.           <el-col :span="8">
41.             <el-form-item prop="qos" label="QoS">
42.               <el-select v-model="publishInfo.qos">
43.                 <el-option
44.                     v-for="qos in qosList"
45.                     :key="qos"
46.                     :label="qos"
47.                     :value="qos"
48.                 ></el-option>
49.               </el-select>
50.             </el-form-item>
51.           </el-col>
52.         </el-row>
53.       </el-form>
54.       <el-col :span="24" class="text-right">
55.         <el-button type="primary" @click="doPublish">发布消息</el-button>
56.       </el-col>
57.     </el-card>
59. </template>

复制代码

2 在Java中利用MQTT

2.1 Eclipse Paho Java Client

详细步骤：
1、创建一个Spring Boot项目，添加如下依赖

1. <parent>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-parent</artifactId>
4.     <version>3.0.5</version>
5. </parent>
7. <dependencies>
9.     <!-- spring boot整合junit单元测试的起步依赖 -->
10.     <dependency>
11.         <groupId>org.springframework.boot</groupId>
12.         <artifactId>spring-boot-starter-test</artifactId>
13.     </dependency>
15.     <!-- mqtt java客户端依赖 -->
16.     <dependency>
17.         <groupId>org.eclipse.paho</groupId>
18.         <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
19.         <version>1.2.5</version>
20.     </dependency>
22. </dependencies>

复制代码

2、创建毗连代码实现

1. @Test
2. public void createConnection() throws MqttException {
4.     // 定义链接相关参数
5.     String broker = "tcp://192.168.136.147:1883";
6.     String username = "zhangsan";
7.     String password = "123";
8.     String clientid = "mqtt_java_client_01";
10.     // 创建MqttJava客户端对象
11.     // MqttClientPersistence: 代表一个持久的数据存储，用于在传输过程中存储出站和入站的信息
12.     MqttClient client = new MqttClient(broker, clientid , new MemoryPersistence());   
13.     MqttConnectOptions options = new MqttConnectOptions();
14.     options.setUserName(username);
15.     options.setPassword(password.toCharArray());
16.     client.connect(options);
18.     // 阻塞当前线程
19.     while (true) ;
20. }

复制代码

3、发布消息代码演示

1. @Test
2. public void sendMessage() throws MqttException {
4.     // 定义链接相关参数
5.     String broker = "tcp://192.168.136.147:1883";
6.     String username = "zhangsan";
7.     String password = "123";
8.     String clientid = "mqtt_java_client_01";
10.     // 创建MqttJava客户端对象
11.     MqttClient client = new MqttClient(broker, clientid , new MemoryPersistence());
12.     MqttConnectOptions options = new MqttConnectOptions();
13.     options.setUserName(username);
14.     options.setPassword(password.toCharArray());
15.     client.connect(options);
17.     // 创建消息对象QoS
18.     String content = "hello mqtt";
19.     MqttMessage message = new MqttMessage(content.getBytes());
20.     message.setQos(2);
21.     message.setRetained(true);
23.     // 发送消息
24.     client.publish("a/c" , message);
26.     // 关闭链接释放资源
27.     client.disconnect();
28.     client.close();
30. }

复制代码

4、订阅主题获取消息

1. @Test
2. public void receiveMessage() throws MqttException {
4.     // 定义链接相关参数
5.     String broker = "tcp://192.168.136.147:1883";
6.     String username = "zhangsan";
7.     String password = "123";
8.     String clientid = "mqtt_java_client_02";
10.     // 创建MqttJava客户端对象
11.     MqttClient client = new MqttClient(broker, clientid , new MemoryPersistence());
12.     MqttConnectOptions options = new MqttConnectOptions();
13.     options.setUserName(username);
14.     options.setPassword(password.toCharArray());
16.     // 添加回调函数获取主题消息
17.     client.setCallback(new MqttCallback() {
18.         
19.         @Override
20.         public void connectionLost(Throwable cause) {  // 连接丢失时被调用
21.             System.out.println("connectionLost: " + cause.getMessage());
22.         }
24.         @Override
25.         public void messageArrived(String topic, MqttMessage message) throws Exception {  // 接收到消息时被调用
26.             System.out.println("topic: " + topic);
27.             System.out.println("Qos: " + message.getQos());
28.             System.out.println("message content: " + new String(message.getPayload()));
29.         }
31.         @Override
32.         public void deliveryComplete(IMqttDeliveryToken token) {  // 消息接收完成时被调用
33.             System.out.println("deliveryComplete---------" + token.isComplete());
34.         }
35.         
36.     });
38.     // 订阅主题
39.     client.connect(options);
40.     client.subscribe("a/d" , 2);
42.     while(true) ;
44. }

复制代码

2.2 spring-integration-mqtt

2.2.1 底子环境搭建

1、创建一个Spring Boot项目，并加入如下依赖：

1. <dependencies>
2.        
3.     <!-- spring boot项目web开发的起步依赖 -->
4.     <dependency>
5.         <groupId>org.springframework.boot</groupId>
6.         <artifactId>spring-boot-starter-web</artifactId>
7.     </dependency>
9.         <!-- spring boot项目集成消息中间件基础依赖 -->
10.     <dependency>
11.         <groupId>org.springframework.boot</groupId>
12.         <artifactId>spring-boot-starter-integration</artifactId>
13.     </dependency>
15.         <!-- spring boot项目和mqtt客户端集成起步依赖 -->
16.     <dependency>
17.         <groupId>org.springframework.integration</groupId>
18.         <artifactId>spring-integration-mqtt</artifactId>
19.         <version>5.4.3</version>
20.     </dependency>
22.     <!-- lombok依赖 -->
23.     <dependency>
24.         <groupId>org.projectlombok</groupId>
25.         <artifactId>lombok</artifactId>
26.     </dependency>
28.     <!-- fastjson依赖 -->
29.     <dependency>
30.         <groupId>com.alibaba</groupId>
31.         <artifactId>fastjson</artifactId>
32.         <version>1.2.83</version>
33.     </dependency>
35. </dependencies>

复制代码

2、编写启动类

1. @EnableConfigurationProperties(value = MqttConfigurationProperties.class)
2. @SpringBootApplication
3. public class MqttDemoApplication {
5.     public static void main(String[] args) {
6.         SpringApplication.run(MqttDemoApplication.class , args) ;
7.     }
9. }

复制代码

3、在application.yml文件中添加如下配置

1. spring:
2.   mqtt:
3.     username: zhangsan
4.     password: 123
5.     url: tcp://192.168.136.147:1883
6.     subClientId: sub_client_id_123
7.     subTopic: liujh/iot/lamp/line
8.     pubClientId: pub_client_id_123

复制代码

4、创建实体类读取自定义配置

1. @Data
2. @ConfigurationProperties(prefix = "spring.mqtt")
3. public class MqttConfigurationProperties {
5.     private String username;
6.     private String password;
7.     private String url;
8.     private String subClientId ;
9.     private String subTopic ;
10.     private String pubClientId ;
12. }

复制代码

5、创建配置类配置链接工厂

1. @Configuration
2. public class MqttConfiguration {
4.     @Autowired
5.     private MqttConfigurationProperties mqttConfigurationProperties ;
7.     @Bean
8.     public MqttPahoClientFactory mqttClientFactory(){
10.         // 创建客户端工厂
11.         DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
13.         // 创建MqttConnectOptions对象
14.         MqttConnectOptions options = new MqttConnectOptions();
15.         options.setCleanSession(true);
16.         options.setUserName(mqttConfigurationProperties.getUsername());
17.         options.setPassword(mqttConfigurationProperties.getPassword().toCharArray());
18.         options.setServerURIs(new String[]{mqttConfigurationProperties.getUrl()});
19.         factory.setConnectionOptions(options);
21.         // 返回
22.         return factory;
23.     }
25. }

复制代码

2.2.2 订阅主题获取消息

详细步骤：
1、配置入站适配器

1. @Configuration
2. public class MqttInboundConfiguration {
4.     @Autowired
5.     private MqttConfigurationProperties mqttConfigurationProperties ;
7.     @Autowired
8.     private ReceiverMessageHandler receiverMessageHandler;
10.     /**
11.      * 配置消息传输通道
12.      * @return
13.      */
14.     @Bean
15.     public MessageChannel mqttInputChannel() {
16.         return new DirectChannel();
17.     }
19.     /**
20.      * 配置入站适配器
21.      */
22.     @Bean
23.     public MessageProducer messageProducer(MqttPahoClientFactory mqttPahoClientFactory) {
24.         MqttPahoMessageDrivenChannelAdapter adapter  =
25.                 new MqttPahoMessageDrivenChannelAdapter(mqttConfigurationProperties.getUrl() ,
26.                         mqttConfigurationProperties.getSubClientId() ,
27.                         mqttPahoClientFactory , mqttConfigurationProperties.getSubTopic().split(",")) ;
28.         adapter.setConverter(new DefaultPahoMessageConverter());
29.         adapter.setQos(1);
30.         adapter.setOutputChannel(mqttInputChannel());
31.         return adapter ;
32.     }
34.     /**
35.      * 配置入站消息处理器
36.      * @return
37.      */
38.     @Bean
39.     @ServiceActivator(inputChannel = "mqttInputChannel")
40.     public MessageHandler messageHandler() {
41.         return this.receiverMessageHandler ;
42.     }
44. }

复制代码

2、定义监听主题消息的处理器

1. @Component
2. public class ReceiverMessageHandler implements MessageHandler {
4.     @Override
5.     public void handleMessage(Message<?> message) throws MessagingException {
6.         MessageHeaders headers = message.getHeaders();
7.         String receivedTopicName = (String) headers.get("mqtt_receivedTopic");
8.         if("liujh/iot/lamp/line".equals(receivedTopicName)) {
9.             System.out.println("接收到消息：" + message.getPayload());
10.         }
11.     }
13. }

复制代码

测试：通过MQTTX向liujh/iot/lamp/line主题发送消息
2.2.3 向指定主题发送消息

详细步骤：
1、配置出站消息处理器

1. @Configuration
2. public class MqttOutboundConfiguration {
4.     @Autowired
5.     private MqttConfigurationProperties mqttConfigurationProperties ;
7.     @Autowired
8.     private MqttPahoClientFactory pahoClientFactory ;
10.     @Bean
11.     public MessageChannel mqttOutputChannel() {
12.         return new DirectChannel();
13.     }
15.     @Bean
16.     @ServiceActivator(inputChannel = "mqttOutputChannel")
17.     public MessageHandler mqttOutboundMassageHandler() {
18.         MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfigurationProperties.getUrl() ,
19.                 mqttConfigurationProperties.getPubClientId() , pahoClientFactory ) ;
20.         messageHandler.setAsync(true);
21.         messageHandler.setDefaultQos(0);
22.         messageHandler.setDefaultTopic("default");
23.         return messageHandler ;
24.     }
26. }

复制代码

2、定义发送消息的网关接口

1. @MessagingGateway(defaultRequestChannel = "mqttOutputChannel")
2. public interface MqttGateway {
4.     /**
5.      * 发送mqtt消息
6.      * @param topic 主题
7.      * @param payload 内容
8.      */
9.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
11.     /**
12.      * 发送包含qos的消息
13.      * @param topic 主题
14.      * @param qos 对消息处理的几种机制。
15.      *          * 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。<br>
16.      *          * 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。<br>
17.      *          * 2 多了一次去重的动作，确保订阅者收到的消息有一次。
18.      * @param payload 消息体
19.      */
20.     void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
21.    
22. }

复制代码

3、定义发送消息的服务类

1. @Component
2. @AllArgsConstructor
3. public class MqttMessageSender {
5.     private MqttGateway mqttGateway;
7.     /**
8.      * 发送mqtt消息
9.      * @param topic 主题
10.      * @param message 内容
11.      */
12.     public void send(String topic, String message) {
13.         mqttGateway.sendToMqtt(topic, message);
14.     }
16.     /**
17.      * 发送包含qos的消息
18.      * @param topic 主题
19.      * @param qos 质量
20.      * @param message 消息体
21.      */
22.     public void send(String topic, int qos, byte[] message){
23.         mqttGateway.sendToMqtt(topic, qos, message);
24.     }
25. }

复制代码

3 智能灯泡案例

需求：
1、智能灯泡装备上线以后向MQTT服务端发送消息，后端服务从MQTT中获取消息记录装备信息到数据库中
2、后端微服务向MQTT服务端发送开灯或者关灯消息，装备端从MQTT中获取消息控制灯泡的开和关
3、装备端对灯泡举行开和关操作的时候向MQTT中发送消息，后端服务获取MQTT消息记录灯泡的开关状态
3.1 服务端获取装备上线消息

3.1.1 环境准备

详细步骤：
1、创建对应的数据库表

1. -- 智能灯泡设备表
2. CREATE TABLE `tb_lamp` (
3.   `id` bigint NOT NULL AUTO_INCREMENT,
4.   `deviceId` varchar(50) DEFAULT NULL,
5.   `status` int DEFAULT NULL COMMENT '1：上线  0：下线',
6.   `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ,
7.   `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
8.   PRIMARY KEY (`id`)
9. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
11. -- 智能灯泡设备状态表
12. CREATE TABLE `tb_lamp_status` (
13.   `id` int NOT NULL AUTO_INCREMENT,
14.   `deviceId` varchar(50) DEFAULT NULL,
15.   `status` int DEFAULT NULL COMMENT '0: 关灯   1：开灯',
16.   `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
17.   PRIMARY KEY (`id`)
18. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

复制代码

2、在spring-integration-mqtt案例中加入如下依赖

1. <dependency>
2.     <groupId>com.baomidou</groupId>
3.     <artifactId>mybatis-plus-boot-starter</artifactId>
4.     <version>3.5.3.1</version>
5. </dependency>
7. <dependency>
8.     <groupId>mysql</groupId>
9.     <artifactId>mysql-connector-java</artifactId>
10.     <version>8.0.32</version>
11. </dependency>

复制代码

3、在application.yml文件中加入如下依赖

1. spring:
2.   datasource:
3.     driver-class-name: com.mysql.cj.jdbc.Driver
4.     url: jdbc:mysql://192.168.136.147:3306/lamp_test?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
5.     username: root
6.     password: 1234
8. mybatis-plus:
9.   configuration:
10.     log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
11.     map-underscore-to-camel-case: true
12.   mapper-locations: classpath*:mapper/*Mapper.xml

复制代码

4、通过mybatis的逆向工程天生tb_lamp和tb_lamp_status表对应的底子代码
5、在启动类上添加@MapperScan注解指定Mapper接口的包路径
3.1.2 接口分析

   接口一：装备上线
  当终端装备毗连上EMQX以后，发奉上线消息到EMQX服务端，分析如下：

1. 主题: liujh/iot/lamp/line
2. 消息内容：
3.     {
4.         "deviceId": "xxxxxx",
5.         "online": 1
6.     }
7. 数据说明：
8.         deviceId: 设备id
9.         online:   上线状态，1表示上线，0表示离线

复制代码

3.1.3 业务代码

对ReceiverMessageHandler类的代码举行如下改造：

1. @Component
2. public class ReceiverMessageHandler implements MessageHandler {
4.     @Autowired
5.     private TbLampService tbLampService ;
7.     @Override
8.     public void handleMessage(Message<?> message) throws MessagingException {
9.         MessageHeaders headers = message.getHeaders();
10.         String receivedTopicName = (String) headers.get("mqtt_receivedTopic");
11.         if("liujh/iot/lamp/line".equals(receivedTopicName)) {
12.             tbLampService.updateLampOnlineStatus(message.getPayload().toString()) ;        // 更新智能灯泡的上线状态
13.         }
14.     }
16. }

复制代码

对TbLampServiceImpl类的代码举行如下改造：

1. @Service
2. public class TbLampServiceImpl extends ServiceImpl<TbLampMapper, TbLamp> implements TbLampService {
4.     @Override
5.     public void updateLampOnlineStatus(String jsonInfo) {
7.         // 解析消息获取设备id和上线状态
8.         Map<String ,  Object> map = JSON.parseObject(jsonInfo, Map.class);
9.         String deviceId = map.get("deviceId").toString();
10.         Integer status = Integer.parseInt(map.get("online").toString());
12.         // 根据设备的id查询设备数据
13.         LambdaQueryWrapper<TbLamp> lambdaQueryWrapper = new LambdaQueryWrapper<>() ;
14.         lambdaQueryWrapper.eq(TbLamp::getDeviceid , deviceId) ;
15.         TbLamp tbLamp = this.getOne(lambdaQueryWrapper);
16.         if(tbLamp == null) {        // 设备不存在，新增设备
17.            tbLamp = new TbLamp() ;
18.            tbLamp.setDeviceid(deviceId);
19.            tbLamp.setStatus(status);
20.            this.save(tbLamp) ;
21.         }else {     // 设备已经存在，修改设备的状态
22.             tbLamp.setStatus(status);
23.             tbLamp.setUpdateTime(new Date());
24.             this.updateById(tbLamp) ;
25.         }
26.     }
28. }

复制代码

3.2 服务端发送关灯开灯消息到MQTT

3.2.1 接口分析

   接口三：后端发送消息控制智能灯泡开关
  后端可以发送控制灯泡状态消息到EMQX中，装备端监听指定主题获取消息，控制灯泡的开关状态，分析如下：

1. 主题: liujh/iot/lamp/server/status
2. 消息内容：
3.         {
4.                 "deviceId": "xxxxxx",
5.                 "status": 0
6.         }
7. 数据说明：               
8.         status：        0：关灯   ， 1：开灯

复制代码

3.2.2 业务代码

在spring-integration-mqtt案例中添加如下controller接口方法

1. @RestController
2. @RequestMapping(value = "/api/lamp")
3. public class LampApiController {
5.     @Autowired
6.     private MqttMessageSender mqttMessageSender;
8.     @GetMapping(value = "/{deviceId}/{status}")
9.     public String sendStatusLampMsg(@PathVariable(value = "deviceId") String deviceId , @PathVariable(value = "status") Integer status) {
10.         Map<String , Object> map = new HashMap<>() ;
11.         map.put("deviceId" , deviceId) ;
12.         map.put("status" , status) ;
13.         String json = JSON.toJSONString(map);
14.         mqttMessageSender.send("liujh/iot/lamp/server/status" , json);
15.         return "ok" ;
16.     }
18. }

复制代码

3.3 服务端获取装备开灯关灯消息

3.3.1 接口分析

   接口四：装备端改变智能灯泡开关的状态，状态发给给后端，后端记录状态

1. 主题：liujh/iot/lamp/device/status
2. 消息内容：
3.         {
4.                 "deviceId": "xxxxx"  
5.                 "status": 0
6.         }
7. 数据说明：       
8.         deviceId：设备id
9.         status：0：关灯   ， 1：开灯

复制代码

3.3.2 业务代码

对ReceiverMessageHandler类的代码举行如下改造：

1. // com.liujh.mqtt.receiver.ReceiverMessageHandler
2. @Override
3. public void handleMessage(Message<?> message) throws MessagingException {
4.     MessageHeaders headers = message.getHeaders();
5.     String receivedTopicName = (String) headers.get("mqtt_receivedTopic");
6.     if("liujh/iot/lamp/line".equals(receivedTopicName)) {
7.         tbLampService.updateLampOnlineStatus(message.getPayload().toString()) ;        // 更新智能灯泡的上线状态
8.     }else if("liujh/iot/lamp/device/status".equals(receivedTopicName)) {
9.         tbLampStatusService.saveDeviceStatus(message.getPayload().toString()) ;
10.     }
11. }

复制代码

对TbLampStatusServiceImpl类的代码举行如下改造：

1. @Service
2. public class TbLampStatusServiceImpl extends ServiceImpl<TbLampStatusMapper, TbLampStatus> implements TbLampStatusService {
4.     @Override
5.     public void saveDeviceStatus(String json) {
7.         // 获取消息内容
8.         Map<String , Object> map = JSON.parseObject(json, Map.class);
9.         String deviceId = map.get("deviceId").toString();
10.         Integer status = Integer.parseInt(map.get("status").toString());
12.         // 创建对象封装消息
13.         TbLampStatus tbLampStatus = new TbLampStatus() ;
14.         tbLampStatus.setDeviceid(deviceId);
15.         tbLampStatus.setStatus(status);
16.         this.save(tbLampStatus) ;
18.     }
20. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。