Spring提供了對(duì)多種消息中間件的整合，其中也包括MQTT。具體請(qǐng)參見(jiàn)以下鏈接：

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

Spring整合MQTT步驟如下：

1. 創(chuàng)建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消費(fèi)端

   添加SpringConfig.java類，添加消息消費(fèi)Bean

   /*****
        * 創(chuàng)建MqttPahoClientFactory覆履，設(shè)置MQTT Broker連接屬性怨酝，如果使用SSL驗(yàn)證，也在這里設(shè)置漆魔。
        * @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注解表明當(dāng)前方法用于處理MQTT消息坷檩，inputChannel參數(shù)指定了用于接收消息信息的channel却音。
       @ServiceActivator(inputChannel = "mqttInputChannel")
       public MessageHandler handler() {
           return message -> {
               String payload = message.getPayload().toString();
               String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
               // 根據(jù)topic分別進(jìn)行消息處理。
               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注解表明當(dāng)前方法用于處理MQTT消息矢炼，inputChannel參數(shù)指定了用于接收消息的channel系瓢。

   當(dāng)接收到消息時(shí)，可以先拿到topic句灌，然后根據(jù)不同的topic分別對(duì)消息進(jìn)行處理夷陋。

3. 配置MQTT消息發(fā)送端。

   在MQTT使用場(chǎng)景中胰锌，一般處理接收消息的同時(shí)骗绕，也會(huì)發(fā)送消息。在SpringConfig.java配置文件中添加如下Bean注入资昧，用于消息發(fā)送酬土。

    @Bean
       public MessageChannel mqttOutboundChannel() {
           return new DirectChannel();
       }
   
       /*****
        * 發(fā)送消息和消費(fèi)消息Channel可以使用相同MqttPahoClientFactory
        * @return
        */
       @Bean
       @ServiceActivator(inputChannel = "mqttOutboundChannel")
       public MessageHandler outbound() {
           // 在這里進(jìn)行mqttOutboundChannel的相關(guān)設(shè)置
           MqttPahoMessageHandler messageHandler =
                   new MqttPahoMessageHandler("publishClient", mqttClientFactory());
           messageHandler.setAsync(true); //如果設(shè)置成true，發(fā)送消息時(shí)將不會(huì)阻塞格带。
           messageHandler.setDefaultTopic("testTopic");
           return messageHandler;
       }
   
       @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
       public interface MqttGateway {
           // 定義重載方法撤缴，用于消息發(fā)送
           void sendToMqtt(String payload);
           // 指定topic進(jìn)行消息發(fā)送
           void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
           void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
       }
   

   @MessagingGateway是一個(gè)用于提供消息網(wǎng)關(guān)代理整合的注解，參數(shù)defaultRequestChannel指定發(fā)送消息綁定的channel叽唱。

   在這里我們定義了MqttGateway接口腹泌，該接口可以被注入到其它類中，用于消息發(fā)送尔觉。

    @Autowired
       private SpringConfig.MqttGateway mqttGateway;
       
       // 然后在類方法中調(diào)用下面方法發(fā)送消息
       // mqttGateway.sendToMqtt("testTopic",message);
   

4. 創(chuàng)建Rest Controller凉袱，通過(guò)http請(qǐng)求發(fā)送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;
       }
   }
   

   在瀏覽器中輸入如下網(wǎng)址進(jìn)行測(cè)試侦铜。

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

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

5. 項(xiàng)目源代碼可參考https://github.com/40925645/zengbiaobiao/new/master/mqtt-demo