随着物联网 (IoT) 和微服务架构的日益普及，MQTT 已经成为一个关键的消息传递协议。在本文中，我们将讨论如何在 Spring Boot 应用程序中集成 MQTT，从而为你的应用带来更好的扩展性和响应性。

1. 什么是 MQTT？

MQTT（Message Queuing Telemetry Transport）是一个轻量级的发布/订阅协议，专为低带宽、高延迟或不可靠的网络设计。它广泛应用于 IoT 场景中，连接传感器、设备和应用。

2. 为什么选择 MQTT？

  • 轻量级：MQTT 的报文头部很小，非常适合于受到带宽限制的网络。
  • 消息等级：它支持不同级别的消息传递保证。
  • 持久会话：支持持久会话，即客户端和服务器之间的会话状态可以被保留。
  • 最后遗言：如果一个客户端断开连接，它可以预先设定一个“最后遗言”消息。

3. Spring Boot 中的 MQTT

要在 Spring Boot 中使用 MQTT，我们首先需要添加相关的依赖。我们将使用 Eclipse Paho Java 客户端作为 MQTT 的客户端库。

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.5.9</version>
</dependency>
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

4. 配置 MQTT

包含接收消息的配置和发送消息的配置

package com.demo.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;

/**
 * mqtt连接配置
 */
@Configuration
public class MqttConfig {

    /**
     * 创建连接
     *
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        
        // mqtt用户名&密码
        String userName = "";
        String pwd = "";
        // mqtt服务地址，可以是多个
        options.setServerURIs(new String[]{"tcp://server:1883"});
        options.setUserName(userName);
        options.setPassword(pwd.toCharArray());
        factory.setConnectionOptions(options);

        return factory;
    }

    /**
     * 2、接收消息的通道
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 接收消息
     *
     * @return
     */
    @Bean
    public MessageProducer inbound() {
        // 订阅主题,保证唯一性
        String inClientId = UUID.randomUUID().toString().replaceAll("-", "");
        // 最后的#相当于通配符的概念
        String[] topic = {"topic_prefix/topic/#"};
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                inClientId,
                mqttClientFactory(),
                topic);

        adapter.setCompletionTimeout(5000);

        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        // 按字节接收消息
//        defaultPahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(defaultPahoMessageConverter);
        // 设置QoS
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }


    /**
     * 3、消息处理
     * ServiceActivator注解表明：当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            String payload = message.getPayload().toString();

            // byte[] bytes = (byte[]) message.getPayload(); // 收到的消息是字节格式
            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();

            // 可以根据topic进行处理不同的业务类型
            System.out.println("主题[" + topic + "]，负载：" + payload);
        };
    }


    /**
     * 发送消息的通道
     *
     * @return
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 发送消息
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler outbound() {

        // 连接clientId保证唯一
        String outClientId = UUID.randomUUID().toString().replaceAll("-", "");

        // 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(outClientId, mqttClientFactory());
        // 如果设置成true，即异步，发送消息时将不会阻塞。
        // messageHandler.setAsync(true);
        // 设置默认的topic
        // messageHandler.setDefaultTopic("defaultTopic");
        // 设置默认QoS
        messageHandler.setDefaultQos(1);

        // Paho消息转换器
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();

        // 发送默认按字节类型发送消息
        // defaultPahoMessageConverter.setPayloadAsBytes(true);
        messageHandler.setConverter(defaultPahoMessageConverter);
        return messageHandler;
    }
}

5. 消息发送

1. 定义消息发送的接口

package com.demo.config;

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;

/**
 * 定义消息发送的接口
 */
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateWay {

    /**
     * 发送消息
     *
     * @param payload 发送的消息
     */
    void sendToMqtt(String payload);

    /**
     * 指定topic消息发送
     *
     * @param topic   指定topic
     * @param payload 消息
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, byte[] payload);
}

2. 定义消息发送的controller

package com.demo.business;

import com.sonli.config.MqttGateWay;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 对外暴露发送消息的controller
 */
@RestController
@RequestMapping("/mqtt")
public class MqttController {

    @Autowired
    private MqttGateWay mqttGateWay;

    @PostMapping("/sendMessage")
    public String sendMessage(String topic, String message) {
        // 发送消息到指定topic
        mqttGateWay.sendToMqtt(topic, 1, message);
        return "send topic: " + topic + ", message : " + message;
    }
}

测试

发送消息

如何在 Spring Boot 中集成 MQTT？-MQTT中文站

消息的监听，收到的消息

如何在 Spring Boot 中集成 MQTT？-MQTT中文站

总结

Spring Boot 与 MQTT 的集成为开发者提供了一个简单但功能强大的方式，用于创建 IoT 和消息驱动的应用程序。这种集成不仅确保了消息传递的可靠性和效率，还使得应用程序更具扩展性和响应性。

希望这篇文章能为你提供在 Spring Boot 中集成 MQTT 的指导。如果你有任何问题或需要进一步的指导，请随时留言或联系我们。

打赏微海报分享