---
url: /diary/4d7zg3i5/index.md
description: 封装 MQTT 消息服务的 Spring Boot Starter，支持消息发布订阅，用于实现实时消息推送和弹幕功能。
---
最近有个项目需要发送视频弹幕，实现的手段还是不少的，这里用Mqtt来实现，使用技术如下：

* Spring Cloud
* Spring Boot
* JDK 8

## 准备

首先是环境准备，这里使用 [EMQX 5.0 ](https://www.emqx.io/docs/zh/v5.0/)社区版作为`Mqtt`的消息服务器，使用`Docker`进行搭建：

下附 `docker-compose`：

```yaml
version: '3.2'
services:
  emqx:
    image: emqx/emqx:v5.0.0
    container_name: emqx
    environment:
      EMQX_NAME: xinwei-emqx
    restart: always
    ports:
      - '1883:1883'
      - '8081:8081'
      - '8083:8083'
      - '8883:8883'
      - '8084:8084'
      - '8099:8080'
      - '18083:18083'
    volumes:
      - /service/env/emqx/data:/opt/emqx/data
      - /service/env/emqx/etc:/opt/emqx/etc
      - /service/env/emqx/log:/opt/emqx/log
```

## 封装

### 配置依赖

`maven` 加入相关依赖，具体版本可以由项目的`spring-boot`版本决定。

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

### 配置类

> \[!NOTE]
> 本文不对`mqtt`进行鉴权，需要的可以自行拓展

1. 基础属性配置
   ```java :collapsed-lines
   @Data
   @ConfigurationProperties(prefix = "spring.mqtt")
   @RefreshScope
   public class MqttProperties {


       private String url;

       private AuthTypeEnum authType = AuthTypeEnum.NONE; 

       private String username;
       private String password;
       private String wsUrl;
       private String client;

       private Topic defaultSendTopic = new Topic("xx_mqtt_default", 1);

       private List<Topic> receiveTopics;

       private Integer keepAlive = 15;

       private Integer completionTimeout = 3000;


       /**
        * Mqtt 权限认证类型枚举
        */
       @NoArgsConstructor
       public enum AuthTypeEnum {
           /**
            * 无
            */
           NONE,
           /**
            * 客户端id
            */
           CLIENT_ID,
           /**
            * 用户名
            */
           USERNAME
       }

       @Data
       @NoArgsConstructor
       @AllArgsConstructor
       public static class Topic {
           private String name;

           private Integer qos;
       }
   }
   ```
2. 注入配置
   ```java :collapsed-lines
   @AutoConfiguration
   @EnableConfigurationProperties(MqttProperties.class)
   public class MqttConfiguration {

       @Bean
       public MessageChannel mqttValueInputChannel() {
           return new DirectChannel();
       }

       /**
        * 不缓存消息
        *
        * @return 消息通道
        */
       @Bean
       public MessageChannel mqttOutputClearSessionChannel() {
           return new DirectChannel();
       }

       /**
        * 缓存消息
        *
        * @return 消息通道
        */
       @Bean
       public MessageChannel mqttOutputRetainSessionChannel() {
           return new DirectChannel();
       }

       @Bean
       public MessageProducer valueInbound(MqttProperties mqttProperties) {
           MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
               mqttProperties.getClient() + StrPool.DASHED + "receiver",
               mqttClearSessionClientFactory(mqttProperties),
               mqttProperties.getReceiveTopics().stream()
                   .map(MqttProperties.Topic::getName)
                   .toArray(String[]::new));
           adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
           adapter.setConverter(new DefaultPahoMessageConverter());
           adapter.setQos(mqttProperties.getReceiveTopics().stream()
               .mapToInt(MqttProperties.Topic::getQos)
               .toArray());
           adapter.setOutputChannel(mqttValueInputChannel());
           return adapter;
       }

       @Bean
       @ServiceActivator(inputChannel = "mqttOutputClearSessionChannel")
       public MessageHandler clearSessionOutbound(MqttProperties mqttProperties) {
           MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getClient() + StrPool.DASHED + "clear-session-sender", mqttClearSessionClientFactory(mqttProperties));
           messageHandler.setAsync(true);
           messageHandler.setDefaultQos(mqttProperties.getDefaultSendTopic().getQos());
           messageHandler.setDefaultTopic(mqttProperties.getDefaultSendTopic().getName());
           return messageHandler;
       }

       @Bean
       public MqttPahoClientFactory mqttClearSessionClientFactory(MqttProperties mqttProperties) {
           DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
           factory.setConnectionOptions(getMqttClearSessionConnectOptions(mqttProperties));
           return factory;
       }

       @Bean
       public MqttConnectOptions getMqttClearSessionConnectOptions(MqttProperties mqttProperties) {
           MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();

           if (mqttProperties.getAuthType().equals(MqttProperties.AuthTypeEnum.USERNAME)) {
               mqttConnectOptions.setUserName(mqttProperties.getUsername());
               mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
           }
           mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getUrl()});
           mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAlive());
           return mqttConnectOptions;
       }

       @Bean
       @ServiceActivator(inputChannel = "mqttOutputRetainSessionChannel")
       public MessageHandler retainSessionOutbound(MqttProperties mqttProperties) {
           MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getClient() + StrPool.DASHED + "retain-session-sender", mqttRetainSessionClientFactory(mqttProperties));
           messageHandler.setAsync(true);
           messageHandler.setDefaultQos(mqttProperties.getDefaultSendTopic().getQos());
           messageHandler.setDefaultTopic(mqttProperties.getDefaultSendTopic().getName());
           return messageHandler;
       }


       @Bean
       public MqttPahoClientFactory mqttRetainSessionClientFactory(MqttProperties mqttProperties) {
           DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
           factory.setConnectionOptions(getMqttRetainSessionConnectOptions(mqttProperties));
           return factory;
       }

       @Bean
       public MqttConnectOptions getMqttRetainSessionConnectOptions(MqttProperties mqttProperties) {
           MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();

           if (mqttProperties.getAuthType().equals(MqttProperties.AuthTypeEnum.USERNAME)) {
               mqttConnectOptions.setUserName(mqttProperties.getUsername());
               mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
           }
           mqttConnectOptions.setCleanSession(false);
           mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getUrl()});
           mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAlive());
           return mqttConnectOptions;
       }
   }
   ```
3. 自动注入配置
   ```java
   /**
    * 消息接收
    */
   @Slf4j
   @AutoConfiguration(after = MqttConfiguration.class)
   public class MqttReceiveConfiguration {

       @Bean
       @ServiceActivator(inputChannel = "mqttValueInputChannel")
       public MessageHandler handlerValue() {
           return message -> {
               MessageHeaders headers = message.getHeaders();
               //获取消息Topic
               String receivedTopic = (String) headers.get(MqttHeaders.RECEIVED_TOPIC);
               log.info("获取到的消息的topic :{} ", receivedTopic);
               //获取消息体
               String payload = (String) message.getPayload();
               log.info("获取到的消息的payload :{} ", payload);
           };
       }
   }
   ```
4. imports

   ::: file-tree

   * resource
     * META-INF
       * spring
         * org.springframework.boot.autoconfigure.AutoConfiguration.imports
           :::

   配置自动装配路径

   ```
   com.xxx.cloud.mqtt.config.MqttConfiguration
   com.xxx.cloud.mqtt.config.MqttReceiveConfiguration
   ```

### 消息发送设置

#### 清除session

```java :collapsed-lines
@MessagingGateway(defaultRequestChannel = "mqttOutputClearSessionChannel")
public interface MqttClearSessionSendHandler {
    /**
     * 使用 Default Topic & Default Qos 发送数据
     *
     * @param data string
     */
    void sendToMqtt(String data);

    /**
     * 使用 Default Topic & 自定义 Qos 发送数据
     *
     * @param qos  自定义 Qos
     * @param data string
     */
    void sendToMqtt(@Header(MqttHeaders.QOS) Integer qos, String data);

    /**
     * 使用 自定义 Topic & Default Qos 发送数据
     *
     * @param topic 自定义 Topic
     * @param data  string
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data);

    /**
     * 使用 自定义 Topic & 自定义 Qos 发送数据
     *
     * @param topic 自定义 Topic
     * @param qos   自定义 Qos
     * @param data  string
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer qos, String data);

    /**
     * 使用 自定义 Topic & 自定义 Qos 发送数据
     *
     * @param topic   自定义 Topic
     * @param qos     自定义 Qos
     * @param payload 自定义负载
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
```

#### 不清除session

```java :collapsed-lines
@MessagingGateway(defaultRequestChannel = "mqttOutputRetainSessionChannel")
public interface MqttRetainSessionSendHandler {

    /**
     * 使用 Default Topic & Default Qos 发送数据
     *
     * @param data string
     */
    void sendToMqtt(String data);

    /**
     * 使用 Default Topic & 自定义 Qos 发送数据
     *
     * @param qos  自定义 Qos
     * @param data string
     */
    void sendToMqtt(@Header(MqttHeaders.QOS) Integer qos, String data);

    /**
     * 使用 自定义 Topic & Default Qos 发送数据
     *
     * @param topic 自定义 Topic
     * @param data  string
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data);

    /**
     * 使用 自定义 Topic & 自定义 Qos 发送数据
     *
     * @param topic 自定义 Topic
     * @param qos   自定义 Qos
     * @param data  string
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer qos, String data);

    /**
     * 使用 自定义 Topic & 自定义 Qos 发送数据
     *
     * @param topic   自定义 Topic
     * @param qos     自定义 Qos
     * @param payload 自定义负载
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
}
```

## 使用

1. 引入上述封装的`starter`依赖
2. 配置信息
   ```yaml
   spring:
     mqtt:
       url: tcp://xxx.xxx.com:1883
       ws-url: ws://xxx.xxx.com:8083
       auth-type: NONE
       username: admin
       password: public
       #    ca-crt: classpath:/certs/broker.emqx.io-ca.crt
       client:  ${spring.application.name}
       receive-topics:
         - qos: 1
           name: driver/${spring.application.name}/device/+
         - qos: 1
           name: driver/${spring.application.name}/gateway/+
       default-send-topic:
         qos: 1
         name: default/${spring.application.name}
       keep-alive: 15
       completion-timeout: 3000
   ```
3. 注入后调用方法即可发送
   ```java
   @Resource
   private MqttClearSessionSendHandler sendHandler;
   ```