【项目实战】使用 mica-mqtt 搭建MQTT的服务端

我不是女神ヾ 2024-03-17 21:13 64阅读 0赞

## 一、MQTT服务端介绍 ##

MQTT（Message Queuing Telemetry Transport）是一种轻量级的消息传递协议，常用于物联网设备之间的通信。MQTT 服务器是 MQTT 协议中的一部分，它充当了消息的中间商，允许设备发布和订阅消息。

## 二、常见的 MQTT 服务器 ##

这些 MQTT 服务器都有自己的特点和优势，您可以根据自己的需求选择合适的服务器。  
一般来说，  
如果您的应用场景比较简单，可以考虑使用 Mosquitto；  
如果您的应用需要高性能和可扩展性，可以考虑使用 HiveMQ；

### 2.1 Mosquitto ###

Mosquitto 是一个开源的 MQTT 服务器，适用于各种平台和设备。它提供了强大的消息传递功能，支持 QoS 0、1 和 2 等级，还支持多种身份验证机制（如 PSK、LDAP 等）。Mosquitto 可以轻松地与其他 MQTT 客户端和工具集成，例如 MQTT.fx、Paho 等。

### 2.2 HiveMQ ###

HiveMQ 是一个商业的 MQTT 服务器，提供高性能、高可用性和可扩展性的解决方案。它支持 MQTT 协议的所有版本（MQTT 3.1.1 和 MQTT 5.0），还支持其他消息传递协议，如 AMQP、STOMP 等。HiveMQ 还提供了一些额外的功能，如消息路由、消息过滤、QoS 等级控制等。

### 2.3 mica-mqtt ###

mica-mqtt 低延迟、高性能的 mqtt 物联网组件。开源，遵循Apache-2.0协议

#### 2.3.1 使用场景有如下 ####

*  物联网（云端 mqtt broker）
 *  物联网（边缘端消息通信）
 *  群组类 IM
 *  消息推送
 *  简单易用的 mqtt 客户端

#### 2.3.2 版本兼容 ####

<table> 
 <thead> 
  <tr> 
   <th>要求</th> 
   <th>Spring boot 版本</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>最低</td> 
   <td>2.1.0.RELEASE</td> 
  </tr> 
  <tr> 
   <td>最高</td> 
   <td>3.x</td> 
  </tr> 
 </tbody> 
</table>

[参考mica-mqtt-server-spring-boot-starter 使用文档][mica-mqtt-server-spring-boot-starter]

## 三、mica-mqtt-server-spring-boot-starter的使用步骤 ##

### 3.1 添加依赖 ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>net.dreamlu</groupId>
        <artifactId>mica-mqtt-server-spring-boot-starter</artifactId>
        <version>2.2.2</version>
    </dependency>

### 3.2 添加配置项 ###

注意端口号+ auth中的认证用户名与认证密码，当为True时，表示开启。

mqtt:
      server:
        enabled: true               # 是否开启服务端，默认：true
        #    ip: 0.0.0.0                # 服务端 ip 默认为空，0.0.0.0，建议不要设置
        port: 1883                  # 端口，默认：1883
        name: Mica-Mqtt-Server      # 名称，默认：Mica-Mqtt-Server
        buffer-allocator: HEAP      # 堆内存和堆外内存，默认：堆内存
        heartbeat-timeout: 120000   # 心跳超时，单位毫秒，默认: 1000 * 120
        read-buffer-size: 8KB       # 接收数据的 buffer size，默认：8k
        max-bytes-in-message: 10MB  # 消息解析最大 bytes 长度，默认：10M
        auth:
          enable: true             # 是否开启 mqtt 认证
          username: burt            # mqtt 认证用户名
          password: burt            # mqtt 认证密码
        debug: true                 # 如果开启 prometheus 指标收集建议关闭
        stat-enable: true           # 开启指标收集，debug 和 prometheus 开启时需要打开，默认开启，关闭节省内存
        web-port: 8083              # http、websocket 端口，默认：8083
        websocket-enable: true      # 是否开启 websocket，默认： true
        http-enable: false          # 是否开启 http api，默认： false
        http-basic-auth:
          enable: false             # 是否开启 http basic auth，默认： false
          username: mica            # http basic auth 用户名
          password: mica            # http basic auth 密码
        ssl:                        # mqtt tcp ssl 认证
          enabled: false            # 是否开启 ssl 认证，2.1.0 开始支持双向认证
          keystore-path:            # 必须参数：ssl keystore 目录，支持 classpath:/ 路径。
          keystore-pass:            # 必选参数：ssl keystore 密码
          truststore-path:          # 可选参数：ssl 双向认证 truststore 目录，支持 classpath:/ 路径。
          truststore-pass:          # 可选参数：ssl 双向认证 truststore 密码
          client-auth: none         # 是否需要客户端认证（双向认证），默认：NONE（不需要）

注意：**ssl** 存在三种情况

<table> 
 <thead> 
  <tr> 
   <th>服务端开启ssl</th> 
   <th>客户端</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ClientAuth 为 NONE（不需要客户端验证）</td> 
   <td>仅仅需要开启 ssl 即可不用配置证书</td> 
  </tr> 
  <tr> 
   <td>ClientAuth 为 OPTIONAL（与客户端协商）</td> 
   <td>需开启 ssl 并且配置 truststore 证书</td> 
  </tr> 
  <tr> 
   <td>ClientAuth 为 REQUIRE (必须的客户端验证)</td> 
   <td>需开启 ssl 并且配置 truststore、 keystore证书</td> 
  </tr> 
 </tbody> 
</table>

### 3.3 可实现接口（注册成 Spring Bean 即可） ###

<table> 
 <thead> 
  <tr> 
   <th>接口</th> 
   <th>是否必须</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>IMqttServerUniqueIdService</td> 
   <td>否</td> 
   <td>用于 clientId 不唯一时，自定义实现唯一标识，后续接口使用它替代 clientId</td> 
  </tr> 
  <tr> 
   <td>IMqttServerAuthHandler</td> 
   <td>是</td> 
   <td>用于服务端认证</td> 
  </tr> 
  <tr> 
   <td>IMqttServerSubscribeValidator</td> 
   <td>否（建议实现）</td> 
   <td>1.1.3 新增，用于对客户端订阅校验</td> 
  </tr> 
  <tr> 
   <td>IMqttServerPublishPermission</td> 
   <td>否（建议实现）</td> 
   <td>1.2.2 新增，用于对客户端发布权限校验</td> 
  </tr> 
  <tr> 
   <td>IMqttMessageListener</td> 
   <td>否（1.3.x为否）</td> 
   <td>消息监听</td> 
  </tr> 
  <tr> 
   <td>IMqttConnectStatusListener</td> 
   <td>是</td> 
   <td>连接状态监听</td> 
  </tr> 
  <tr> 
   <td>IMqttSessionManager</td> 
   <td>否</td> 
   <td>session 管理</td> 
  </tr> 
  <tr> 
   <td>IMqttSessionListener</td> 
   <td>否</td> 
   <td>session 监听</td> 
  </tr> 
  <tr> 
   <td>IMqttMessageStore</td> 
   <td>集群是，单机否</td> 
   <td>遗嘱和保留消息存储</td> 
  </tr> 
  <tr> 
   <td>AbstractMqttMessageDispatcher</td> 
   <td>集群是，单机否</td> 
   <td>消息转发，（遗嘱、保留消息转发）</td> 
  </tr> 
  <tr> 
   <td>IpStatListener</td> 
   <td>否</td> 
   <td>t-io ip 状态监听</td> 
  </tr> 
  <tr> 
   <td>IMqttMessageInterceptor</td> 
   <td>否</td> 
   <td>消息拦截器，1.3.9 新增</td> 
  </tr> 
 </tbody> 
</table>

### 3.4 IMqttMessageListener (用于监听客户端上传的消息) 使用示例 ###

@Service
    public class MqttServerMessageListener implements IMqttMessageListener {
        
        private static final Logger logger = LoggerFactory.getLogger(MqttServerMessageListener.class);
    
    	//注意：此处与官网的指引不一样。
        @Override
        public void onMessage(ChannelContext channelContext, String clientId, String topic, MqttQoS mqttQoS, MqttPublishMessage mqttPublishMessage) {
        
            logger.info("clientId:{} message:{} payload:{}", clientId, topic, new String(mqttPublishMessage.getPayload(), StandardCharsets.UTF_8));
        }
    }

### 3.5 自定义配置（可选） ###

@Configuration(proxyBeanMethods = false)
    public class MqttServerCustomizerConfiguration {
        
        @Bean
        public MqttServerCustomizer mqttServerCustomizer() {
        
            return new MqttServerCustomizer() {
        
                @Override
                public void customize(MqttServerCreator creator) {
        
                    // 此处可自定义配置 creator，会覆盖 yml 中的配置
                    System.out.println("----------------MqttServerCustomizer-----------------");
                }
            };
        }
    }

### 3.6 MqttServerTemplate 使用示例 ###

如果有新的MQTT的服务端，只要进行叠加即可。

@Service
    public class ServerService {
        
    
        @Autowired
        private MqttServerTemplate server;
    
        public boolean publish123(String body) {
        
            server.publishAll("/test/123", body.getBytes(StandardCharsets.UTF_8));
            return true;
        }
    
        public boolean publish456(String body) {
        
            server.publishAll("/test/456", body.getBytes(StandardCharsets.UTF_8));
            return true;
        }
    }

### 3.7 客户端上下线监听 ###

使用 Spring event 解耦客户端上下线监听，注意： `1.3.4` 开始支持。会跟自定义的 `IMqttConnectStatusListener` 实现冲突，取一即可。

@Service
    public class MqttConnectStatusListener {
        
    	private static final Logger logger = LoggerFactory.getLogger(MqttConnectStatusListener.class);
    
    	@EventListener
    	public void online(MqttClientOnlineEvent event) {
        
    		logger.info("MqttClientOnlineEvent:{}", event);
    	}
    
    	@EventListener
    	public void offline(MqttClientOfflineEvent event) {
        
    		logger.info("MqttClientOfflineEvent:{}", event);
    	}
    
    }

### 3.8 验证MQTT的服务端 ###

客户端发送了这个消息  
![在这里插入图片描述][eac67c6585ef48b49ced3d2a62c3115f.png]  
服务运行后，输出如下的信息，将客户端发送的消息+客户端的IP等信息都打印了出来。  
![在这里插入图片描述][980c0a0ae129404991885d1c37e1ffab.png]

[mica-mqtt-server-spring-boot-starter]: https://gitee.com/596392912/mica-mqtt/edit/master/starter/mica-mqtt-server-spring-boot-starter/README.md
[eac67c6585ef48b49ced3d2a62c3115f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/b90a9134bed348d7ae9e4c55d6e21832.png
[980c0a0ae129404991885d1c37e1ffab.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/0e4d5acdd9b243cbac7efb1bced9357c.png