Springboot 写一个mqtt 发布/订阅案例(一)

深藏阁楼爱情的钟 2023-02-14 10:49 409阅读 0赞

# Springboot 写一个mqtt 发布/订阅案例 #

### 一、配置MQTTfx 软件 ###

MQTTfx 这个软件 能够模拟 “发布”功能，也能模拟“订阅”功能。  
**这里的 “地址和端口” 与 程序中 “mqtt地址” 是一致的，还有用户名和密码。**  
（这里的地址，是我这边的服务器地址）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center]

# 二、直接贴代码 #

### 2.1 添加依赖 ###

<dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.0</version>
    </dependency>
    
    <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>

### 2.2 springboot的启动类 ###

package com.sprintboot_mqtt.boot_mqtt;
    
    import com.sprintboot_mqtt.boot_mqtt.config.MQTTServer;
    import com.sprintboot_mqtt.boot_mqtt.config.MQTTSubsribe;
    import com.sprintboot_mqtt.boot_mqtt.config.PushCallback;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import javax.annotation.PostConstruct;
    
    
    @SpringBootApplication
    public class BootMqttApplication {
        
    
        //最后在springboot的启动类中添加
        @Autowired
        private MQTTSubsribe mqttSubsribe;
        ///private MQTTServer mqttServer;
        //接受订阅的接口和消息,mqtt消费端
        private MQTTServer mqttServer;
        {
        
            try {
        
                mqttServer = new MQTTServer();
            } catch (MqttException e) {
        
                e.printStackTrace();
            }
        }
    
        private PushCallback pushCallback;
    
    
        @PostConstruct
        public void consumeMqttClient() throws MqttException {
        
    
            mqttSubsribe.init();           // 订阅 消息
            mqttServer.PublishMessage();   // 发布 消息
        }
    
        public static void main(String[] args) {
        
            SpringApplication.run(BootMqttApplication.class, args);
        }
    }

### 2.3 新建类 MQTTConnect ###

package com.sprintboot_mqtt.boot_mqtt.config;
    
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    
    
    public class MQTTConnect {
        
        private String userName = "client";  //生成配置对象，用户名，密码等
        private String passWord = "client";
    
    //    * 把配置里的 cleanSession 设为false，客户端掉线后 服务器端不会清除session，
    //    * 当重连后可以接收之前订阅主题的消息。当客户端上线后会接受到它离线的这段时间的消息，
    //    * 如果短线需要删除之前的消息则可以设置为true
        public MqttConnectOptions getOptions() {
        
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(userName);
            options.setPassword(passWord.toCharArray());
            options.setConnectionTimeout(10);
            //设置心跳
            options.setKeepAliveInterval(20);
            return options;
        }
    
        public MqttConnectOptions getOptions(MqttConnectOptions options) {
        
    
            options.setCleanSession(false);
            options.setUserName(userName);
            options.setPassword(passWord.toCharArray());
            options.setConnectionTimeout(10);
            options.setKeepAliveInterval(20);
            return options;
        }
    
    
    }

### 2.4 新建类 MQTTServer ###

package com.sprintboot_mqtt.boot_mqtt.config;
    
    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.stereotype.Component;
    import java.util.concurrent.TimeUnit;
    
    
    /**
     * 发布端
     * Title:Server
     * Description: 服务器向多个客户端推送主题，即不同客户端可向服务器订阅相同主题
     **/
    @Component
    public class MQTTServer {
        
    
        private static final Logger LOGGER = LoggerFactory.getLogger(MQTTServer.class);
        public static final String HOST = "tcp://177.177.7.177:1777";  //你的mqtt地址  （2）
        private static final String clientid = "server";
    
        public MqttClient client;
        public MqttTopic topic;
        public MqttMessage message;
    
        private static MQTTConnect mqttConnect = new MQTTConnect();
        String Message_Subsrib_To_Publish = "start";
    
        public MQTTServer() throws MqttException {
        
            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
    //        client = new MqttClient(HOST, clientid, new MemoryPersistence());
            connect();
        }
    
    
        public void connect() throws MqttException {
        
            //防止重复创建MQTTClient实例
            if (client==null) {
        
                //就是这里的clientId，服务器用来区分用户的，不能重复
                client = new MqttClient(HOST, clientid, new MemoryPersistence());// MemoryPersistence设置clientid的保存形式，默认为以内存保存
    //            client.setCallback(new PushCallback());
            }
            MqttConnectOptions options = mqttConnect.getOptions();
            //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
            if (!client.isConnected()) {
        
                client.connect(options);
                LOGGER.info("---------------------连接成功");
            }else {
        //这里的逻辑是如果连接成功就重新连接
                client.disconnect();
                client.connect(mqttConnect.getOptions(options));
                LOGGER.info("---------------------连接成功");
            }
        }
    
    
        public  boolean publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException,
                MqttException {
        
    
            MqttDeliveryToken token = topic.publish(message);
            token.waitForCompletion();
            System.out.println("message is published completely! "
                    + token.isComplete());
            return token.isComplete();
        }
    
        /**
         * MQTT发送指令
         // @param page
         // @param equipment
         * @return
         * @throws MqttException
         */
        public void sendMQTTMessage(String topic,String data) throws MqttException {
        
            MQTTServer server = new MQTTServer();
            server.topic = server.client.getTopic(topic);
            server.message = new MqttMessage();
            server.message.setQos(2);           //消息等级//level 0：最多一次的传输  //level 1：至少一次的传输，(鸡肋)  //level 2： 只有一次的传输
            server.message.setRetained(false);  //如果重复消费，则把值改为true,然后发送一条空的消息，之前的消息就会覆盖，然后在改为false
            server.message.setPayload(data.getBytes());
            server.publish(server.topic , server.message);
        }
    
    
        public void PublishMessage() throws MqttException {
        
            while (true) {
        
                System.out.println("Message_Subsrib_To_Publish22222:  " + Message_Subsrib_To_Publish);
                try {
        
                    TimeUnit.SECONDS.sleep(1);
                    //取出messageMap数据
                    //发送数据
                    sendMQTTMessage("jowoiot/toServer/bruce/lptestpub", "242332ggggggggggggggggggggggggggg35234");  // 你的发布主题  ,,,
                    Message_Subsrib_To_Publish = null;
                    System.out.println("Message_Subsrib_To_Publish44444:  " + Message_Subsrib_To_Publish);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
    
    }

### 2.5 新建类 MQTTSubsribe ###

package com.sprintboot_mqtt.boot_mqtt.config;
    
    import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    
    /**
     * 订阅端
     */
    @Component
    public class MQTTSubsribe {
        
        private static final Logger LOGGER = LoggerFactory.getLogger(MQTTSubsribe.class);
        public static final String HOST = "tcp://177.177.7.177:1777";  //你的mqtt地址   
        // 测试和正式环境不要使用同样的clientId 和主题
        //如果和正式环境一样，正式环境启动后，本地再次启动会频繁断开重连，订阅的主题一样的话，测试的数据正式环境也会消费这些数据
        private static final String clientid = "lptestsub";//测试clientId   
    //    private static final String clientid = "正式ClientID";//正式
    
        private String topic = "jowoiot/toServer/bruce/lptestsub";//测试环境主题  
    //    private String topic = "正式环境主题";//正式
    
        public MqttClient client;
        private MQTTConnect mqttConnect= new MQTTConnect();
    
    
    
        //方法实现说明 断线重连方法，如果是持久订阅，重连是不需要再次订阅，如果是非持久订阅，重连是需要重新订阅主题 取决于options.setCleanSession(true);
        // true为非持久订阅
        public void connect() throws MqttException {
        
            //防止重复创建MQTTClient实例
            if (client==null) {
        
                //就是这里的clientId，服务器用来区分用户的，不能重复,clientId不能和发布的clientId一样，否则会出现频繁断开连接和重连的问题
                //不仅不能和发布的clientId一样，而且也不能和其他订阅的clientId一样，如果想要接收之前的离线数据，这就需要将client的 setCleanSession
                // 设置为false，这样服务器才能保留它的session，再次建立连接的时候，它就会继续使用这个session了。 这时此连接clientId 是不能更改的。
                //但是其实还有一个问题，就是使用热部署的时候还是会出现频繁断开连接和重连的问题，可能是因为刚启动时的连接没断开，然后热部署的时候又进行了重连，重启一下就可以了
                //+ System.currentTimeMillis()
                client = new MqttClient(HOST, clientid, new MemoryPersistence());// MemoryPersistence设置clientid的保存形式，默认为以内存保存
                //如果是订阅者则添加回调类，发布不需要
                client.setCallback(new PushCallback(MQTTSubsribe.this));
    //            client.setCallback(new PushCallback());
            }
            MqttConnectOptions options = mqttConnect.getOptions();
            //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
            if (!client.isConnected()) {
        
                client.connect(options);
                LOGGER.info("----------连接成功");
            }else {
        //这里的逻辑是如果连接成功就重新连接
                client.disconnect();
                client.connect(mqttConnect.getOptions(options));
                LOGGER.info("----------连接成功");
            }
        }
    
    
    
        /**
         * 订阅某个主题，qos默认为0
         * @param topic .
         * @param qos .
         */
        public void subscribe(String topic, int qos) {
        
    
            try {
        
                client.subscribe(topic,2);
                //MQTT 协议中订阅关系是持久化的，因此如果不需要订阅某些 Topic，需要调用 unsubscribe 方法取消订阅关系。
    //            client.unsubscribe("需要解除订阅关系的主题");
            } catch (MqttException e) {
        
                e.printStackTrace();
            }
        }
    
    
        public void init() {
        
    
            try {
        
                connect();
                subscribe(topic,2);
            } catch (MqttException e) {
        
                e.printStackTrace();
            }
    }

### 2.6 新建类 PushCallback ###

package com.sprintboot_mqtt.boot_mqtt.config;
    
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.concurrent.TimeUnit;
    
    
    /**
     * 发布消息的回调类
     *
     * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
     * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
     * 在回调中，将它用来标识已经启动了该回调的哪个实例。
     * 必须在回调类中实现三个方法：
     *
     *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
     *
     *  public void connectionLost(Throwable cause)在断开连接时调用。
     *
     *  public void deliveryComplete(MqttDeliveryToken token))
     *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
     *  由 MqttClient.connect 激活此回调。
     *
     */
    
    public class PushCallback implements MqttCallback {
        
    
        private static final Logger LOGGER = LoggerFactory.getLogger(PushCallback.class);
        MQTTServer mqttServer = new MQTTServer();
    
        /**
         * 主题
         */
        private String topic = "jowoiot/toServer/bruce/lptestsub";//测试环境主题  
    //    private String topic = "正式环境主题";//正式
    
        private MQTTSubsribe mqttSubsribe;
    //    private MQTTSubsribe mqttSubsribe = new MQTTSubsribe();
    
        public PushCallback(MQTTSubsribe subsribe) throws MqttException {
        
            this.mqttSubsribe = subsribe;
        }
    
    
        public void connectionLost(Throwable cause) {
        
            // 连接丢失后，一般在这里面进行重连
            LOGGER.info("---------------------连接断开，可以做重连");
            // deliveryComplete(null);
    
            while (true){
        
                try {
        //如果没有发生异常说明连接成功，如果发生异常，则死循环
                    Thread.sleep(1000);
                    mqttSubsribe.init();
                    break;
                }catch (Exception e){
        
    //                e.printStackTrace();
                    continue;
                }
            }
    
        }
    
        public void deliveryComplete(IMqttDeliveryToken token) {
        
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    
    
        //  如何 将messageArrived 中的 result 引用到MQTTServer中还是一个问题
    //    public String getMessage_Subsrib_To_Publish(String message) {
        
    //        System.out.println("Message_Subsrib_To_Publish11111:   " + message);
    //        return message;
    //    }
    
    
        public void messageArrived(String topic, MqttMessage message) throws Exception {
        
            // subscribe后得到的消息会执行到这里面
            String result = new String(message.getPayload(),"UTF-8");
            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + result);
            //这里可以针对收到的消息做处理
        }
    
    }

# 三、mqttfx软件和程序调试展示 #

### 3.1 程序端“发布”，MQTTfx软件“订阅” ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center 1]

### 3.2 MQTTfx软件“发布”，程序端“订阅” ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center 2]

注：  
代码还可以参考文章：[springboot集成mqtt发布订阅][springboot_mqtt]

#### 这篇文章订阅和发布 的消息是不同的，是相互独立的，就是消息的内容没什么关系。 ####

#### 后续还有一篇文章，实现将 订阅的消息，同时再 发布出去。 ####

[Springboot 写一个mqtt 发布/订阅案例(二)][Springboot _mqtt]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20200602173634263.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20200602165717131.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20200602165733836.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3Nzg4Mzgz,size_16,color_FFFFFF,t_70#pic_center
[springboot_mqtt]: https://blog.csdn.net/sipengfei_/article/details/102646550
[Springboot _mqtt]: https://blog.csdn.net/qq_37788383/article/details/106545561