RuoYi-Vue前后端分离版集成MQTT

拼搏现实的明天。 2024-04-08 11:14 61阅读 0赞

1.  下载[MQTT X测试工具][MQTT X]，进行安装
    
    ![在这里插入图片描述][e54ccce8e5c14c1cbf42c8326f67258a.png_pic_center]
2.  打开MQTT X，新建连接，输入名称，点击连接
    
    ![在这里插入图片描述][7ea595883b214737be1e30f4fce48b4b.png_pic_center]
3.  在RuoYi-Vue项目的ruoyi-common模块下的pom.xml加入jar包依赖
    
        
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>
4.  在application.yml添加相关配置，加在Spring配置下
    
        # mqtt
          mqtt:
            username: admin # 用户名
            password: admin # 密码
            hostUrl: tcp://broker.emqx.io:1883 # tcp://ip:端口
            clientId: clientId # 客户端id
            defaultTopic: topic,topic1 # 订阅主题
            timeout: 100 # 超时时间 （单位：秒）
            keepalive: 60 # 心跳 （单位：秒）
            enabled: true # 是否使用mqtt功能
    
    ![在这里插入图片描述][b938224977f24c16aa33845e4fd66a60.png_pic_center]
5.  在ruoyi-common\\src\\main\\java\\com\\ruoyi\\common\\utils目录下新建mqtt文件夹，添加以下三个文件
    
    > MqttConfig.java
    
        package com.ruoyi.common.utils.mqtt;
        
        import com.ruoyi.common.utils.StringUtils;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.beans.factory.annotation.Value;
        import org.springframework.boot.context.properties.ConfigurationProperties;
        import org.springframework.context.annotation.Bean;
        import org.springframework.context.annotation.Configuration;
        import org.springframework.stereotype.Component;
        
        @Component
        @ConfigurationProperties("spring.mqtt")
        public class MqttConfig {
              
            @Autowired
            private MqttPushClient mqttPushClient;
        
            /**
             * 用户名
             */
            private String username;
            /**
             * 密码
             */
            private String password;
            /**
             * 连接地址
             */
            private String hostUrl;
            /**
             * 客户Id
             */
            private String clientId;
            /**
             * 默认连接话题
             */
            private String defaultTopic;
            /**
             * 超时时间
             */
            private int timeout;
            /**
             * 保持连接数
             */
            private int keepalive;
            /**
             * mqtt功能使能
             */
            private boolean enabled;
        
            public String getUsername() {
              
                return username;
            }
        
            public void setUsername(String username) {
              
                this.username = username;
            }
        
            public String getPassword() {
              
                return password;
            }
        
            public void setPassword(String password) {
              
                this.password = password;
            }
        
            public String getHostUrl() {
              
                return hostUrl;
            }
        
            public void setHostUrl(String hostUrl) {
              
                this.hostUrl = hostUrl;
            }
        
            public String getClientId() {
              
                return clientId;
            }
        
            public void setClientId(String clientId) {
              
                this.clientId = clientId;
            }
        
            public String getDefaultTopic() {
              
                return defaultTopic;
            }
        
            public void setDefaultTopic(String defaultTopic) {
              
                this.defaultTopic = defaultTopic;
            }
        
            public int getTimeout() {
              
                return timeout;
            }
        
            public void setTimeout(int timeout) {
              
                this.timeout = timeout;
            }
        
            public int getKeepalive() {
              
                return keepalive;
            }
        
            public void setKeepalive(int keepalive) {
              
                this.keepalive = keepalive;
            }
        
            public boolean isEnabled() {
              
                return enabled;
            }
        
            public void setEnabled(boolean enabled) {
              
                this.enabled = enabled;
            }
        
            @Bean
            public MqttPushClient getMqttPushClient() {
              
                if(enabled == true){
              
                    String mqtt_topic[] = StringUtils.split(defaultTopic, ",");
                    mqttPushClient.connect(hostUrl, clientId, username, password, timeout, keepalive);//连接
                    for(int i=0; i<mqtt_topic.length; i++){
              
                        mqttPushClient.subscribe(mqtt_topic[i], 0);//订阅主题
                    }
                }
                return mqttPushClient;
            }
        }
    
    > MqttPushClient.java
    
        package com.ruoyi.common.utils.mqtt;
        
        import com.ruoyi.common.core.domain.AjaxResult;
        import org.eclipse.paho.client.mqttv3.*;
        import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
        import org.slf4j.Logger;
        import org.slf4j.LoggerFactory;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.stereotype.Component;
        
        import static com.ruoyi.common.core.domain.AjaxResult.error;
        import static com.ruoyi.common.core.domain.AjaxResult.success;
        
        @Component
        public class MqttPushClient {
              
            private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);
        
            @Autowired
            private PushCallback pushCallback;
        
            private static MqttClient client;
        
            private static MqttClient getClient() {
              
                return client;
            }
        
            private static void setClient(MqttClient client) {
              
                MqttPushClient.client = client;
            }
        
            /**
             * 客户端连接
             *
             * @param host      ip+端口
             * @param clientID  客户端Id
             * @param username  用户名
             * @param password  密码
             * @param timeout   超时时间
             * @param keepalive 保留数
             */
            public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
              
                MqttClient client;
                try {
              
                    client = new MqttClient(host, clientID, new MemoryPersistence());
                    MqttConnectOptions options = new MqttConnectOptions();
                    options.setCleanSession(true);
                    options.setUserName(username);
                    options.setPassword(password.toCharArray());
                    options.setConnectionTimeout(timeout);
                    options.setKeepAliveInterval(keepalive);
                    MqttPushClient.setClient(client);
                    try {
              
                        client.setCallback(pushCallback);
                        client.connect(options);
                    } catch (Exception e) {
              
                        e.printStackTrace();
                    }
                } catch (Exception e) {
              
                    e.printStackTrace();
                }
            }
        
            /**
             * 发布
             *
             * @param qos         连接方式
             * @param retained    是否保留
             * @param topic       主题
             * @param pushMessage 消息体
             */
            public AjaxResult publish(int qos, boolean retained, String topic, String pushMessage) {
              
                MqttMessage message = new MqttMessage();
                message.setQos(qos);
                message.setRetained(retained);
                message.setPayload(pushMessage.getBytes());
                MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
                if (null == mTopic) {
              
                    logger.error("topic not exist");
                }
                MqttDeliveryToken token;
                try {
              
                    token = mTopic.publish(message);
                    token.waitForCompletion();
                    return success();
                } catch (MqttPersistenceException e) {
              
                    e.printStackTrace();
                    return error();
                } catch (MqttException e) {
              
                    e.printStackTrace();
                    return error();
                }
            }
        
            /**
             * 订阅某个主题
             *
             * @param topic 主题
             * @param qos   连接方式
             */
            public void subscribe(String topic, int qos) {
              
                logger.info("开始订阅主题" + topic);
                try {
              
                    MqttPushClient.getClient().subscribe(topic, qos);
                } catch (MqttException e) {
              
                    e.printStackTrace();
                }
            }
        
        }
    
    > PushCallback.java
    
        package com.ruoyi.common.utils.mqtt;
        
        import com.alibaba.fastjson2.JSONObject;
        import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
        import org.eclipse.paho.client.mqttv3.MqttCallback;
        import org.eclipse.paho.client.mqttv3.MqttClient;
        import org.eclipse.paho.client.mqttv3.MqttMessage;
        import org.slf4j.Logger;
        import org.slf4j.LoggerFactory;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.stereotype.Component;
        
        @Component
        public class PushCallback implements MqttCallback {
              
            private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);
        
            @Autowired
            private MqttConfig mqttConfig;
        
            private static MqttClient client;
        
            private static String _topic;
            private static String _qos;
            private static String _msg;
        
            @Override
            public void connectionLost(Throwable throwable) {
              
                // 连接丢失后，一般在这里面进行重连
                logger.info("连接断开，可以做重连");
                if (client == null || !client.isConnected()) {
              
                    mqttConfig.getMqttPushClient();
                }
            }
        
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
              
                // subscribe后得到的消息会执行到这里面
                logger.info("接收消息主题 : " + topic);
                logger.info("接收消息Qos : " + mqttMessage.getQos());
                logger.info("接收消息内容 : " + new String(mqttMessage.getPayload()));
        
                _topic = topic;
                _qos = mqttMessage.getQos()+"";
                _msg = new String(mqttMessage.getPayload());
            }
        
            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
              
                logger.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
            }
        
            //别的Controller层会调用这个方法来  获取  接收到的硬件数据
            public String receive() {
              
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("topic", _topic);
                jsonObject.put("qos", _qos);
                jsonObject.put("msg", _msg);
                return jsonObject.toString();
            }
        
        }
    
    **注：项目报红的地方可以按Alt+Enter键导包**
6.  运行若依项目，后端可以接收MQTTX发来的数据
    
    ![在这里插入图片描述][43a920af141942bca2b269ba6fcf2c0c.png_pic_center]

[MQTT X]: https://www.emqx.com/zh/try?product=MQTTX
[e54ccce8e5c14c1cbf42c8326f67258a.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/2f4a5c34f94d4791bb920be6949001a5.png
[7ea595883b214737be1e30f4fce48b4b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e3effe2f7cdf411886596c047bd1ed34.png
[b938224977f24c16aa33845e4fd66a60.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/892c100c578d4a2e82bb4b43b0033d57.png
[43a920af141942bca2b269ba6fcf2c0c.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/a7e29c3fcf3d4162846141fb18cf1ffd.png