基于Springboot实现Mqtt

心已赠人 2022-11-03 05:49 141阅读 0赞

转载：[基于Springboot实现Mqtt][Springboot_Mqtt]

**Java端开发：**  
pom.xml：

<dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.2</version>
    </dependency>

Client.java

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 javax.annotation.PostConstruct;
     
    @Component
    public class Client { 
        Logger logger = LoggerFactory.getLogger(Client.class);
     
        @Autowired
        private OnMessageCallback onMessageCallback;
     
        public final String HOST = "tcp://连接ip:端口号"; //连接ip
        public final String TOPIC = "duilie"; //订阅队列
        public final String clientId = "jy";  //连接id，注意不要和已有的连接id重复
     
        private MqttClient client;
        private MqttConnectOptions connOpts;
        private String userName = "userName";
        private String passWord = "passWord";
     
        @PostConstruct
        public void init() { 
            start();
        }
     
        public void start() { 
            try { 
                client = new MqttClient(HOST, clientId, new MemoryPersistence());
                connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true); // 清空session
                connOpts.setUserName(userName);
                connOpts.setPassword(passWord.toCharArray());
                connOpts.setConnectionTimeout(10);// 设置超时时间
                connOpts.setKeepAliveInterval(20);// 设置会话心跳时间
                client.setCallback(onMessageCallback); // 设置回调函数
                System.out.println("Connecting to broker: " + HOST);
                int Qos = 1;
                client.connect(connOpts); // 创立连接
                client.subscribe(TOPIC); // 订阅
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
     
        public void publish(String topic, byte[] payload) { 
            try { 
                this.client.publish(topic, payload, 1, false);
            } catch (MqttException e) { 
                e.printStackTrace();
            }
        }
     
        public void close() { 
            try { 
                this.client.close();
            } catch (MqttException e) { 
                e.printStackTrace();
            }
        }
     
        public void reConnect() throws Exception { 
            if (this.client != null) { 
                this.logger.info("开始重连");
                this.client.connect(this.connOpts);
                int Qos = 1;
                this.logger.info("主题：" + TOPIC);
                this.client.subscribe(TOPIC, Qos);
            }
        }
     
    }

OnMessageCallback.java  
回调消息处理类OnMessageCallback：

import org.eclipse.paho.client.mqttv3.MqttCallback;
     
     
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
     
    @Component
    public class OnMessageCallback implements MqttCallback { 
     
        @Autowired
        private ServiceImpl service; //修改成自己的业务
        @Autowired
        private Client client;
     
        public void connectionLost(Throwable cause) { 
            // 连接丢失后，一般在这里面进行重连
            System.out.println("连接断开，可以做重连");
            while (true) { 
                try { 
                    Thread.sleep(5000L);
                    this.client.reConnect();
                    break;
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
        }
     
        public void messageArrived(String topic, MqttMessage message) throws Exception { 
            // subscribe后得到的消息会执行到这里面
            System.out.println("接收消息主题:" + topic);
            System.out.println("接收消息Qos:" + message.getQos());
            System.out.println("接收消息内容:" + new String(message.getPayload()));
            service.insert(new String(message.getPayload()));  //修改成自己的业务
        }
     
        public void deliveryComplete(IMqttDeliveryToken token) { 
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    }

这里加入了断线重连。在 messageArrived 方法中调用业务即可。

**连接成功会在控制台可以看到连接的id**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1NDMwNTYz_size_16_color_FFFFFF_t_70]  
**Qos参数：**  
level 0：最多一次的传输

消息是基于TCP/IP网络传输的。没有回应，在协议中也没有定义重传的语义。消息可能到达服务器1次，也可能根本不会到达。  
level 1：至少一次的传输（一般配置为1）

服务器接收到消息会被确认，通过传输一个PUBACK信息。如果有一个可以辨认的传输失败，无论是通讯连接还是发送设备，还是过了一段时间确认信息没有收到，发送方都会将消息头的DUP位置1，然后再次发送消息。消息最少一次到达服务器。SUBSCRIBE和UNSUBSCRIBE都使用level 1 的QoS。  
如果客户端没有接收到PUBACK信息（无论是应用定义的超时，还是检测到失败然后通讯session重启），客户端都会再次发送PUBLISH信息，并且将DUP位置1。  
当它从客户端接收到重复的数据，服务器重新发送消息给订阅者，并且发送另一个PUBACK消息。  
level 2： 只有一次的传输

在QoS level 1上附加的协议流保证了重复的消息不会传送到接收的应用。这是最高级别的传输，当重复的消息不被允许的情况下使用。这样增加了网络流量，但是它通常是可以接受的，因为消息内容很重要。  
QoS level 2在消息头有Message ID。  
接收消息和写入数据库解藕：

这里是直接调用 server 写入数据库的写法。可以改成接收到消息写入到一个 BlockingQueue 里，再由业务层去取数。

遇见问题：  
clientId：是每一个去订阅连接的 client 的名称，注意不要和原有的的连接名重复。  
Mqtt是不会消息堆积，也就是我这里设备一直发送消息，而我的java客户端断了5分钟，那么这5分钟的消息在重连时是获取不到的，只能获得之后的消息。（具体的原理没有研究）

[Springboot_Mqtt]: https://blog.csdn.net/weixin_42765516/article/details/113778830?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522161439388716780274146538%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fall.%2522%257D&request_id=161439388716780274146538&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~first_rank_v2~rank_v29-3-113778830.pc_search_result_no_baidu_js&utm_term=mqtt%20%20%20OnMessageCallback%20onMessageCallback
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1NDMwNTYz_size_16_color_FFFFFF_t_70]: /images/20221024/5ee9ef4e8fd44fb49d2cee2afee79b9d.png