Spring 整合MQTT

Spring提供了对多种消息中间件的整合，其中也包括MQTT。具体请参见以下链接：

https://docs.spring.io/spring-integration/reference/html/

Spring整合MQTT步骤如下：

1. 创建Spring Boot Maven工程，引入如下依赖：

   <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
   
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
   
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <version>5.1.3.RELEASE</version>
        </dependency>
    </dependencies>
   

2. 配置MQTT消费端

   添加SpringConfig.java类，添加消息消费Bean

   /*****
        * 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置。
        * @return
        */
       @Bean
       public MqttPahoClientFactory mqttClientFactory() {
           DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
           MqttConnectOptions options = new MqttConnectOptions();
           options.setServerURIs(new String[]{"tcp://10.69.94.176:1883"});
           factory.setConnectionOptions(options);
           return factory;
       }
   
       @Bean
       public MessageChannel mqttInputChannel() {
           return new DirectChannel();
       }
   
       @Bean
       public MessageProducer inbound() {
           MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("consumerClient",
                   mqttClientFactory(), "topic1", "topic2");
           adapter.setCompletionTimeout(5000);
           adapter.setConverter(new DefaultPahoMessageConverter());
           adapter.setQos(1);
           adapter.setOutputChannel(mqttInputChannel());
           return adapter;
       }
   
       @Bean
       //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息信息的channel。
       @ServiceActivator(inputChannel = "mqttInputChannel")
       public MessageHandler handler() {
           return message -> {
               String payload = message.getPayload().toString();
               String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
               // 根据topic分别进行消息处理。
               if (topic.equals("topic1")) {
                   System.out.println("topic1: 处理消息 " + payload);
               } else if (topic.equals("topic2")) {
                   System.out.println("topic2: 处理消息 " + payload);
               } else {
                   System.out.println(topic + ": 丢弃消息 " + payload);
               }
           };
       }
   

   @ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息的channel。

   当接收到消息时，可以先拿到topic，然后根据不同的topic分别对消息进行处理。

3. 配置MQTT消息发送端。

   在MQTT使用场景中，一般处理接收消息的同时，也会发送消息。在SpringConfig.java配置文件中添加如下Bean注入，用于消息发送。

    @Bean
       public MessageChannel mqttOutboundChannel() {
           return new DirectChannel();
       }
   
       /*****
        * 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
        * @return
        */
       @Bean
       @ServiceActivator(inputChannel = "mqttOutboundChannel")
       public MessageHandler outbound() {
           // 在这里进行mqttOutboundChannel的相关设置
           MqttPahoMessageHandler messageHandler =
                   new MqttPahoMessageHandler("publishClient", mqttClientFactory());
           messageHandler.setAsync(true); //如果设置成true，发送消息时将不会阻塞。
           messageHandler.setDefaultTopic("testTopic");
           return messageHandler;
       }
   
       @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
       public interface MqttGateway {
           // 定义重载方法，用于消息发送
           void sendToMqtt(String payload);
           // 指定topic进行消息发送
           void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
           void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
       }
   

   @MessagingGateway是一个用于提供消息网关代理整合的注解，参数defaultRequestChannel指定发送消息绑定的channel。

   在这里我们定义了MqttGateway接口，该接口可以被注入到其它类中，用于消息发送。

    @Autowired
       private SpringConfig.MqttGateway mqttGateway;
       
       // 然后在类方法中调用下面方法发送消息
       // mqttGateway.sendToMqtt("testTopic",message);
   

4. 创建Rest Controller，通过http请求发送MQTT消息。

   @RestController
   public class MqttController {
   
       @Autowired
       private SpringConfig.MqttGateway mqttGateway;
   
       @RequestMapping("/send/{topic}/{message}")
       public String send(@PathVariable String topic, @PathVariable String message) {
           mqttGateway.sendToMqtt(topic, message);
           return "send message : " + message;
       }
   }
   

   在浏览器中输入如下网址进行测试。

   http://localhost:8080/send/topic1/message1

   http://localhost:8080/send/topic2/message2

5. 项目源代码可参考https://github.com/40925645/zengbiaobiao/new/master/mqtt-demo