Springboot集成mqtt【发布订阅模式】

桃扇骨 2021-06-10 20:39 670阅读 0赞

最近公司做项目需要用到mqtt，也是第一次接触mqtt，所以也是在摸索阶段，百度了很多现成的代码，根据项目的业务做了很多改动，直接上代码：

先导入jar包：

<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>

后面三个包我也不知道是干什么，但是我也没有删除测试，也不知道删除会不会影响功能，所以就放上了

这里我先写了一个公共的获取连接信息的类：

把配置里的 cleanSession 设为false，客户端掉线后 服务器端不会清除session，当重连后可以接收之前订阅主题的消息。当客户端上线后会接受到它离线的这段时间的消息，如果短线需要删除之前的消息则可以设置为true

public class MQTTConnect {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(MQTTConnect.class);
    
        public static final String HOST = "你的mqtt地址";
    
        private static final String clientid = "server";
    
        public MqttClient client;
        public MqttTopic topic;
        private String userName = "admin";
        private String passWord = "admin";
    
        //生成配置对象，用户名，密码等
        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;
        }
    
    
    }

这里是发布端，注意，如果发布端不需要回调的话则不需要此行代码`client.setCallback(new PushCallback());`  
因为我把发送消息的类写成了静态类，发布端在使用的时候可以直接**类名.发送消息的方法名**，在你需要的发送消息的地方这样写就可以了`MQTTServer.sendMQTTMessage(topic,sendData);`

/**
     * 发布端
     * Title:Server
     * Description: 服务器向多个客户端推送主题，即不同客户端可向服务器订阅相同主题
     * 
     */
    @Component
    public class MQTTServer {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(MQTTServer.class);
    
    
        public static final String HOST = "你的mqtt地址";
    
        private static final String clientid = "server";
    
        public MqttClient client;
        public MqttTopic topic;
    
        public MqttMessage message;
    
        private static MQTTConnect mqttConnect = new MQTTConnect();
    
        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 static void sendMQTTMessage(String topic,String data) throws MqttException {
    
            MQTTServer server = new MQTTServer();
            server.topic = server.client.getTopic(topic);
            server.message = new MqttMessage();
            //消息等级
            //level 0：最多一次的传输
            //level 1：至少一次的传输，(鸡肋)
            //level 2： 只有一次的传输
            server.message.setQos(2);
            //如果重复消费，则把值改为true,然后发送一条空的消息，之前的消息就会覆盖，然后在改为false
            server.message.setRetained(false);
    
            server.message.setPayload(data.getBytes());
            server.publish(server.topic , server.message);
    
    	}
    
    	public static void main(String[] args) throws MqttException {
    		sendMQTTMessage("你的发布主题","");
    	}
    
    }

这里是订阅端，这里有很多需要注意的：

**就是这里的clientId，服务器用来区分用户的，不能重复,clientId不能和发布的clientId一样，否则会出现频繁断开连接和重连的问题，不仅不能和发布的clientId一样，而且也不能和其他订阅的clientId一样，如果想要接收之前的离线数据，这就需要将client的 setCleanSession，设置为false，这样服务器才能保留它的session，再次建立连接的时候，它就会继续使用这个session了。  
这时此连接clientId 是不能更改的。如果连接一次就换一个clientId，但是订阅的主题是同一个则会把之前已经消费过的数据再次全部都消费一遍，因为换了clientId就相当于换了个用户，所以服务器要把缓存下来的信息给新的用户，也就是说一个clientId只能在同一台硬件机器上连接，这样就不会出现频繁的断开连接并进行重连的问题。  
但是其实还有一个问题，就是使用热部署的时候还是会出现频繁断开连接和重连的问题，可能是因为刚启动时的连接没断开，然后热部署的时候又进行了重连，重启一下就可以了**

再补充一下，mqtt的订阅是持久化的，如果订阅过很多主题的话，即使现在不再订阅了，在订阅的回调里还是会收到以前订阅的主题发送的消息，如果不想再收到以前订阅主题的消息，需要调用 `unsubscribe(要解除的主题)`方法取消订阅关系。

/**
     * 订阅端
     */
    @Component
    public class MQTTSubsribe {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(MQTTSubsribe.class);
    
        public static final String HOST = "你的mqtt地址";
    
        /**
         * 测试和正式环境不要使用同样的clientId 和主题
         * 如果和正式环境一样，正式环境启动后，本地再次启动会频繁断开重连，订阅的主题一样的话，测试的数据正式环境也会消费这些数据
         */
        private static final String clientid = "测试clientId";//测试
    
    //    private static final String clientid = "正式ClientID";//正式
    
        /**
         * 主题
         */
        private String topic = "测试环境主题";//测试
    
    //    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("----------连接成功");
            }
        }
    
        public void init() {
            try {
                connect();
                subscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 订阅某个主题，qos默认为0
         *
         * @param topic .
         */
        public void subscribe(String topic) {
            subscribe(topic,2);
        }
    
        /**
         * 订阅某个主题
         *
         * @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();
            }
        }
    }

回调：

/**  
     * 发布消息的回调类  
     *   
     * 必须实现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);
    
        /**
         * 主题
         */
        private String topic = "测试环境主题";//测试
    
    //    private String topic = "正式环境主题";//正式
    
    	private MQTTSubsribe mqttSubsribe;
    //    private MQTTSubsribe mqttSubsribe = new MQTTSubsribe();
    
        public PushCallback(MQTTSubsribe subsribe) {
            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());  
        }
     
        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);
            //这里可以针对收到的消息做处理
        }
    
    }

最后在springboot的启动类中添加：

@Autowired
        private MQTTSubsribe mqttSubsribe;
        
        /**
         * 接受订阅的接口和消息,mqtt消费端
         */
        @PostConstruct
        public void consumeMqttClient() throws MqttException {
            mqttSubsribe.init();
        }

现在拿mqtt测试工具测试一下订阅：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpcGVuZ2ZlaV8_size_16_color_FFFFFF_t_70]  
idea打印乱码真让人头痛，时好时坏，打印第一行是主题，第二行是qos，第三行是内容。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpcGVuZ2ZlaV8_size_16_color_FFFFFF_t_70]: /images/20210527/c341fcb9a18243bfa61bc9473b1be682.png