• 【EMQX】 Spring Cloud 集成MQTT并异步入库（mongodb）

    1. 需求：

            1. Spring boot项目，集成mqtt，订阅物联网设备数据上报topic1

            2. 处理topic1，并异步入库到mongodb

            3.订阅MQTT客户端上下线主题

            4. mongodb 根据设备id分集合

    2. 代码实现

    准备工作：需要在EMQX 5.0允许订阅：$SYS/brokers/+/clients/# （客户端上下线主题）

    第一步：打开授权文件

    第二步：添加允许后，更新即可

    {allow, all, subscribe, ["$SYS/brokers/+/clients/#"]}.

    2.1 引入pom

    • 子项目（本项目）
    1.         
    2.         <dependency>
    3.             <groupId>org.eclipse.pahogroupId>
    4.             <artifactId>org.eclipse.paho.client.mqttv3artifactId>
    5.             <version>1.2.2version>
    6.         dependency>
    7.         
    8.         <dependency>
    9.             <groupId>org.springframework.bootgroupId>
    10.             <artifactId>spring-boot-starter-data-mongodbartifactId>
    11.         dependency>
    12.         <dependency>
    13.             <groupId>com.spring4allgroupId>
    14.             <artifactId>mongodb-plus-spring-boot-starterartifactId>
    15.         dependency>
    • 父-引入pom
    1.     <parent>
    2.         <groupId>org.springframework.bootgroupId>
    3.         <artifactId>spring-boot-starter-parentartifactId>
    4.         <version>2.3.5.RELEASEversion>
    5.         <relativePath/>
    6.     parent>
    7.     <dependencyManagement>
    8.         <dependencies>
    9.             
    10.             <dependency>
    11.                 <groupId>org.springframework.cloudgroupId>
    12.                 <artifactId>spring-cloud-dependenciesartifactId>
    13.                 <version>Hoxton.SR12version>
    14.                 <type>pomtype>
    15.                 <scope>importscope>
    16.             dependency>
    17.         dependencies>

    2.2 启动类引入注解 @EnableMongoPlus

    2.3  application.yml mongodb配置

    1. spring:
    2.   profiles:
    3.     active: prod
    4.   data:
    5.     mongodb:
    6.       uri: mongodb://root:${mongo-password}@${host}:${mongodbport}/hzd
    7.       option:
    8.         # 最小连接数
    9.         min-connection-per-host: 10
    10.         # 最大连接数
    11.         max-connection-per-host: 200
    12.         #允许阻塞连接乘数的线程
    13.         threads-allowed-to-block-for-connection-multiplier: 5
    14.         #服务器超时
    15.         server-selection-timeout: 30000
    16.         #最长等待时间
    17.         max-wait-time: 120000
    18.         #最大连接空闲时间
    19.         max-connection-idle-time: 0
    20.         #最大连接生存时间
    21.         max-connection-life-time: 0
    22.         #连接超时
    23.         connect-timeout: 10000
    24.         #套接字超时10000
    25.         socket-timeout: 0
    26.         socket-keep-alive: false
    27.         ssl-enabled: false
    28.         ssl-invalid-host-name-allowed: false
    29.         always-use-m-beans: false
    30.         heartbeat-socket-timeout: 20000
    31.         heartbeat-connect-timeout: 20000
    32.         min-heartbeat-frequency: 500
    33.         heartbeat-frequency: 10000
    34.         local-threshold: 15

    2.4 代码

    1. EmqClient.java                  连接emqx

    • init()  使用@PostConstruct注解，开机自动连接EMQX + 订阅
    • reConnect()  断线重连后，订阅主题
    1. package com.hzd.mqtt.client;
    2.  
    3.  
    4. import com.hzd.mqtt.enums.QosEnum;
    5. import com.hzd.mqtt.properties.MqttProperties;
    6. import lombok.extern.slf4j.Slf4j;
    7. import org.eclipse.paho.client.mqttv3.*;
    8. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    9. import org.springframework.beans.factory.annotation.Autowired;
    10. import org.springframework.stereotype.Component;
    11.  
    12. import javax.annotation.PostConstruct;
    13. import javax.annotation.PreDestroy;
    14.  
    15. /**
    16.  * Created by
    17.  */
    18. @Component
    19. @Slf4j
    20. public class EmqClient {
    21.  
    22.     private IMqttClient client;
    23.  
    24.     @Autowired
    25.     private MqttProperties mqttProperties;
    26.  
    27.     @Autowired
    28.     private MqttCallback mqttCallback;
    29.  
    30.     @Autowired
    31.     private MqttProperties properties;
    32.  
    33.     private static final String topic_hzd = "/hzd/pub/#";
    34.     private static final String topic_device_connect = "$SYS/brokers/+/clients/#";
    35.  
    36.     @PostConstruct
    37.     public void init() {
    38.         MqttClientPersistence mempersitence = new MemoryPersistence();
    39.         try {
    40.             if (client == null) {
    41.                 client = new MqttClient(mqttProperties.getBrokerUrl(), mqttProperties.getClientId(), mempersitence);
    42.             }
    43.         } catch (MqttException e) {
    44.             log.error("初始化客户端mqttClient对象失败,errormsg={},brokerUrl={},clientId={}", e.getMessage(), mqttProperties.getBrokerUrl(), mqttProperties.getClientId());
    45.         }
    46.         //连接服务端
    47.         connect(properties.getUsername(), properties.getPassword());
    48.         //订阅一个主题
    49.  
    50.         subscribe(topic_hzd, QosEnum.QoS2);
    51.         subscribe(topic_device_connect, QosEnum.QoS2);
    52.         log.info("EmqClient连接成功，订阅成功.topic={}，{}", topic_hzd, topic_device_connect);
    53.  
    54.     }
    55.  
    56.  
    57.     /**
    58.      * 连接broker
    59.      *
    60.      * @param username
    61.      * @param password
    62.      */
    63.     public void connect(String username, String password) {
    64.         MqttConnectOptions options = new MqttConnectOptions();
    65.         options.setAutomaticReconnect(true);
    66.         options.setUserName(username);
    67.         options.setPassword(password.toCharArray());
    68.         options.setCleanSession(true);
    69.         client.setCallback(mqttCallback);
    70.  
    71.         try {
    72.             client.connect(options);
    73.         } catch (MqttException e) {
    74.             log.error("mqtt客户端连接服务端失败,失败原因{}", e.getMessage());
    75.         }
    76.     }
    77.  
    78.     /**
    79.      * 断开连接
    80.      */
    81.     @PreDestroy
    82.     public void disConnect() {
    83.         try {
    84.             client.disconnect();
    85.         } catch (MqttException e) {
    86.             log.error("断开连接产生异常,异常信息{}", e.getMessage());
    87.         }
    88.     }
    89.  
    90.     /**
    91.      * 重连
    92.      */
    93.     public void reConnect() {
    94.         try {
    95.             client.reconnect();
    96.             while (true){
    97.                     if (client.isConnected()){
    98.                         subscribe(topic_hzd, QosEnum.QoS2);
    99.                         subscribe(topic_device_connect, QosEnum.QoS2);
    100.                         break;
    101.                     }else{
    102.                         Thread.sleep(100);
    103.                         log.info("重连中，请稍后......");
    104.                     }
    105.             }
    106.             log.info("EmqClient连接成功，订阅成功.topic={}，{}", topic_hzd, topic_device_connect);
    107.         } catch (MqttException | InterruptedException e) {
    108.             log.error("重连失败,失败原因{}", e.getMessage());
    109.         }
    110.     }
    111.  
    112.     /**
    113.      * 发布消息
    114.      *
    115.      * @param topic
    116.      * @param msg
    117.      * @param qos
    118.      * @param retain
    119.      */
    120.     public void publish(String topic, String msg, QosEnum qos, boolean retain) {
    121.  
    122.         MqttMessage mqttMessage = new MqttMessage();
    123.         mqttMessage.setPayload(msg.getBytes());
    124.         mqttMessage.setQos(qos.value());
    125.         mqttMessage.setRetained(retain);
    126.         try {
    127.             client.publish(topic, mqttMessage);
    128.         } catch (MqttException e) {
    129.             log.error("发布消息失败,errormsg={},topic={},msg={},qos={},retain={}", e.getMessage(), topic, msg, qos.value(), retain);
    130.         }
    131.  
    132.     }
    133.  
    134.     /**
    135.      * 订阅
    136.      *
    137.      * @param topicFilter
    138.      * @param qos
    139.      */
    140.     public void subscribe(String topicFilter, QosEnum qos) {
    141.         try {
    142.             client.subscribe(topicFilter, qos.value());
    143.         } catch (MqttException e) {
    144.             log.error("订阅主题失败,errormsg={},topicFilter={},qos={}", e.getMessage(), topicFilter, qos.value());
    145.         }
    146.  
    147.     }
    148.  
    149.     /**
    150.      * 取消订阅
    151.      *
    152.      * @param topicFilter
    153.      */
    154.     public void unSubscribe(String topicFilter) {
    155.         try {
    156.             client.unsubscribe(topicFilter);
    157.         } catch (MqttException e) {
    158.             log.error("取消订阅失败,errormsg={},topicfiler={}", e.getMessage(), topicFilter);
    159.         }
    160.     }
    161.  
    162. }

    2. MessageCallback.java         处理订阅消息

    • init() 加载异步入库处理类DataStorage
    • connectionLost(Throwable cause) 断线重连
    • messageArrived(String topic, MqttMessage message)  处理订阅消息
    1. package com.hzd.mqtt.client;
    2.  
    3. import com.hzd.mqtt.async.DataStorage;
    4. import lombok.extern.slf4j.Slf4j;
    5. import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    6. import org.eclipse.paho.client.mqttv3.MqttCallback;
    7. import org.eclipse.paho.client.mqttv3.MqttMessage;
    8. import org.springframework.beans.factory.annotation.Autowired;
    9. import org.springframework.stereotype.Component;
    10.  
    11. import javax.annotation.PostConstruct;
    12.  
    13. /**
    14.  *
    15.  */
    16. @Component
    17. @Slf4j
    18. public class MessageCallback implements MqttCallback {
    19.  
    20.  
    21.     @Autowired
    22.     private DataStorage dataStorage;
    23.  
    24.     @Autowired
    25.     private EmqClient emqClient;
    26.  
    27.     public static MessageCallback MessageCallback;
    28.  
    29.  
    30.     @PostConstruct //通过@PostConstruct实现初始化bean之前进行的操作
    31.     public void init() {
    32.         log.info("初始化");
    33.         MessageCallback = this;
    34. //        MessageCallback.dataPageRepository = this.dataPageRepository;
    35.         MessageCallback.dataStorage = this.dataStorage;
    36.     }
    37.  
    38.  
    39.     /**
    40.      * 丢失了对服务端的连接后触发的回调
    41.      *
    42.      * @param cause
    43.      */
    44.     @Override
    45.     public void connectionLost(Throwable cause) {
    46.         // 资源的清理  重连
    47.         log.info("断开emqx连接，开始重连......");
    48.         emqClient.reConnect();
    49.         log.info("emqx连接成功......");
    50.     }
    51.  
    52.     /**
    53.      * 应用收到消息后触发的回调
    54.      *
    55.      * @param topic
    56.      * @param message
    57.      * @throws Exception
    58.      */
    59.     @Override
    60.     public void messageArrived(String topic, MqttMessage message) throws Exception {
    61.         dataStorage.saveDataToMongo(topic, message);
    62.     }
    63.  
    64.     /**
    65.      * 消息发布者消息发布完成产生的回调
    66.      *
    67.      * @param token
    68.      */
    69.     @Override
    70.     public void deliveryComplete(IMqttDeliveryToken token) {
    71.         int messageId = token.getMessageId();
    72.         String[] topics = token.getTopics();
    73.         log.info("消息发布完成,messageid={},topics={}", messageId, topics);
    74.     }
    75. }


    3. DataStorage.java          异步入库处理类

    1. package com.hzd.mqtt.async;
    2.  
    3. import com.alibaba.fastjson2.JSON;
    4. import com.hzd.entity.po.ControlLog;
    5. import com.hzd.entity.po.DataExtend;
    6. import com.hzd.entity.po.MqttClientLog;
    7. import com.hzd.entity.vo.MqttPayload;
    8. import com.hzd.entity.vo.MqttTopic;
    9. import com.hzd.mqtt.enums.TopicType;
    10. import lombok.extern.slf4j.Slf4j;
    11. import org.eclipse.paho.client.mqttv3.MqttMessage;
    12. import org.springframework.data.mongodb.UncategorizedMongoDbException;
    13. import org.springframework.data.mongodb.core.MongoTemplate;
    14. import org.springframework.data.mongodb.core.query.Criteria;
    15. import org.springframework.data.mongodb.core.query.Query;
    16. import org.springframework.data.mongodb.core.query.Update;
    17. import org.springframework.scheduling.annotation.Async;
    18. import org.springframework.stereotype.Component;
    19.  
    20. import javax.annotation.Resource;
    21. import java.util.Date;
    22.  
    23.  
    24. /**
    25.  * Coding by 李炯 on 2022/10/28 15:12
    26.  */
    27. @Component
    28. @Slf4j
    29. public class DataStorage {
    30.  
    31.     @Resource
    32.     private MongoTemplate mongoTemplate;
    33.  
    34. //    @Resource
    35. //    private DataPageRepository dataPageRepository;
    36.  
    37.     @Async
    38.     public void saveDataToMongo(String topic, MqttMessage message) {
    39.         MqttTopic t = MqttTopic.init(topic);
    40.         String payload = new String(message.getPayload());
    41.         if (t == null) return;
    42.         if (payload.contains("\"type\":\"data\"")) {
    43.             DataExtend dataExtend = DataExtend.init(payload);
    44.             mongoTemplate.insert(dataExtend, t.getFac_id());
    45.             log.info("设备数据上报,topic={},fac_id={},messageid={},qos={},payload={}",
    46.                     topic, t.getFac_id(), message.getId(), message.getQos(), payload);
    47.         } else if (payload.contains("\"type\":\"config\"")) {
    48.             ControlLog controlLog = ControlLog.init(payload);
    49.             Query query = new Query(Criteria.where("_id").is(controlLog.get_id()));
    50.             Update update = Update.update("status", controlLog.getStatus());
    51.             update.set("status", controlLog.getStatus());
    52.             update.set("msg", controlLog.getMsg());
    53.             if (null != controlLog.getRelayIndexLst()) {
    54.                 update.set("relayIndexLst", controlLog.getRelayIndexLst());
    55.             }
    56.             if (null != controlLog.getRelayNumLst()) {
    57.                 update.set("relayNumLst", controlLog.getRelayNumLst());
    58.             }
    59.             if (null != controlLog.getReadInterval()) {
    60.                 update.set("readInterval", controlLog.getReadInterval());
    61.             }
    62.             if (null != controlLog.getRelayphoto()) {
    63.                 update.set("relayphoto", controlLog.getRelayphoto());
    64.             }
    65.             if (null != controlLog.getLongitude()) {
    66.                 update.set("longitude", controlLog.getLongitude());
    67.             }
    68.             if (null != controlLog.getLatitude()) {
    69.                 update.set("latitude", controlLog.getLatitude());
    70.             }
    71.             if (null != controlLog.getControlStatus()) {
    72.                 update.set("controlStatus", controlLog.getControlStatus());
    73.             }
    74.             if (null != controlLog.getReserve()) {
    75.                 update.set("reserve", controlLog.getReserve());
    76.             }
    77.             update.set("updateTime", new Date());
    78.             mongoTemplate.upsert(query, update, "log_control");
    79.             log.info("{},topic={},messageid={},qos={},payload={}",
    80.                     "配置返回", topic, message.getId(), message.getQos(), payload);
    81.         } else if (t.getTopicType() == TopicType.connect) {
    82.             MqttPayload pl = MqttPayload.init(payload);
    83.             MqttClientLog clientLog = MqttClientLog.init(pl);
    84.             updateConnectStatus(topic, clientLog);
    85.             log.info("{},topic={},messageid={},qos={},payload={}",
    86.                     clientLog.getStatus(), topic, message.getId(), message.getQos(), payload);
    87.         }
    88.     }
    89.  
    90.     @Async
    91.     public void updateConnectStatus(String topic, MqttClientLog clientLog) {
    92.         try{
    93.             MqttTopic t = MqttTopic.init(topic);
    94.             if (t == null) return;
    95.             mongoTemplate.insert(clientLog, "log_mqtt_client");
    96.             Query query = new Query(Criteria.where("username").is(clientLog.getUserName())
    97.                     .and("deviceid").is(clientLog.getDeviceId()));
    98.             Update update = Update.update("connectStatus", clientLog.getConnectStatus());
    99.             update.set("status", clientLog.getStatus());
    100.             update.set("updateTime", clientLog.getUpdateTime());
    101.             mongoTemplate.upsert(query, update, "status_mqtt_client");
    102.         }catch (Exception e){
    103.             log.error("mongoDB 错误：MqttClientLog={}，错误消息为{}",JSON.toJSONString(clientLog),e.getMessage());
    104.         }
    105.  
    106.  
    107.     }
    108. }

  • 相关阅读:
    笔记整体梳理
    tf.metrics
    微信支付 APP端 第三弹 申请退款
    2、NewObject的三种原型(来自GPT)
    Mysql索引学习笔记
    SSM+网上书店管理系统 毕业设计-附源码082255
    论文解读（gCooL）《Graph Communal Contrastive Learning》
    Matlab：基于Matlab实现人工智能算法应用的简介(SVM支撑向量机&GA遗传算法&PSO粒子群优化算法)、案例应用之详细攻略
    020—pandas 根据历史高考分段推断当前位次的分数
    Spring MVC 的返回值有哪些
  • 原文地址：https://blog.csdn.net/ladymorgana/article/details/127890217