MQTT队列组件 (MQTT 3.1.1)

MQTT 3.1.1协议

组件作用

本组件主要提供了MQTT队列的连接驱动，用于在项目中连接MQTT队列，同时对MQTT队列的消息收发进行封装，并提供了幂等去重、消息重发、异常处理、消息监听等功能。

组件用法

引入依赖

MQTT组件配置方式如下：

首先在项目所属的Maven模块中引入组件坐标：

<dependency>
    <groupId>cn.com.rydeen.component</groupId>
    <artifactId>rydeen-component-mqtt</artifactId>
</dependency>

配置文件

MQTT组件的配置分为服务端和客户端两部分，公司产品的使用需求中暂无双向通信的需求，所以服务端通常为Java后端程序配置，客户端通常为前端程序（包括：Web、Android、iOS、Electron或其他Java程序）配置。

服务端配置

服务端配置文件application.yml中配置如下：

platform:
    # 组件配置
    component:
        # MQTT 配置
        mqtt:
            # 是否启动项目时就执行连接（true：启动时连接，false：启动时不连接）
            enable: true
            # 客户端类型（可选值：aliyun、rydeen）
            type: rydeen
            # 幂等去重服务使用的数据源类型
            datasource: redis
            # MQTT 客户端类型（server：服务端，client：客户端）
            client-type: server
            # 根主题
            parent-topic: platform
            # 分组ID
            group-id: GID-platform
            # MQTT服务器配置
            server:
                # 默认网络类型
                default-network-type: PUBLIC
                # MQTT 使用的交互协议
                protocol: tcp
                # MQTT 公网地址
                host: 10.100.0.117
                # MQTT 公网端口
                port: 11883
                # 是否清空会话（true：清空，false：保留）
                clear-session: false
                # MQTT 服务器用户名
                username: savills
                # MQTT 服务器密码
                password: Savills20191213
                # QoS 等级
                qos: 1
                # 推送消息超时时长（单位：毫秒）
                time-to-wait: 3000
                # 服务器连接超时时长（单位：毫秒）
                connection-timeout: 6000
                # ssl配置（如果无需加密，请删除该节点）
                ssl:
                  ca-cert: MIIDQTCCAimgAwIBAgITBmyfz5m/jAo54vB4ikPmljZbyjANBgkqhkiG9w0BAQsFADA5MQswCQYDVQQGEwJVUzEPMA0GA1UEChMGQW1hem9uMRkwFwYDVQQDExBBbWF6b24gUm9vdCBDQSAxMB4XDTE1MDUyNjAwMDAwMFoXDTM4MDExNzAwMDAwMFowOTELMAkGA1UEBhMCVVMxDzANBgNVBAoTBkFtYXpvbjEZMBcGA1UEAxMQQW1hem9uIFJvb3QgQ0EgMTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALJ4gHHKeNXjca9HgFB0fW7Y14h29Jlo91ghYPl0hAEvrAIthtOgQ3pOsqTQNroBvo3bSMgHFzZM9O6II8c+6zf1tRn4SWiw3te5djgdYZ6k/oI2peVKVuRF4fn9tBb6dNqcmzU5L/qwIFAGbHrQgLKm+a/sRxmPUDgH3KKHOVj4utWp+UhnMJbulHheb4mjUcAwhmahRWa6VOujw5H5SNz/0egwLX0tdHA114gk957EWW67c4cX8jJGKLhD+rcdqsq08p8kDi1L93FcXmn/6pUCyziKrlA4b9v7LWIbxcceVOF34GfID5yHI9Y/QCB/IIDEgEw+OyQmjgSubJrIqg0CAwEAAaNCMEAwDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAYYwHQYDVR0OBBYEFIQYzIU07LwMlJQuCFmcx7IQTgoIMA0GCSqGSIb3DQEBCwUAA4IBAQCY8jdaQZChGsV2USggNiMOruYou6r4lK5IpDB/G/wkjUu0yKGX9rbxenDIU5PMCCjjmCXPI6T53iHTfIUJrU6adTrCC2qJeHZERxhlbI1Bjjt/msv0tadQ1wUsN+gDS63pYaACbvXy8MWy7Vu33PqUXHeeE6V/Uq2V8viTO96LXFvKWlJbYK8U90vvo/ufQJVtMVT8QtPHRh8jrdkPSHCa2XV4cdFyQzR1bldZwgJcJmApzyMZFo6IQ6XU5MsI+yMRQ+hDKXJioaldXgjUkK642M4UwtBV8ob2xJNDd2ZhwLnoQdeXeGADbkpyrqXRfboQnoZsG4q5WTP468SQvvG5
                  client-cert: MIIDWTCCAkGgAwIBAgIUbm6vJI27xd6xz0IPB236VX/tmDgwDQYJKoZIhvcNAQELBQAwTTFLMEkGA1UECwxCQW1hem9uIFdlYiBTZXJ2aWNlcyBPPUFtYXpvbi5jb20gSW5jLiBMPVNlYXR0bGUgU1Q9V2FzaGluZ3RvbiBDPVVTMB4XDTIzMDMxNTExMTY0M1oXDTQ5MTIzMTIzNTk1OVowHjEcMBoGA1UEAwwTQVdTIElvVCBDZXJ0aWZpY2F0ZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALt5hcWCFiVIiSHUnx+TWIOvNOCI7IWuxputoudanIicwEJZfW925PIaZSJ5cmJIWeL3m70IkDfyPSnbrD8f1mprJEugsA/KGOeiRF9z3cVZFlVQDIAlAEZbAvX3C6QMfm1p0zkcWWyGiGQagrSMG2px3cKP7+nNRstZgKF5Z4FXX4h4Fhvu8rwVhpohemDtI/nZ2UDnG2bUrx0CFGPeR7U1GGLb3kW6SDE53/rQiYeIL0Z6pT/G01r3RPLW/MjMqAnglBhNhTbQEVGMGHecJRKTfUZs/PW4SRzCbzb3KOW3WKkESCpd/uGmCkJo/sMOIaY28sZQTPkl93AXbuWeWH0CAwEAAaNgMF4wHwYDVR0jBBgwFoAU5Cziw9S6BSs9gZrPKiPRvV5pq6wwHQYDVR0OBBYEFPkaAi+D7pRxaFktZ0gXMRIVWBESMAwGA1UdEwEB/wQCMAAwDgYDVR0PAQH/BAQDAgeAMA0GCSqGSIb3DQEBCwUAA4IBAQB/+UO8o2VxPDRDuwoZrprQZ1U2xGiJot/4lK9Ly15KWvgGEkr5MTPWj0e+LamOPCLBne091kkdCQGHSvHvtD7iQTqaUshAdDwh3sz5oMCz4rf4rQqQ42BJn5ruBRXNl5zaERkXHH+rfPRn8RZEG3z3n7lbqo3hu4o70ytMgI9RiDRmlaRXn3Cl1D8E8Ypc0WtqyOExZp/aW4G8/Ba4/DlOmyk8NZrjQW2tIgYY4zJ9W6BhcX2zzt5Xt3epIpz8Jy/c4th5at7Kzm0XX1gP5u/IuZt7dVi1Ufp48prP86DHqtfW/6AOjOxgF5P+ZnOwZO6f3QAFBju3w4bZv/YO0Pu7
                  client-key: MIIEpAIBAAKCAQEAu3mFxYIWJUiJIdSfH5NYg6804Ijsha7Gm62i51qciJzAQll9b3bk8hplInlyYkhZ4vebvQiQN/I9KdusPx/WamskS6CwD8oY56JEX3PdxVkWVVAMgCUARlsC9fcLpAx+bWnTORxZbIaIZBqCtIwbanHdwo/v6c1Gy1mAoXlngVdfiHgWG+7yvBWGmiF6YO0j+dnZQOcbZtSvHQIUY95HtTUYYtveRbpIMTnf+tCJh4gvRnqlP8bTWvdE8tb8yMyoCeCUGE2FNtARUYwYd5wlEpN9Rmz89bhJHMJvNvco5bdYqQRIKl3+4aYKQmj+ww4hpjbyxlBM+SX3cBdu5Z5YfQIDAQABAoIBAC1pc2jzaO/hES67tUvNwlwrnbBiGoELZBqjxdkqzs4cwkmQKd1NPIoUsLjYGEgMZJGhDv2MTHrw17YC+0dTK/odN+6F3wAEIWJCoTNJ+sdqHAnQWSbVT6q7Mw1nqU80o49XWg30ewjZ0cIF1fFsGtv+9wvBGAlXazWo2YrDwLnIiSYOZNOtSrFwnnngM8N89OLTbrUQZC5F3opIbOV6lZY6aLqgzSlOaAKOb0XEjbhtaaVL7OV0roH/wScESB5pWA9BqEx+yKkU4sek5Ij9hMd4ng8vfNnMzdTHhrw2yra+VSDAuqN1hMK5RBchgcgEjyUZrQmXT36NLiefJ0IZmxUCgYEA5IZ8Vxo/DXkuhAeBTMcbjnFhxbK1CEHWBMtL4pjgShv5ZfquPe9IrHQSQCi7cXUNkJpaB8GdzAHUd6KCF1e4EXqDtXkInRhRDnWXODmStcZw0A9Pu10XffJl1b79oCeE6Y8WtT/Pf0GSuThjqItWu0eLxwRv9V7oEgkZy8brdysCgYEA0gOWVOmgDcpOpzHoQD9bPuAkEUow1EIyuRaP6JUfpP1DQrvhMUrbLhlS4H+qUOaEdJq337Mwdp9kg1C6T3V5sMFLGx/XHlY8H3Eo0JodiA0LUVEFGVMx9zZWajlEtvOOoNk7ngpGHx+djS7DKDvnVv243BMUyFNK7j4XHkD8GvcCgYB1vay2m/k8keEbeVpvGm7iNxyu5WJMfM8t7SiN+UImLL+UgVHg6zZAVBVLJTFeRXbVuBEGygTpJVA4y2vHZxSZPMw7Zb7AYdTxZDijaVZjfXiGWj00RcRiyc8aD6tRJsteIrEKdXQNFtXnxFvkbNYk67nisjCU9W8QS8wUp8qNvQKBgQCZpLo0wP6xWBKWyzJI3pEYvM5mP8rdXtjRBCqgU4jlIcw1ZFJpx4P2g9xYFc46hYrd69sds3B/ODlGwKatPxqaL6+M1oyAy0INFrJvenCkejaPdSCv2GeznnM2H1OgLStpjcPJ6WZHms+e3Fk2dCzZPaIij2si65zkE1TCc3R9HQKBgQDDIxyyzsKKB8ZpiWsYIUnoGMc/upuncz7ySpicL1xG7j10Wdj6MryMpll/EpNMqhNgtYDbArCTMP9g4gwzU1cpupo9GyGj35qxN+sbK3qI2DmKVeoS82GWaSXQExQZHFibc5g3jvVDPDJ8rR8Mr/2Qya2MXCgji7YXBrI55TK7jw==

客户端配置

客户端配置文件application.yml中配置如下：

platform:
   # 组件配置
    component:
        # MQTT 配置
        mqtt:
            # 是否启用MQTT客户端
            enable: true
            # 是否在启动时就执行连接（true：启动时连接，false：启动时不连接）
            init-client: true
            # 客户端类型（可选值：ALIYUN、RYDEEN）
            type: rydeen
            # 幂等去重服务使用的数据源类型
            datasource: memory
            # MQTT 客户端类型（SERVER：服务端，CLIENT：客户端）
            client-type: client
            # 根主题
            parent-topic: richie696
            # 分组ID
            group-id: GID_test
            # 服务器配置
            server:
                # 默认网络类型
                default-network-type: PUBLIC
                # MQTT 使用的交互协议
                protocol: tcp
                # MQTT 公网地址
                host: 10.100.0.117
                # MQTT 公网端口
                port: 11883
                # MQTT 内网地址
                vpc-host: 10.100.0.117
                # MQTT 内网端口
                vpc-port: 11883
                # 是否清空会话（true：清空，false：保留）
                clear-session: false
                # MQTT 服务器用户名
                username: savills
                # MQTT 服务器密码
                password: Savills20191213
                # QoS 等级
                qos: 1
                # 推送消息超时时长（单位：毫秒）
                time-to-wait: 5000
                # 服务器连接超时时长（单位：毫秒）
                connection-timeout: 20000
                # ssl配置（如果无需加密，请删除该节点）
                ssl:
                    ca-cert: MIIDQTCCAimgAwIBAgITBmyfz5m/jAo54vB4ikPmljZbyjANBgkqhkiG9w0BAQsFADA5MQswCQYDVQQGEwJVUzEPMA0GA1UEChMGQW1hem9uMRkwFwYDVQQDExBBbWF6b24gUm9vdCBDQSAxMB4XDTE1MDUyNjAwMDAwMFoXDTM4MDExNzAwMDAwMFowOTELMAkGA1UEBhMCVVMxDzANBgNVBAoTBkFtYXpvbjEZMBcGA1UEAxMQQW1hem9uIFJvb3QgQ0EgMTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALJ4gHHKeNXjca9HgFB0fW7Y14h29Jlo91ghYPl0hAEvrAIthtOgQ3pOsqTQNroBvo3bSMgHFzZM9O6II8c+6zf1tRn4SWiw3te5djgdYZ6k/oI2peVKVuRF4fn9tBb6dNqcmzU5L/qwIFAGbHrQgLKm+a/sRxmPUDgH3KKHOVj4utWp+UhnMJbulHheb4mjUcAwhmahRWa6VOujw5H5SNz/0egwLX0tdHA114gk957EWW67c4cX8jJGKLhD+rcdqsq08p8kDi1L93FcXmn/6pUCyziKrlA4b9v7LWIbxcceVOF34GfID5yHI9Y/QCB/IIDEgEw+OyQmjgSubJrIqg0CAwEAAaNCMEAwDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAYYwHQYDVR0OBBYEFIQYzIU07LwMlJQuCFmcx7IQTgoIMA0GCSqGSIb3DQEBCwUAA4IBAQCY8jdaQZChGsV2USggNiMOruYou6r4lK5IpDB/G/wkjUu0yKGX9rbxenDIU5PMCCjjmCXPI6T53iHTfIUJrU6adTrCC2qJeHZERxhlbI1Bjjt/msv0tadQ1wUsN+gDS63pYaACbvXy8MWy7Vu33PqUXHeeE6V/Uq2V8viTO96LXFvKWlJbYK8U90vvo/ufQJVtMVT8QtPHRh8jrdkPSHCa2XV4cdFyQzR1bldZwgJcJmApzyMZFo6IQ6XU5MsI+yMRQ+hDKXJioaldXgjUkK642M4UwtBV8ob2xJNDd2ZhwLnoQdeXeGADbkpyrqXRfboQnoZsG4q5WTP468SQvvG5
                    client-cert: MIIDWTCCAkGgAwIBAgIUbm6vJI27xd6xz0IPB236VX/tmDgwDQYJKoZIhvcNAQELBQAwTTFLMEkGA1UECwxCQW1hem9uIFdlYiBTZXJ2aWNlcyBPPUFtYXpvbi5jb20gSW5jLiBMPVNlYXR0bGUgU1Q9V2FzaGluZ3RvbiBDPVVTMB4XDTIzMDMxNTExMTY0M1oXDTQ5MTIzMTIzNTk1OVowHjEcMBoGA1UEAwwTQVdTIElvVCBDZXJ0aWZpY2F0ZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALt5hcWCFiVIiSHUnx+TWIOvNOCI7IWuxputoudanIicwEJZfW925PIaZSJ5cmJIWeL3m70IkDfyPSnbrD8f1mprJEugsA/KGOeiRF9z3cVZFlVQDIAlAEZbAvX3C6QMfm1p0zkcWWyGiGQagrSMG2px3cKP7+nNRstZgKF5Z4FXX4h4Fhvu8rwVhpohemDtI/nZ2UDnG2bUrx0CFGPeR7U1GGLb3kW6SDE53/rQiYeIL0Z6pT/G01r3RPLW/MjMqAnglBhNhTbQEVGMGHecJRKTfUZs/PW4SRzCbzb3KOW3WKkESCpd/uGmCkJo/sMOIaY28sZQTPkl93AXbuWeWH0CAwEAAaNgMF4wHwYDVR0jBBgwFoAU5Cziw9S6BSs9gZrPKiPRvV5pq6wwHQYDVR0OBBYEFPkaAi+D7pRxaFktZ0gXMRIVWBESMAwGA1UdEwEB/wQCMAAwDgYDVR0PAQH/BAQDAgeAMA0GCSqGSIb3DQEBCwUAA4IBAQB/+UO8o2VxPDRDuwoZrprQZ1U2xGiJot/4lK9Ly15KWvgGEkr5MTPWj0e+LamOPCLBne091kkdCQGHSvHvtD7iQTqaUshAdDwh3sz5oMCz4rf4rQqQ42BJn5ruBRXNl5zaERkXHH+rfPRn8RZEG3z3n7lbqo3hu4o70ytMgI9RiDRmlaRXn3Cl1D8E8Ypc0WtqyOExZp/aW4G8/Ba4/DlOmyk8NZrjQW2tIgYY4zJ9W6BhcX2zzt5Xt3epIpz8Jy/c4th5at7Kzm0XX1gP5u/IuZt7dVi1Ufp48prP86DHqtfW/6AOjOxgF5P+ZnOwZO6f3QAFBju3w4bZv/YO0Pu7
                    client-key: MIIEpAIBAAKCAQEAu3mFxYIWJUiJIdSfH5NYg6804Ijsha7Gm62i51qciJzAQll9b3bk8hplInlyYkhZ4vebvQiQN/I9KdusPx/WamskS6CwD8oY56JEX3PdxVkWVVAMgCUARlsC9fcLpAx+bWnTORxZbIaIZBqCtIwbanHdwo/v6c1Gy1mAoXlngVdfiHgWG+7yvBWGmiF6YO0j+dnZQOcbZtSvHQIUY95HtTUYYtveRbpIMTnf+tCJh4gvRnqlP8bTWvdE8tb8yMyoCeCUGE2FNtARUYwYd5wlEpN9Rmz89bhJHMJvNvco5bdYqQRIKl3+4aYKQmj+ww4hpjbyxlBM+SX3cBdu5Z5YfQIDAQABAoIBAC1pc2jzaO/hES67tUvNwlwrnbBiGoELZBqjxdkqzs4cwkmQKd1NPIoUsLjYGEgMZJGhDv2MTHrw17YC+0dTK/odN+6F3wAEIWJCoTNJ+sdqHAnQWSbVT6q7Mw1nqU80o49XWg30ewjZ0cIF1fFsGtv+9wvBGAlXazWo2YrDwLnIiSYOZNOtSrFwnnngM8N89OLTbrUQZC5F3opIbOV6lZY6aLqgzSlOaAKOb0XEjbhtaaVL7OV0roH/wScESB5pWA9BqEx+yKkU4sek5Ij9hMd4ng8vfNnMzdTHhrw2yra+VSDAuqN1hMK5RBchgcgEjyUZrQmXT36NLiefJ0IZmxUCgYEA5IZ8Vxo/DXkuhAeBTMcbjnFhxbK1CEHWBMtL4pjgShv5ZfquPe9IrHQSQCi7cXUNkJpaB8GdzAHUd6KCF1e4EXqDtXkInRhRDnWXODmStcZw0A9Pu10XffJl1b79oCeE6Y8WtT/Pf0GSuThjqItWu0eLxwRv9V7oEgkZy8brdysCgYEA0gOWVOmgDcpOpzHoQD9bPuAkEUow1EIyuRaP6JUfpP1DQrvhMUrbLhlS4H+qUOaEdJq337Mwdp9kg1C6T3V5sMFLGx/XHlY8H3Eo0JodiA0LUVEFGVMx9zZWajlEtvOOoNk7ngpGHx+djS7DKDvnVv243BMUyFNK7j4XHkD8GvcCgYB1vay2m/k8keEbeVpvGm7iNxyu5WJMfM8t7SiN+UImLL+UgVHg6zZAVBVLJTFeRXbVuBEGygTpJVA4y2vHZxSZPMw7Zb7AYdTxZDijaVZjfXiGWj00RcRiyc8aD6tRJsteIrEKdXQNFtXnxFvkbNYk67nisjCU9W8QS8wUp8qNvQKBgQCZpLo0wP6xWBKWyzJI3pEYvM5mP8rdXtjRBCqgU4jlIcw1ZFJpx4P2g9xYFc46hYrd69sds3B/ODlGwKatPxqaL6+M1oyAy0INFrJvenCkejaPdSCv2GeznnM2H1OgLStpjcPJ6WZHms+e3Fk2dCzZPaIij2si65zkE1TCc3R9HQKBgQDDIxyyzsKKB8ZpiWsYIUnoGMc/upuncz7ySpicL1xG7j10Wdj6MryMpll/EpNMqhNgtYDbArCTMP9g4gwzU1cpupo9GyGj35qxN+sbK3qI2DmKVeoS82GWaSXQExQZHFibc5g3jvVDPDJ8rR8Mr/2Qya2MXCgji7YXBrI55TK7jw==

配置详解

服务端配置主要提供连接和消息下发的能力，当然他也具备接收消息的能力，只是从业务角度将其定义为消息生产者的服务端，配置中的：

• enable：是否启动项目时就执行连接（true：启动时连接，false：启动时不连接）。
• type：客户端类型，可选值为aliyun、rydeen，默认为rydeen。rydeen可连接任意自建的MQTT服务器，aliyun仅可连接阿里云的MQTT服务器。
• datasource：幂等去重服务使用的数据源类型，可选值为memory、redis，默认为memory。memory为内存数据源，通常为物联网设备客户端配置，仅在当前实例内存中进行消息去重，redis为Redis数据源，可用于分布式场景。
• client-type：MQTT 客户端类型，可选值为server、client，默认为server。server为服务端，client为客户端。
• parent-topic：为根主题，通常是云PaaS的MQTT服务器强制要求配置的根主题，全局唯一。
• group-id：为分组ID，为连接的客户端进行分组，通常是云PaaS的MQTT服务器强制要求配置的分组ID。
• server：节点中的配置为MQTT服务器的配置，
• server - default-network-type：默认网络类型，可选值为PUBLIC、VPC，默认为PUBLIC。PUBLIC为公网，VPC为内网。本客户端支持内外网手工切换，通过MQTT客户端暴露的接口进行手工切换（自动切换，可能会带来位置问题，所以切换的行为转交给业务端自行决定）
• server - protocol：MQTT 使用的交互协议，可选值为tcp、ssl，默认为tcp。
• server - host：MQTT 公网地址。
• server - port：MQTT 公网端口。
• server - vpc-host：MQTT 内网地址。
• server - vpc-port：MQTT 内网端口。
• server - clear-session：是否清空会话，可选值为true、false，默认为false。此值需要注意，如果设值为false，当客户端连接断开时，Topic中的消息会全部丢失，被MQTT服务器自动删除，所以在不可丢失消息的场景中，该值一定要为false，否则会造成数据丢失的问题，切记！
• server - username：MQTT 服务器用户名。
• server - password：MQTT 服务器密码。
• server - qos：QoS 等级，可选值为0、1、2，默认为1。
  • QoS = 0：表示消息最多只会发送1次(0次或1次)。消息不会被持久保存到磁盘，也 不会通过 ACK 确认，这种 QoS 的速度是最快的，但是应该只用于没有价值的消息。
  • QoS = 1：表示消息至少会发送1次(>=1)。消息只有在能够持久保存的情况下才 能安全地传递，因此应用程序必须使用 MqttConnectOptions 提供一种持久方法。 如果没有指定持久化机制，则在客户端失败时不会传递消息（QoS 默认值）。
  • QoS = 2：表示消息只会传递一次。消息将被持久化到磁盘，并通过网络进行两 阶段确认。消息只有在能够持久保存的情况下才能安全地传递，因此应用程序必须 使用 MqttConnectOptions 提供一种持久方法。如果没有指定持久性机制，则在客 户端失败时不会传递消息。
  • 重点提示：如果没有配置持久化，当出现网络或服务器问题时，仍然会传递 QoS 1 和 2 消息， 因为客户机将在内存中保存状态。如果MQTT客户端关闭或失败，且没有配置持久化， 则无法维护 QoS 1 和 2 消息的传递，因为客户端状态将丢失。
• server - time-to-wait：推送消息超时时长（单位：毫秒），默认为3000。
• server - connection-timeout：服务器连接超时时长（单位：毫秒），默认为6000。
• server - ssl：节点为SSL配置，使用CA证书，如果无需加密，请删除该节点。

MqttClientApi接口详解

MQTT客户端提供了一系列的接口供业务端调用，接口如下：

package cn.com.rydeen.component.mqtt.client;

import cn.com.rydeen.component.mqtt.client.beans.ConsumerListener;
import cn.com.rydeen.component.mqtt.client.beans.ConsumerMessage;
import cn.com.rydeen.component.mqtt.client.beans.MqttServerInfo;
import cn.com.rydeen.component.mqtt.enums.NetworkTypeEnum;
import org.springframework.lang.NonNull;

import java.util.Set;
import java.util.function.Consumer;

/**
 * MQTT客户端接口
 *
 * @author richie696
 * @version 1.0
 * @since 2022-09-09 14:48:37
 */
public interface MqttClientApi {

    /**
     * 下发消息的方法
     * （注：本接口不会保留最新下发的消息内容，即：retained = false）
     *
     * @param topic 下发消息的主题
     * @param value 下发的内容
     */
    void doPublish(String topic, byte[] value);

    /**
     * 下发消息的方法
     *
     * @param topic    下发消息的主题
     * @param value    下发的内容
     * @param retained 是否在服务器中保留本条消息（注：服务器只会保留最新下发的1条消息在内存或磁盘文件中，当有新消息进行发送时会被最新的消息覆盖）
     */
    void doPublish(String topic, byte[] value, boolean retained);

    /**
     * 获取当前 MQTT 客户端 ID 的方法
     *
     * @return 返回 MQTT 客户端 ID
     */
    String getClientId();

    /**
     * 获取分组ID的方法
     *
     * @return 返回分组ID
     */
    String getGroupId();

    /**
     * 获取根主题的方法
     * @return 返回根主题
     */
    String getParentTopic();

    /**
     * 注册消费者监听事件的方法
     *
     * @param topic    绑定的主题
     * @param callback 消费者监听事件
     */
    void registerConsumer(@NonNull String topic, @NonNull Consumer<ConsumerMessage> callback);

    /**
     * 解绑对应主题监听事件的方法
     *
     * @param topic 需要解绑的主题
     */
    void unregisterConsumer(@NonNull String topic);

    /**
     * 批量注册消费者监听事件的方法
     *
     * @param listeners 订阅的主题监听列表
     */
    void registerConsumers(@NonNull Set<ConsumerListener> listeners);

    /**
     * 批量解绑对应主题监听事件的方法
     *
     * @param topics 解绑的主题列表
     */
    void unregisterConsumers(@NonNull String... topics);

    /**
     * 批量解绑对应主题监听事件的方法
     *
     * @param topics 解绑的主题集合
     */
    void unregisterConsumers(@NonNull Set<String> topics);

    /**
     * 初始化MQTT客户端的方法
     *
     * @param serverInfo MQTT 服务器信息
     */
    void initialClient(@NonNull MqttServerInfo serverInfo);

    /**
     * 初始化MQTT客户端的方法
     *
     * @param serverInfo MQTT 服务器信息
     * @param enable 是否启动MQTT
     */
    void initialClient(@NonNull MqttServerInfo serverInfo, boolean enable);

    /**
     * 更改连接的MQTT服务器地址的方法
     *
     * @param networkType 网络类型
     * @param host        自定义主机地址
     * @param port        自定义主机端口
     */
    void changeServer(@NonNull NetworkTypeEnum networkType, @NonNull String host, int port);

    /**
     * 切换已有公网或VPC内网服务器地址的方法
     *
     * @param networkType 网络类型
     */
    void changeServer(@NonNull NetworkTypeEnum networkType);

    /**
     * 断开连接的方法
     */
    void disconnect();

    /**
     * 销毁当前 MQTT 客户端的方法
     */
    void destroy();

    /**
     * 重新初始化新的 MQTT 客户端的方法
     * （注意：本方法仅用于客户端无法连接一类的异常时，
     * 结合业务场景手工重新初始化连接的情况下使用，其他时间调用可能会造成
     * 数据丢失的问题，使用时请小心。）
     */
    void reinitialize();
}

接口的使用，在上面的接口展示中已经有了详细的注释，这里就不再赘述了。此处仅对void doPublish(String topic, byte[] value, boolean retained);方法再做一下补充说明，此方法中的第三个参数boolean retained的含义是是否保留消息，其作用如下：

• 当 retained = true 时，当前发布的消息会被消息队列引擎保留下来，每个 Topic 只会保留 1 条消息，并且与 QoS 配置无关。 （补充说明： 该功能在物联网行业使用比较广泛，比如智能硬件与 MQTT 服务器连接，当我们 打开手机 App 时，绑定在App 账户内的所有硬件设备的 Topic 都会主动推送对应 硬件的联网状态，我们在手机 App 中就可以看到对应物联网设备是在线状态，还 是离线状态了。）
• 当 retained = false 时，为普通消息，消息是否保留受到 QoS 配置的控制。
• 清空保留消息：发送1条 retained = true 并提供一个空字节数组作为 payload 消息（例如: new byte[0]），消息队列引擎将从服务器清除该保留消息。

使用示例

这里假设MQTT组件的配置文件您已经正确的编写，MQTT服务端的使用示例如下：

服务端示例

package cn.com.rydeen.component.template.mqtt.controller;

import cn.com.rydeen.base.common.api.vo.ResultVO;
import cn.com.rydeen.component.mqtt.client.MqttClientApi;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.*;

import java.nio.charset.StandardCharsets;

/**
 * MQTT功能测试接口
 *
 * @author richie696
 * @version 1.0
 * @since 2022-10-02 10:24:08
 */
@Tag(name = "MQTT功能测试接口")
@RestController
@RequestMapping("/mqtt")
@RequiredArgsConstructor
public class MqttDemoController {

    private final MqttClientApi mqttClient;

    @Operation(summary = "下发MQTT普通消息的接口", method = "GET", description = "指定消息体内容，通过本接口给所有队列订阅方发下消息的接口。")
    @RequestMapping("/send")
    public ResultVO<Void> doSend(
            @Parameter(name = "要发送的Topic主题", required = true) @RequestParam("topic") String topic,
            @Parameter(name = "要发送的消息内容", required = true) @RequestParam("content") String content) {
        mqttClient.doPublish(topic, content.getBytes(StandardCharsets.UTF_8));
        return ResultVO.getSuccess("发送成功", null);
    }
}

自动注册的客户端示例

MQTT客户端的使用示例如下：

注册方式一：通过@PostConstruct注解的方式注册消费者

package cn.com.rydeen.component.template.mqtt.consumer;

import cn.com.rydeen.component.mqtt.client.MqttClientApi;
import cn.com.rydeen.component.mqtt.client.beans.ConsumerMessage;
import cn.com.rydeen.component.template.mqtt.service.MqttDemoService;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

/**
 * MQTT消费者类
 * 要想使用消费者，需要引入mqtt-client的配置文件，执行Client初始化， 推荐使用2个IDEA进行模拟，一个引入mqtt-server，另一个引入mqtt-client。
 *
 * @author richie696
 * @version 1.0
 * @since 2022-10-02 10:31:59
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MqttDemoConsumer implements Consumer<ConsumerMessage> {

    private final MqttClientApi mqttClient;

    private final MqttDemoService mqttDemoService;

    @PostConstruct
    public void initial() {
        // 需要绑定监听器的 topic
        String topic = STR."\{mqttClient.getParentTopic()}/9996/object";
        // 执行消费者注册（即：将 topic 绑定回调事件）
        mqttClient.registerConsumer(topic, this);
    }

    @Override
    public void accept(ConsumerMessage message) {
        // 收到消息时执行该回调进行处理
        mqttDemoService.handleMessage(message);
    }

    public void destroy() {
        // 取消注册消费者
        mqttClient.unregisterConsumers(
                STR."\{mqttClient.getParentTopic()}/9996/00FFBE87C719",
                STR."\{mqttClient.getParentTopic()}/9996/object"
        );
    }

}

注册方式二：通过自定义回调函数的方式注册消费者

package cn.com.rydeen.component.template.mqtt.consumer;

import cn.com.rydeen.component.mqtt.client.MqttClientApi;
import cn.com.rydeen.component.template.mqtt.service.MqttDemoService;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

/**
 * MQTT消费者类
 * <p>要想使用消费者，需要引入mqtt-client的配置文件，执行Client初始化， 推荐使用2个IDEA进行模拟，一个引入mqtt-server，另一个引入mqtt-client。</p>
 *
 * @author richie696
 * @version 1.0
 * @since 2022-10-02 10:31:59
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MqttDemoConsumer {

    private final MqttClientApi mqttClient;

    private final MqttDemoService mqttDemoService;

    @PostConstruct
    public void initial() {
        // 第二种注册方式
        innerRegister();
    }

    private void innerRegister() {
        // 需要绑定监听器的 topic
        String topic = STR."\{mqttClient.getParentTopic()}/9996/00FFBE87C719";
        // 执行消费者注册（即：将 topic 绑定回调事件）
        mqttClient.registerConsumer(topic, message -> {
            log.info("\n\n\n\n=========【MQTT_Consumer】===========");
            log.info("读取到消息：{}\n\n\n\n", message.toString());
        });
    }

    public void destroy() {
        // 取消注册消费者
        mqttClient.unregisterConsumers(
                STR."\{mqttClient.getParentTopic()}/9996/00FFBE87C719",
                STR."\{mqttClient.getParentTopic()}/9996/object"
        );
    }

}

手工注册的客户端示例

当MQTT客户端的连接信息需要从服务端动态获取时，可以使用手工注册的方式进行注册，示例如下：

配置文件中的配置：

platform:
   # 组件配置
    component:
        # MQTT 配置
        mqtt:
            # 是否启用MQTT客户端
            enable: false
            # 是否在启动时就执行连接（true：启动时连接，false：启动时不连接）
            init-client: false
            # 其它配置省略
            # ...

Java代码：

package cn.com.rydeen.component;

import cn.com.rydeen.component.mqtt.client.MqttClientApi;
import cn.com.rydeen.component.mqtt.client.beans.MqttServerInfo;
import cn.com.rydeen.component.mqtt.config.ServerInfo;
import cn.com.rydeen.component.mqtt.enums.NetworkTypeEnum;
import cn.com.rydeen.component.mqtt.enums.ServerTypeEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.dynamictp.core.spring.EnableDynamicTp;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.scheduling.annotation.EnableScheduling;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

/**
 * DEMO程序
 *
 * @author richie696
 * @version 1.0
 * @since 2022-09-09 17:16:58
 */
@Slf4j
@SpringBootApplication
@RequiredArgsConstructor
public class DemoApplication implements CommandLineRunner {

    private final MqttClientApi mqttClient;

    public static void main(String[] args) throws UnknownHostException {
        SpringApplication.run(DemoApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        initMqttClient();
    }


    private void initMqttClient() {
        mqttClient.initialClient(MqttServerInfo.builder()
                .networkType(NetworkTypeEnum.PUBLIC)
                .protocol("ssl")
                .host("ax8xtbyesmjaw-ats.iot.cn-northwest-1.amazonaws.com.cn")
                .port(8883)
                .vpcHost("ax8xtbyesmjaw-ats.iot.cn-northwest-1.amazonaws.com.cn")
                .vpcPort(8883)
                .username("")
                .password("")
                .deviceId("TEST236263434")
                .groupId("GID_KDS")
                .parentTopic("KDS_V1")
                .serverType(ServerTypeEnum.RYDEEN)
                .ssl(new ServerInfo.Ssl()
                        .setCaCert("MIIDQTCCAimgAwIBAgITBmyfz5m/jAo54vB4ikPmljZbyjANBgkqhkiG9w0BAQsFADA5MQswCQYDVQQGEwJVUzEPMA0GA1UEChMGQW1hem9uMRkwFwYDVQQDExBBbWF6b24gUm9vdCBDQSAxMB4XDTE1MDUyNjAwMDAwMFoXDTM4MDExNzAwMDAwMFowOTELMAkGA1UEBhMCVVMxDzANBgNVBAoTBkFtYXpvbjEZMBcGA1UEAxMQQW1hem9uIFJvb3QgQ0EgMTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALJ4gHHKeNXjca9HgFB0fW7Y14h29Jlo91ghYPl0hAEvrAIthtOgQ3pOsqTQNroBvo3bSMgHFzZM9O6II8c+6zf1tRn4SWiw3te5djgdYZ6k/oI2peVKVuRF4fn9tBb6dNqcmzU5L/qwIFAGbHrQgLKm+a/sRxmPUDgH3KKHOVj4utWp+UhnMJbulHheb4mjUcAwhmahRWa6VOujw5H5SNz/0egwLX0tdHA114gk957EWW67c4cX8jJGKLhD+rcdqsq08p8kDi1L93FcXmn/6pUCyziKrlA4b9v7LWIbxcceVOF34GfID5yHI9Y/QCB/IIDEgEw+OyQmjgSubJrIqg0CAwEAAaNCMEAwDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAYYwHQYDVR0OBBYEFIQYzIU07LwMlJQuCFmcx7IQTgoIMA0GCSqGSIb3DQEBCwUAA4IBAQCY8jdaQZChGsV2USggNiMOruYou6r4lK5IpDB/G/wkjUu0yKGX9rbxenDIU5PMCCjjmCXPI6T53iHTfIUJrU6adTrCC2qJeHZERxhlbI1Bjjt/msv0tadQ1wUsN+gDS63pYaACbvXy8MWy7Vu33PqUXHeeE6V/Uq2V8viTO96LXFvKWlJbYK8U90vvo/ufQJVtMVT8QtPHRh8jrdkPSHCa2XV4cdFyQzR1bldZwgJcJmApzyMZFo6IQ6XU5MsI+yMRQ+hDKXJioaldXgjUkK642M4UwtBV8ob2xJNDd2ZhwLnoQdeXeGADbkpyrqXRfboQnoZsG4q5WTP468SQvvG5")
                        .setClientCert("MIIDWTCCAkGgAwIBAgIUbm6vJI27xd6xz0IPB236VX/tmDgwDQYJKoZIhvcNAQELBQAwTTFLMEkGA1UECwxCQW1hem9uIFdlYiBTZXJ2aWNlcyBPPUFtYXpvbi5jb20gSW5jLiBMPVNlYXR0bGUgU1Q9V2FzaGluZ3RvbiBDPVVTMB4XDTIzMDMxNTExMTY0M1oXDTQ5MTIzMTIzNTk1OVowHjEcMBoGA1UEAwwTQVdTIElvVCBDZXJ0aWZpY2F0ZTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALt5hcWCFiVIiSHUnx+TWIOvNOCI7IWuxputoudanIicwEJZfW925PIaZSJ5cmJIWeL3m70IkDfyPSnbrD8f1mprJEugsA/KGOeiRF9z3cVZFlVQDIAlAEZbAvX3C6QMfm1p0zkcWWyGiGQagrSMG2px3cKP7+nNRstZgKF5Z4FXX4h4Fhvu8rwVhpohemDtI/nZ2UDnG2bUrx0CFGPeR7U1GGLb3kW6SDE53/rQiYeIL0Z6pT/G01r3RPLW/MjMqAnglBhNhTbQEVGMGHecJRKTfUZs/PW4SRzCbzb3KOW3WKkESCpd/uGmCkJo/sMOIaY28sZQTPkl93AXbuWeWH0CAwEAAaNgMF4wHwYDVR0jBBgwFoAU5Cziw9S6BSs9gZrPKiPRvV5pq6wwHQYDVR0OBBYEFPkaAi+D7pRxaFktZ0gXMRIVWBESMAwGA1UdEwEB/wQCMAAwDgYDVR0PAQH/BAQDAgeAMA0GCSqGSIb3DQEBCwUAA4IBAQB/+UO8o2VxPDRDuwoZrprQZ1U2xGiJot/4lK9Ly15KWvgGEkr5MTPWj0e+LamOPCLBne091kkdCQGHSvHvtD7iQTqaUshAdDwh3sz5oMCz4rf4rQqQ42BJn5ruBRXNl5zaERkXHH+rfPRn8RZEG3z3n7lbqo3hu4o70ytMgI9RiDRmlaRXn3Cl1D8E8Ypc0WtqyOExZp/aW4G8/Ba4/DlOmyk8NZrjQW2tIgYY4zJ9W6BhcX2zzt5Xt3epIpz8Jy/c4th5at7Kzm0XX1gP5u/IuZt7dVi1Ufp48prP86DHqtfW/6AOjOxgF5P+ZnOwZO6f3QAFBju3w4bZv/YO0Pu7")
                        .setClientKey("MIIEpAIBAAKCAQEAu3mFxYIWJUiJIdSfH5NYg6804Ijsha7Gm62i51qciJzAQll9b3bk8hplInlyYkhZ4vebvQiQN/I9KdusPx/WamskS6CwD8oY56JEX3PdxVkWVVAMgCUARlsC9fcLpAx+bWnTORxZbIaIZBqCtIwbanHdwo/v6c1Gy1mAoXlngVdfiHgWG+7yvBWGmiF6YO0j+dnZQOcbZtSvHQIUY95HtTUYYtveRbpIMTnf+tCJh4gvRnqlP8bTWvdE8tb8yMyoCeCUGE2FNtARUYwYd5wlEpN9Rmz89bhJHMJvNvco5bdYqQRIKl3+4aYKQmj+ww4hpjbyxlBM+SX3cBdu5Z5YfQIDAQABAoIBAC1pc2jzaO/hES67tUvNwlwrnbBiGoELZBqjxdkqzs4cwkmQKd1NPIoUsLjYGEgMZJGhDv2MTHrw17YC+0dTK/odN+6F3wAEIWJCoTNJ+sdqHAnQWSbVT6q7Mw1nqU80o49XWg30ewjZ0cIF1fFsGtv+9wvBGAlXazWo2YrDwLnIiSYOZNOtSrFwnnngM8N89OLTbrUQZC5F3opIbOV6lZY6aLqgzSlOaAKOb0XEjbhtaaVL7OV0roH/wScESB5pWA9BqEx+yKkU4sek5Ij9hMd4ng8vfNnMzdTHhrw2yra+VSDAuqN1hMK5RBchgcgEjyUZrQmXT36NLiefJ0IZmxUCgYEA5IZ8Vxo/DXkuhAeBTMcbjnFhxbK1CEHWBMtL4pjgShv5ZfquPe9IrHQSQCi7cXUNkJpaB8GdzAHUd6KCF1e4EXqDtXkInRhRDnWXODmStcZw0A9Pu10XffJl1b79oCeE6Y8WtT/Pf0GSuThjqItWu0eLxwRv9V7oEgkZy8brdysCgYEA0gOWVOmgDcpOpzHoQD9bPuAkEUow1EIyuRaP6JUfpP1DQrvhMUrbLhlS4H+qUOaEdJq337Mwdp9kg1C6T3V5sMFLGx/XHlY8H3Eo0JodiA0LUVEFGVMx9zZWajlEtvOOoNk7ngpGHx+djS7DKDvnVv243BMUyFNK7j4XHkD8GvcCgYB1vay2m/k8keEbeVpvGm7iNxyu5WJMfM8t7SiN+UImLL+UgVHg6zZAVBVLJTFeRXbVuBEGygTpJVA4y2vHZxSZPMw7Zb7AYdTxZDijaVZjfXiGWj00RcRiyc8aD6tRJsteIrEKdXQNFtXnxFvkbNYk67nisjCU9W8QS8wUp8qNvQKBgQCZpLo0wP6xWBKWyzJI3pEYvM5mP8rdXtjRBCqgU4jlIcw1ZFJpx4P2g9xYFc46hYrd69sds3B/ODlGwKatPxqaL6+M1oyAy0INFrJvenCkejaPdSCv2GeznnM2H1OgLStpjcPJ6WZHms+e3Fk2dCzZPaIij2si65zkE1TCc3R9HQKBgQDDIxyyzsKKB8ZpiWsYIUnoGMc/upuncz7ySpicL1xG7j10Wdj6MryMpll/EpNMqhNgtYDbArCTMP9g4gwzU1cpupo9GyGj35qxN+sbK3qI2DmKVeoS82GWaSXQExQZHFibc5g3jvVDPDJ8rR8Mr/2Qya2MXCgji7YXBrI55TK7jw==")
                )
                .build(), true);
        mqttClient.registerConsumer("KDS_V1/210137/00FFBE87C719", message -> {
            log.debug("IMqttClientMessageCallback.callback e = {}", message);
        });
    }
}

ConsumerMessage 详解

MQTT发送的消息对象是组件封装的cn.com.rydeen.component.mqtt.client.beans.ConsumerMessage对象，该对象中包含了消息的主题、消息内容、消息的QoS等信息，该对象的定义如下：

package cn.com.rydeen.component.mqtt.client.beans;

import cn.com.rydeen.base.utils.data.JsonUtils;
import cn.com.rydeen.base.utils.security.HashUtils;
import cn.com.rydeen.component.mqtt.enums.QosEnum;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.io.Serializable;

/**
 * 消费者消息接口
 *
 * @author richie696
 * @version 1.0
 * @since 2022-09-13 11:07:24
 */
@Slf4j
@AllArgsConstructor
public class ConsumerMessage implements Serializable {

    /**
     * 本消息内容是否可修改
     */
    protected boolean mutable = true;
    /**
     * 消息体
     */
    private byte[] payload;
    /**
     * QoS级别
     */
    private int qos = 1;
    /**
     * 是否在服务器保留本条消息（服务器仅会保留最新的1条）
     */
    private boolean retained = false;

    protected boolean dup = false;
    /**
     * 消息 ID
     */
    private int messageId;
    /**
     * 消息来源主题
     */
    private String topic;

    /**
     * 验证QoS值是否有效的方法
     *
     * @param qos 待验证的 QoS 值
     * @throws IllegalArgumentException 如果 QoS 值不等于0、1、2则抛出错误参数异常.
     */
    public static void validateQos(int qos) {
        if (QosEnum.isValid(qos)) {
            return;
        }
        // 如果无效则抛出QoS值无效异常
        throw new IllegalArgumentException("The qos value is invalid.");
    }

    /**
     * 默认构造函数
     * <p>（注：通过本构造函数创建的消息体禁止消费者进行修改操作）</p>
     *
     * <p>默认值为：</p>
     * <ul>
     *    <li>消息报文为空</li>
     *    <li>消息 QoS 值默认为：1</li>
     *    <li>消息的 retained 值为：false，消息将不会被服务器“保留”</li>
     * </ul>
     */
    public ConsumerMessage() {
        setPayload(new byte[]{});
    }

    /**
     * 根据消息报文字节数组创建消息的构造函数
     * <p>（注：通过本构造函数创建的消息体禁止消费者进行修改操作）</p>
     *
     * <p>其它默认值为：</p>
     * <ul>
     *    <li>消息 QoS 值默认为：1</li>
     *    <li>消息的 retained 值为：false，消息将不会被服务器“保留”</li>
     * </ul>
     *
     * @param payload The Bytearray of the payload
     */
    public ConsumerMessage(byte[] payload) {
        setPayload(payload);
    }

    /**
     * 根据原始报文创建消息的构造函数
     * <p>（注：通过本构造函数创建的消息体禁止消费者进行修改操作）</p>
     *
     * @param topic 订阅的主题
     * @param mqttMessage 原始消息体
     */
    public ConsumerMessage(String topic, MqttMessage mqttMessage) {
        this.payload = mqttMessage.getPayload();
        this.qos = mqttMessage.getQos();
        this.retained = mqttMessage.isRetained();
        this.dup = mqttMessage.isDuplicate();
        this.messageId = mqttMessage.getId();
        this.topic = topic;
    }

    /**
     * 根据原始报文创建消息的构造函数
     *
     * @param topic 订阅的主题
     * @param mqttMessage 原始消息体
     * @param mutable 是否允许修改消息（true：允许，false：不允许）
     */
    public ConsumerMessage(String topic, MqttMessage mqttMessage, boolean mutable) {
        this.payload = mqttMessage.getPayload();
        this.qos = mqttMessage.getQos();
        this.retained = mqttMessage.isRetained();
        this.dup = mqttMessage.isDuplicate();
        this.messageId = mqttMessage.getId();
        this.topic = topic;
        this.mutable = mutable;
    }

    /**
     * 获取消息报文字节数组的方法
     *
     * @return 返回消息报文字节数组
     */
    public byte[] getPayload() {
        return payload;
    }

    /**
     * 获取字符串类型的消息报文的方法
     *
     * @return 返回消息报文字符串
     */
    public String getPayloadString() {
        return new String(payload);
    }

    /**
     * 获取指定类型的消息体（将字节数组转换为消息体）
     *
     * @param cls 消息体类型
     * @param <T> 消息体具体的类型
     * @return 返回消息体
     * @throws IllegalArgumentException 当消息体无法正确转换时抛出该异常
     */
    public <T> T getBody(Class<T> cls) {
        return JsonUtils.getInstance().deserialize(getPayloadString(), cls);
    }

    /**
     * 清空消息报文的方法
     *
     * @return 返回当前消息
     * @throws IllegalStateException 如果当前消息不可被修改则抛出该异常
     */
    public ConsumerMessage clearPayload() {
        checkMutable();
        this.payload = new byte[]{};
        return this;
    }

    /**
     * 获取 Payload 哈希值的方法
     * @return 返回 Payload 哈希值
     */
    public String getHash() {
        return HashUtils.sha256(payload);
    }

    /**
     * 设置消息报文的方法
     *
     * @param payload 消息报文的字节数组
     * @return 返回当前消息
     * @throws IllegalStateException 如果消息不允许被修改，则抛出该异常
     * @throws NullPointerException  如果消息报文体为null，则抛出该异常
     */
    public ConsumerMessage setPayload(byte[] payload) {
        checkMutable();
        if (payload == null) {
            throw new NullPointerException();
        }
        this.payload = payload.clone();
        return this;
    }

    /**
     * 判断该消息是否被保留在服务器的方法
     * <ul>
     *     <li>当该值为 true 时，表示该条消息会被服务器缓存（当有新消息的时候该消息会被冲掉，服务器只会保留最近的一条消息）</li>
     *     <li>当该值为 false 时，表示服务器不会保留该消息</li>
     * </ul>
     *
     * @return 返回判断结果
     */
    public boolean isRetained() {
        return retained;
    }

    /**
     * 设置是否在消息队列引擎中保留本条消息的方法
     * <ul>
     *     <li>当 retained = true 时，当前发布的消息会被消息队列引擎保留下来，每个
     *     Topic 只会保留 1 条消息，并且与 QoS 配置无关。<br><span>（补充说明：
     *     该功能在物联网行业使用比较广泛，比如智能硬件与 MQTT 服务器连接，当我们
     *     打开手机 App 时，绑定在App 账户内的所有硬件设备的 Topic 都会主动推送对应
     *     硬件的联网状态，我们在手机 App 中就可以看到对应物联网设备是在线状态，还
     *     是离线状态了。）</span></li>
     *     <li>当 retained = false 时，为普通消息，消息是否保留受到 QoS 配置的控制。</li>
     *     <li>清空保留消息：发送1条 retained = true 并提供一个空字节数组作为 payload
     *     消息（例如: new byte[0]），消息队列引擎将从服务器清除该保留消息。</li>
     * </ul>
     *
     * @param retained 是否保留消息（true：保留，false：不保留）
     * @return 返回当前对象
     * @throws IllegalStateException 如果消息不允许被修改，则抛出该异常
     */
    public ConsumerMessage setRetained(boolean retained) {
        checkMutable();
        this.retained = retained;
        return this;
    }

    /**
     * 获取本条消息 QoS 值的方法
     *
     * @return 返回 QoS 值
     * @see #setQos(int)
     */
    public int getQos() {
        return qos;
    }

    /**
     * 设置本条消息 QoS 值的方法
     * <ul>
     * <li>QoS = 0：表示消息最多只会发送1次(0次或1次)。消息不会被持久保存到磁盘，也
     * 不会通过 ACK 确认，这种 QoS 的速度是最快的，但是应该只用于没有价值的消息。</li>
     *
     * <li>QoS = 1：表示消息至少会发送1次(&gt;=1)。消息只有在能够持久保存的情况下才
     * 能安全地传递，因此应用程序必须使用 MqttConnectOptions 提供一种持久方法。
     * 如果没有指定持久化机制，则在客户端失败时不会传递消息（QoS 默认值）。</li>
     *
     * <li>QoS = 2：表示消息只会传递一次。消息将被持久化到磁盘，并通过网络进行两
     * 阶段确认。消息只有在能够持久保存的情况下才能安全地传递，因此应用程序必须
     * 使用 MqttConnectOptions 提供一种持久方法。如果没有指定持久性机制，则在客
     * 户端失败时不会传递消息。</li>
     * </ul>
     *
     * <p>如果没有配置持久化，当出现网络或服务器问题时，仍然会传递 QoS 1 和 2 消息，
     * 因为客户机将在内存中保存状态。如果MQTT客户端关闭或失败，且没有配置持久化，
     * 则无法维护 QoS 1 和 2 消息的传递，因为客户端状态将丢失。</p>
     *
     * @param qos QoS 等级（可选择：0、1、2）
     * @return 返回当前消息
     * @throws IllegalArgumentException 如果 QoS 等级的值不是 0、1、2时抛出该异常
     * @throws IllegalStateException 如果消息不允许被修改，则抛出该异常
     */
    public ConsumerMessage setQos(int qos) {
        checkMutable();
        validateQos(qos);
        this.qos = qos;
        return this;
    }

    /**
     * 转换传递的消息体为字符串的方法
     *
     * @return 返回传递的消息体
     */
    @Override
    public String toString() {
        return getPayloadString();
    }

    /**
     * 检查此消息是否可能是已接收到的消息的副本的方法
     * <p>（注：本参数只能在服务端进行设置，需要配合 messageId 来回复确
     * 认，以保证消息传输完成，但本参数不能用于确定消息是否重复发送。）</p>
     *
     * @return 返回检查结果（<code>true</code>：消息可能是重复的，<code>false</code>：不重复）
     */
    public boolean isDuplicate() {
        return this.dup;
    }

    /**
     * 设置消息 MQTT ID 的方法
     * <p>这仅用于服务消费者从服务器接收到的消息的 MQTT ID，仅服务订阅
     * 者可用，此字段对服务发布者不起作用。</p>
     *
     * @param messageId 消息 ID
     * @return 返回当前对象
     */
    public ConsumerMessage setId(int messageId) {
        this.messageId = messageId;
        return this;
    }

    /**
     * 获取消息 MQTT ID 的方法
     *
     * @return 返回消息的 MQTT ID
     */
    public int getId() {
        return this.messageId;
    }

    /**
     * 设置此消息是否允许更改的方法
     *
     * @param mutable 是否允许修改（true：允许，false：不允许）
     * @return 返回当前对象
     */
    protected ConsumerMessage setMutable(boolean mutable) {
        this.mutable = mutable;
        return this;
    }

    /**
     * 检查当前消息是否允许修改的方法
     *
     * @throws IllegalStateException 如果消息不允许被修改，则抛出该异常
     */
    protected void checkMutable() throws IllegalStateException {
        if (!mutable) {
            throw new IllegalStateException();
        }
    }

    /**
     * 设置此消息是否可能是已接收到的消息的副本的方法
     *
     * @param dup 是否重复（<code>true</code>：消息可能是重复的，<code>false</code>：不重复）
     * @return 返回当前消息
     */
    protected ConsumerMessage setDuplicate(boolean dup) {
        this.dup = dup;
        return this;
    }

    /**
     * 获取消息所属 Topic 名称的方法
     *
     * @return 返回消息所属 Topic 名称
     */
    public String getTopic() {
        return topic;
    }

    /**
     * 设置消息所属 Topic 名称的方法
     *
     * @param topic 消息所属 Topic 名称
     * @return 返回当前消息
     */
    public ConsumerMessage setTopic(String topic) {
        this.topic = topic;
        return this;
    }

}

具体解释请看上文中的Javadoc文档，此处不做赘述。

至此，MQTT的客户端和服务端的使用示例已经介绍完毕，如有疑问请联系架构组。