物联网之对接MQTT最佳实践

小伙伴们，你们好呀，我是老寇，跟我一起学习对接MQTT
安装EMQX

采用docker-compose一键式启动！！！
还没有安装docker朋友，参考文章下面两篇文章
# Ubuntu20.04安装Docker
# Centos7安装Docker 23.0.6
使用 emqx 5.4.1，按照老夫的教程来，请不要改版本号！！！

使用 emqx 5.4.1，按照老夫的教程来，请不要改版本号！！！

使用 emqx 5.4.1，按照老夫的教程来，请不要改版本号！！！

1. services:
2.     emqx:
3.       image: emqx/emqx:5.4.1
4.       container_name: emqx
5.       # 保持容器在没有守护程序的情况下运行
6.       tty: true
7.       restart: always
8.       privileged: true
9.       ports:
10.         - "1883:1883"
11.         - "8083:8083"
12.         - "8883:8883"
13.         - "18083:18083"
14.       environment:
15.         - TZ=Asia/Shanghai
16.       volumes:
17.         # 挂载数据存储
18.         - ./emqx/data:/opt/emqx/data
19.         # 挂载日志文件
20.         - ./emqx/log:/opt/emqx/log
21.       networks:
22.         - laokou_network
23. networks:
24.   laokou_network:
25.     driver: bridge

复制代码

访问 http://127.0.0.1:18083 设置密码

EMQX MQTT【摘抄自官方文档】

EMQX官方文档
MQTT 是物联网 (IoT) 的 OASIS 标准消息传递协议。它被设计为一种极轻量的发布/订阅消息传输协议，非常适合以较小的代码占用空间和极低的网络带宽连接远程设备。MQTT 目前广泛应用于汽车、制造、电信、石油和天然气等众多行业。
EMQX 完全兼容 MQTT 5.0 和 3.x，本节将介绍 MQTT 相关功能的基本配置项，包括基本 MQTT 设置、订阅设置、会话设置、强制关闭设置和强制垃圾回收设置等
客户端对接

本文章采用三种客户端对接
维度PahoHivemq-MQTT-ClientVert.x MQTT Client协议支持MQTT 3.1.1（5.0 实验性）MQTT 5.0 完整支持MQTT 5.0（较新版本）性能中（同步模式）高（异步非阻塞）极高（响应式架构）依赖复杂度低中（仅 Netty）高（需 Vert.x 生态）社区资源丰富较少中等适用场景传统 IoT、跨语言项目企业级 MQTT 5.0、高吞吐响应式系统、高并发微服务Paho【不推荐，连接不稳定】

Paho代码地址
引入依赖

1. <dependencies>
2.     <dependency>
3.         <groupId>org.eclipse.paho</groupId>
4.         org.eclipse.paho.mqttv5.client</artifactId>
5.         <version>1.2.5</version>
6.     </dependency>
7.     <dependency>
8.         <groupId>org.eclipse.paho</groupId>
9.         org.eclipse.paho.client.mqttv3</artifactId>
10.         <version>1.2.5</version>
11.     </dependency>
12. </dependencies>

复制代码

项目集成

PahoProperties

1. /**
2. * @author laokou
3. */
4. @Data
5. public class PahoProperties {
7.     private boolean auth = true;
9.     private String username = "emqx";
11.     private String password = "laokou123";
13.     private String host = "127.0.0.1";
15.     private int port = 1883;
17.     private String clientId;
19.     private int subscribeQos = 1;
21.     private int publishQos = 0;
23.     private int willQos = 1;
25.     private int connectionTimeout = 60;
27.     private boolean manualAcks = false;
29.     // @formatter:off
30.     /**
31.      * 控制是否创建新会话（true=新建，false=复用历史会话）. clearStart=true => Broker 会在连接断开后立即清除所有会话信息.
32.      * clearStart=false => Broker 会在连接断开后保存会话信息，并在重新连接后复用会话信息.
33.      * ...
34.      */
35.     // @formatter:on
36.     private boolean clearStart = false;
38.     private int receiveMaximum = 10000;
40.     private int maximumPacketSize = 10000;
42.     // @formatter:off
43.     /**
44.      * 默认会话保留一天.
45.      * 最大值，4294967295L，会话过期时间【永不过期，单位秒】.
46.      * 定义客户端断开后会话保留的时间（仅在 Clean Session = false 时生效）.
47.      */
48.     private long sessionExpiryInterval = 86400L;
49.     // @formatter:on
51.     /**
52.      * 心跳包每隔60秒发一次.
53.      */
54.     private int keepAliveInterval = 60;
56.     private boolean automaticReconnect = true;
58.     private Set<String> topics = new HashSet<>(0);
60. }

复制代码

PahoMqttClientMessageCallbackV5

1. /**
2. * @author laokou
3. */
4. @Slf4j
5. @RequiredArgsConstructor
6. public class PahoMqttClientMessageCallbackV5 implements MqttCallback {
8.     private final List<MessageHandler> messageHandlers;
10.     @Override
11.     public void disconnected(MqttDisconnectResponse disconnectResponse) {
12.        log.error("【Paho-V5】 => MQTT关闭连接");
13.     }
15.     @Override
16.     public void mqttErrorOccurred(MqttException ex) {
17.        log.error("【Paho-V5】 => MQTT报错，错误信息：{}", ex.getMessage());
18.     }
20.     @Override
21.     public void messageArrived(String topic, MqttMessage message) {
22.        for (MessageHandler messageHandler : messageHandlers) {
23.           if (messageHandler.isSubscribe(topic)) {
24.              log.info("【Paho-V5】 => MQTT接收到消息，Topic：{}", topic);
25.              messageHandler.handle(new org.laokou.sample.mqtt.handler.MqttMessage(message.getPayload(), topic));
26.           }
27.        }
28.     }
30.     @Override
31.     public void deliveryComplete(IMqttToken token) {
32.        log.info("【Paho-V5】 => MQTT消息发送成功，消息ID：{}", token.getMessageId());
33.     }
35.     @Override
36.     public void connectComplete(boolean reconnect, String uri) {
37.        if (reconnect) {
38.           log.info("【Paho-V5】 => MQTT重连成功，URI：{}", uri);
39.        }
40.        else {
41.           log.info("【Paho-V5】 => MQTT建立连接，URI：{}", uri);
42.        }
43.     }
45.     @Override
46.     public void authPacketArrived(int reasonCode, MqttProperties properties) {
47.        log.info("【Paho-V5】 => 接收到身份验证数据包：{}", reasonCode);
48.     }
50. }

复制代码

PahoV5MqttClientTest

1. /**
2. * @author laokou
3. */
4. @SpringBootTest
5. @RequiredArgsConstructor
6. @TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
7. class PahoV5MqttClientTest {
9.     private final List<MessageHandler> messageHandlers;
11.     @Test
12.     void testMqttClient() throws InterruptedException {
13.         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(16);
15.         PahoProperties pahoProperties = new PahoProperties();
16.         pahoProperties.setClientId("test-client-3");
17.         pahoProperties.setTopics(Set.of("/test-topic-3/#"));
18.         PahoMqttClientV5 pahoMqttClientV5 = new PahoMqttClientV5(pahoProperties, messageHandlers, scheduledExecutorService);
19.         pahoMqttClientV5.open();
20.         Thread.sleep(1000);
21.         pahoMqttClientV5.publish("/test-topic-3/789", "Hello World789".getBytes());
22.     }
24. }

复制代码

PahoMqttClientMessageCallbackV3

1. /**
2. * @author laokou
3. */
4. @Slf4j
5. @RequiredArgsConstructor
6. public class PahoMqttClientMessageCallbackV3 implements MqttCallback {
8.     private final List<MessageHandler> messageHandlers;
10.     @Override
11.     public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
12.        log.info("【Paho-V3】 => MQTT消息发送成功，消息ID：{}", iMqttDeliveryToken.getMessageId());
13.     }
15.     @Override
16.     public void connectionLost(Throwable throwable) {
17.        log.error("【Paho-V3】 => MQTT关闭连接");
18.     }
20.     @Override
21.     public void messageArrived(String topic, MqttMessage message) throws Exception {
22.        for (MessageHandler messageHandler : messageHandlers) {
23.           if (messageHandler.isSubscribe(topic)) {
24.              log.info("【Paho-V3】 => MQTT接收到消息，Topic：{}", topic);
25.              messageHandler.handle(new org.laokou.sample.mqtt.handler.MqttMessage(message.getPayload(), topic));
26.           }
27.        }
28.     }
29. }

复制代码

PahoV3MqttClientTest

1. /**
2. * @author laokou
3. */
4. @SpringBootTest
5. @RequiredArgsConstructor
6. @TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
7. class PahoV3MqttClientTest {
9.     private final List<MessageHandler> messageHandlers;
11.     @Test
12.     void testMqttClient() throws InterruptedException {
13.         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(16);
15.         PahoProperties pahoProperties2 = new PahoProperties();
16.         pahoProperties2.setClientId("test-client-4");
17.         pahoProperties2.setTopics(Set.of("/test-topic-4/#"));
18.         PahoMqttClientV3 pahoMqttClientV3 = new PahoMqttClientV3(pahoProperties2, messageHandlers, scheduledExecutorService);
19.         pahoMqttClientV3.open();
20.         Thread.sleep(1000);
21.         pahoMqttClientV3.publish("/test-topic-4/000", "Hello World000".getBytes());
22.     }
24. }

复制代码

Hivemq-MQTT-Client【不推荐】

注意：订阅一段时间收不到数据，标准mqtt5.0协议，不兼容emqx broker mqtt5.0
Hivemq代码地址
引入依赖

1. <dependencies>
2.     <dependency>
3.         <groupId>com.hivemq</groupId>
4.         hivemq-mqtt-client-reactor</artifactId>
5.         <version>1.3.5</version>
6.     </dependency>
7.     <dependency>
8.         <groupId>com.hivemq</groupId>
9.         hivemq-mqtt-client-epoll</artifactId>
10.         <version>1.3.5</version>
11.         <type>pom</type>
12.     </dependency>
13. <dependencies>

复制代码

项目集成

HivemqProperties

1. /**
2. * @author laokou
3. */
4. @Data
5. public class HivemqProperties {
7.     private boolean auth = true;
9.     private String username = "emqx";
11.     private String password = "laokou123";
13.     private String host = "127.0.0.1";
15.     private int port = 1883;
17.     private String clientId;
19.     private int subscribeQos = 1;
21.     private int publishQos = 0;
23.     private int willQos = 1;
25.     // @formatter:off
26.     /**
27.      * 控制是否创建新会话（true=新建，false=复用历史会话）. clearStart=true => Broker 会在连接断开后立即清除所有会话信息.
28.      * clearStart=false => Broker 会在连接断开后保存会话信息，并在重新连接后复用会话信息.
29.      * ...
30.      */
31.     // @formatter:on
32.     private boolean clearStart = false;
34.     private int receiveMaximum = 10000;
36.     private int sendMaximum = 10000;
38.     private int maximumPacketSize = 10000;
40.     private int sendMaximumPacketSize = 10000;
42.     private int topicAliasMaximum = 1024;
44.     private int sendTopicAliasMaximum = 2048;
46.     private long messageExpiryInterval = 86400L;
48.     private boolean requestProblemInformation = true;
50.     private boolean requestResponseInformation = true;
52.     // @formatter:off
53.     /**
54.      * 默认会话保留一天.
55.      * 最大值，4294967295L，会话过期时间【永不过期，单位秒】.
56.      * 定义客户端断开后会话保留的时间（仅在 Clean Session = false 时生效）.
57.      */
58.     private long sessionExpiryInterval = 86400L;
59.     // @formatter:on
61.     /**
62.      * 心跳包每隔60秒发一次.
63.      */
64.     private int keepAliveInterval = 60;
66.     private boolean automaticReconnect = true;
68.     private long automaticReconnectMaxDelay = 5;
70.     private long automaticReconnectInitialDelay = 1;
72.     private Set<String> topics = new HashSet<>(0);
74.     private int nettyThreads = 32;
76.     private boolean retain = false;
78.     private boolean noLocal = false;
80. }

复制代码

HivemqClientV5
[code]/** * @author laokou */@Slf4jpublic class HivemqClientV5 {    /**     * 响应主题.     */    private final String RESPONSE_TOPIC = "response/topic";    /**     * 服务下线数据.     */    private final byte[] WILL_PAYLOAD = "offline".getBytes(UTF_8);    /**     * 相关数据.     */    private final byte[] CORRELATION_DATA = "correlationData".getBytes(UTF_8);    private final HivemqProperties hivemqProperties;    private final List messageHandlers;    private volatile Mqtt5RxClient client;    private final Object lock = new Object();    private volatile Disposable connectDisposable;    private volatile Disposable subscribeDisposable;    private volatile Disposable unSubscribeDisposable;    private volatile Disposable publishDisposable;    private volatile Disposable disconnectDisposable;    private volatile Disposable consumeDisposable;    public HivemqClientV5(HivemqProperties hivemqProperties, List messageHandlers) {        this.hivemqProperties = hivemqProperties;        this.messageHandlers = messageHandlers;    }    public void open() {        if (Objects.isNull(client)) {            synchronized (lock) {                if (Objects.isNull(client)) {                    client = getMqtt5ClientBuilder().buildRx();                }            }        }        connect();        consume();    }    public void close() {        if (!Objects.isNull(client)) {            disconnectDisposable = client.disconnectWith()                    .sessionExpiryInterval(hivemqProperties.getSessionExpiryInterval())                    .applyDisconnect()                    .subscribeOn(Schedulers.io())                    .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                            .takeWhile(retryCount -> retryCount != -1)                            .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                    .subscribe(() -> log.info("【Hivemq-V5】 => MQTT断开连接成功，客户端ID：{}", hivemqProperties.getClientId()),                            e -> log.error("【Hivemq-V5】 => MQTT断开连接失败，错误信息：{}", e.getMessage(), e));        }    }    public void subscribe() {        String[] topics = getTopics();        subscribe(topics, getQosArray(topics));    }    public String[] getTopics() {        return hivemqProperties.getTopics().toArray(String[]::new);    }    public int[] getQosArray(String[] topics) {        return Stream.of(topics).mapToInt(item -> hivemqProperties.getSubscribeQos()).toArray();    }    public void subscribe(String[] topics, int[] qosArray) {        checkTopicAndQos(topics, qosArray);        if (!Objects.isNull(client)) {            List subscriptions = new ArrayList(topics.length);            for (int i = 0; i < topics.length; i++) {                subscriptions.add(Mqtt5Subscription.builder()                        .topicFilter(topics)                        .qos(getMqttQos(qosArray))                        .retainAsPublished(hivemqProperties.isRetain())                        .noLocal(hivemqProperties.isNoLocal())                        .build());            }            subscribeDisposable = client.subscribeWith()                    .addSubscriptions(subscriptions)                    .applySubscribe()                    .subscribeOn(Schedulers.io())                    .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                            .takeWhile(retryCount -> retryCount != -1)                            .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                    .subscribe(ack -> log.info("【Hivemq-V5】 => MQTT订阅成功，主题: {}", String.join("、", topics)), e -> log                            .error("【Hivemq-V5】 => MQTT订阅失败，主题：{}，错误信息：{}", String.join("、", topics), e.getMessage(), e));        }    }    public void unSubscribe() {        String[] topics = hivemqProperties.getTopics().toArray(String[]::new);        unSubscribe(topics);    }    public void unSubscribe(String[] topics) {        checkTopic(topics);        if (!Objects.isNull(client)) {            List matchedTopics = new ArrayList(topics.length);            for (String topic : topics) {                matchedTopics.add(MqttTopicFilter.of(topic));            }            unSubscribeDisposable = client.unsubscribeWith()                    .addTopicFilters(matchedTopics)                    .applyUnsubscribe()                    .subscribeOn(Schedulers.io())                    .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                            .takeWhile(retryCount -> retryCount != -1)                            .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                    .subscribe(ack -> log.info("【Hivemq-V5】 => MQTT取消订阅成功，主题：{}", String.join("、", topics)), e -> log                            .error("【Hivemq-V5】 => MQTT取消订阅失败，主题：{}，错误信息：{}", String.join("、", topics), e.getMessage(), e));        }    }    public void publish(String topic, byte[] payload, int qos) {        if (!Objects.isNull(client)) {            publishDisposable = client                    .publish(Flowable.just(Mqtt5Publish.builder()                            .topic(topic)                            .qos(getMqttQos(qos))                            .payload(payload)                            .noMessageExpiry()                            .retain(hivemqProperties.isRetain())                            .messageExpiryInterval(hivemqProperties.getMessageExpiryInterval())                            .correlationData(CORRELATION_DATA)                            .payloadFormatIndicator(Mqtt5PayloadFormatIndicator.UTF_8)                            .contentType("text/plain")                            .responseTopic(RESPONSE_TOPIC)                            .build()))                    .subscribeOn(Schedulers.io())                    .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                            .takeWhile(retryCount -> retryCount != -1)                            .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                    .subscribe(ack -> log.info("【Hivemq-V5】 => MQTT消息发布成功，topic：{}", topic),                            e -> log.error("【Hivemq-V5】 => MQTT消息发布失败，topic：{}，错误信息：{}", topic, e.getMessage(), e));        }    }    public void publish(String topic, byte[] payload) {        publish(topic, payload, hivemqProperties.getPublishQos());    }    public void dispose(Disposable disposable) {        if (!Objects.isNull(disposable) && !disposable.isDisposed()) {            // 显式取消订阅            disposable.dispose();        }    }    public void dispose() {        dispose(connectDisposable);        dispose(subscribeDisposable);        dispose(unSubscribeDisposable);        dispose(publishDisposable);        dispose(consumeDisposable);        dispose(disconnectDisposable);    }    public void reSubscribe() {        log.info("【Hivemq-V5】 => MQTT重新订阅开始");        dispose(subscribeDisposable);        subscribe();        log.info("【Hivemq-V5】 => MQTT重新订阅结束");    }    private MqttQos getMqttQos(int qos) {        return MqttQos.fromCode(qos);    }    private void connect() {        connectDisposable = client.connectWith()                .keepAlive(hivemqProperties.getKeepAliveInterval())                .cleanStart(hivemqProperties.isClearStart())                .sessionExpiryInterval(hivemqProperties.getSessionExpiryInterval())                .willPublish()                .topic("will/topic")                .payload(WILL_PAYLOAD)                .qos(getMqttQos(hivemqProperties.getWillQos()))                .retain(true)                .messageExpiryInterval(100)                .delayInterval(10)                .payloadFormatIndicator(Mqtt5PayloadFormatIndicator.UTF_8)                .contentType("text/plain")                .responseTopic(RESPONSE_TOPIC)                .correlationData(CORRELATION_DATA)                .applyWillPublish()                .restrictions()                .receiveMaximum(hivemqProperties.getReceiveMaximum())                .sendMaximum(hivemqProperties.getSendMaximum())                .maximumPacketSize(hivemqProperties.getMaximumPacketSize())                .sendMaximumPacketSize(hivemqProperties.getSendMaximumPacketSize())                .topicAliasMaximum(hivemqProperties.getTopicAliasMaximum())                .sendTopicAliasMaximum(hivemqProperties.getSendTopicAliasMaximum())                .requestProblemInformation(hivemqProperties.isRequestProblemInformation())                .requestResponseInformation(hivemqProperties.isRequestResponseInformation())                .applyRestrictions()                .applyConnect()                .toFlowable()                .firstElement()                .subscribeOn(Schedulers.io())                .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                        .takeWhile(retryCount -> retryCount != -1)                        .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                .subscribe(                        ack -> log.info("【Hivemq-V5】 => MQTT连接成功，主机：{}，端口：{}，客户端ID：{}", hivemqProperties.getHost(),                                hivemqProperties.getPort(), hivemqProperties.getClientId()),                        e -> log.error("【Hivemq-V5】 => MQTT连接失败，错误信息：{}", e.getMessage(), e));    }    private void consume() {        if (!Objects.isNull(client)) {            consumeDisposable = client.publishes(MqttGlobalPublishFilter.ALL)                    .onBackpressureBuffer(8192)                    .observeOn(Schedulers.computation(), false, 8192)                    .doOnSubscribe(subscribe -> {                        log.info("【Hivemq-V5】 => MQTT开始订阅消息，请稍候。。。。。。");                        reSubscribe();                    })                    .subscribeOn(Schedulers.io())                    .retryWhen(errors -> errors.scan(1, (retryCount, error) -> retryCount > 5 ? -1 : retryCount + 1)                            .takeWhile(retryCount -> retryCount != -1)                            .flatMap(retryCount -> Flowable.timer((long) Math.pow(2, retryCount) * 100, TimeUnit.MILLISECONDS)))                    .subscribe(publish -> {                                for (MessageHandler messageHandler : messageHandlers) {                                    if (messageHandler.isSubscribe(publish.getTopic().toString())) {                                        log.info("【Hivemq-V5】 => MQTT接收到消息，Topic：{}", publish.getTopic());                                        messageHandler                                                .handle(new MqttMessage(publish.getPayloadAsBytes(), publish.getTopic().toString()));                                    }                                }                            }, e -> log.error("【Hivemq-V5】 => MQTT消息处理失败，错误信息：{}", e.getMessage(), e),                            () -> log.info("【Hivemq-V5】 => MQTT订阅消息结束，请稍候。。。。。。"));        }    }    private Mqtt5ClientBuilder getMqtt5ClientBuilder() {        Mqtt5ClientBuilder builder = Mqtt5Client.builder().addConnectedListener(listener -> {                    Optional