六、Java中SpringBoot组件集成接入【MQTT中间件】

六、Java中SpringBoot组件集成接入【MQTT中间件】

• • 1.MQTT介绍
  • 2.搭建MQTT服务器
  • • 1.Windows
    • 2.Ubuntu
    • 3.Docker
    • 4.其他方式
    • 3.mqtt可视化客户端MQTTX及快速使用教程
    • 4.SpringBoot接入MQTT
    • • 1、maven依赖
      • 2、MQTT配置
      • 3、MQTT组件具体代码
      • • 1.定义通道名字
        • 2.消息发布器
        • 3.MQTT配置、生产者、消费者
        • 4.调用生产者像指定通道发布消息
        • 5.测试
        • • 1.先在客户端测试一下，可以看到mqtt服务是正常的
          • 2.调用生产者发布消息的接口，可以看到成功发布了
          • 3.从日志中也可以看到之前的消息都被程序接收到了，只需要在接收位置做业务逻辑即可
          • 5.其他参考文章

            1.MQTT介绍

            MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅消息传输协议，专门用于物联网（IoT）领域的通信。它旨在提供可靠且高效的消息传递机制，使设备和应用程序能够在低带宽、不稳定网络环境下进行实时通信。

            MQTT协议采用了发布/订阅模式，其中包含两个主要角色：发布者（Publisher）和订阅者（Subscriber）。发布者负责将消息发布到特定的主题（Topic），而订阅者则通过订阅特定主题来接收相关消息。这种松耦合的通信方式使得系统能够轻松地扩展和管理。

            MQTT协议具有以下特点：

            • 轻量级：MQTT协议使用简单的二进制协议头，消息开销很小，适合在低带宽、低功耗设备上使用。
            • 可靠性：MQTT支持确认和重传机制，确保消息在不稳定网络中的可靠传输。
            • 异步通信：发布者和订阅者不需要互相了解对方的存在，使得系统更加松耦合。
            • 灵活的消息过滤：订阅者可以使用通配符来过滤感兴趣的主题，灵活地选择接收哪些消息。
            • 安全性：MQTT支持通过TLS/SSL协议进行加密传输，并且可以使用认证机制进行身份验证。

              由于其轻量级和简单易用的特点，MQTT被广泛用于物联网场景中，如传感器数据采集、远程监控、智能家居等。

              官方资料链接：

              MQTT ORG

              EMQX

              2.搭建MQTT服务器

              1.Windows

              1. 下载emqx-5.1.0-windows-amd64.zip，解压
              2. 命令行下进入解压路径，执行以下命令启动 emqx

              ./emqx/bin/emqx start

              2.Ubuntu

              1.通过以下命令配置 EMQX Apt 源：

              curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash

              2.运行以下命令安装 EMQX：

              sudo apt-get install emqx

              3.运行以下命令启动 EMQX：

              sudo systemctl start emqx

              3.Docker

              1.运行以下命令获取 Docker 镜像：

              docker pull emqx/emqx:5.4.0

              2.运行以下命令启动 Docker 容器。

              docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.4.0

              4.其他方式

              参考：EMQX文档

              3.mqtt可视化客户端MQTTX及快速使用教程

              MQTTX下载地址：MQTT 5.0 客户端工具

              快速使用教程参考文章：Windows版mqtt可视化客户端快速使用教程

              4.SpringBoot接入MQTT

              1、maven依赖

              org.springframework.integrationspring-integration-core5.4.2org.springframework.integrationspring-integration-mqtt5.5.5

              2、MQTT配置

              在bootstrap.yml中增加MQTT配置信息，可根据自身情况修改相关字段

              #是否启用MQTT服务spring.mqtt.enable=true#MQTT-服务端地址（本地服务器）spring.mqtt.url=tcp://127.0.0.1:1883#MQTT-服务端用户名spring.mqtt.username=admin#MQTT-服务端密码（本地服务器）spring.mqtt.password=public#MQTT-客户端clientidspring.mqtt.clientid=test001#MQTT-Warning默认主题spring.mqtt.topicWarning=topic_warning

              3、MQTT组件具体代码

              1.定义通道名字

              public class ChannelName {    /**     * 订阅的bean名称     */    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";    /**     * 发布的bean名称     */    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";}

              2.消息发布器

              import com.funfan.autoCodeDemo.component.mqtt.model.ChannelName;import org.springframework.integration.annotation.MessagingGateway;import org.springframework.integration.mqtt.support.MqttHeaders;import org.springframework.messaging.handler.annotation.Header;import org.springframework.stereotype.Component;@Component@MessagingGateway(defaultRequestChannel = ChannelName.CHANNEL_NAME_OUT)public interface IMqttMessageGateway {    /**     * 发送信息到MQTT服务器     *     * @param payload 发送的文本     */    void sendToMqtt(String payload);    /**     * 发送信息到MQTT服务器     *     * @param topic   主题     * @param payload 消息主体     */    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);    /**     * 发送信息到MQTT服务器     *     * @param topic   主题     * @param qos     对消息处理的几种机制。     *                0 表示的是订阅者没收到消息不会再次发送，消息会丢失。     *                1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。     *                2 多了一次去重的动作，确保订阅者收到的消息有一次。     * @param payload 消息主体     */    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);}

              3.MQTT配置、生产者、消费者

              import com.funfan.autoCodeDemo.component.mqtt.model.ChannelName;import lombok.extern.slf4j.Slf4j;import org.eclipse.paho.client.mqttv3.MqttConnectOptions;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;import org.springframework.cloud.context.config.annotation.RefreshScope;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.integration.annotation.IntegrationComponentScan;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.Message;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.MessageHandler;import javax.annotation.PostConstruct;@Configuration@IntegrationComponentScan@Slf4j@ConditionalOnProperty(prefix = "spring.mqtt", name = "enable", havingValue = "true")@RefreshScopepublic class MqttConfig {    private static final Logger LOGGER = LoggerFactory.getLogger(MqttConfig.class);    @Value("${spring.mqtt.username}")    private String username;    @Value("${spring.mqtt.password}")    private String password;    @Value("${spring.mqtt.url}")    private String hostUrl;    @Value("${spring.mqtt.clientid}")    private String clientid;    @Value("${spring.mqtt.topicWarning}")    private String topicWarning;        /**     * 初始化日志：看有没有获取到相应配置     */    @PostConstruct    public void init() {        log.info("username:{} password:{} hostUrl:{} clientId :{} topicWarning:{}",                this.username, this.password, this.hostUrl, this.clientid, this.topicWarning);    }    /**     * MQTT连接器选项     */    @Bean    public  MqttConnectOptions getMqttConnectOptions() {        MqttConnectOptions options = new MqttConnectOptions();        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，        // 这里设置为true表示每次连接到服务器都以新的身份连接        options.setCleanSession(false);        // 设置连接的用户名        options.setUserName(username);        // 设置连接的密码        options.setPassword(password.toCharArray());        //可以是多个服务端        options.setServerURIs(new String[]{hostUrl});        // 设置超时时间 单位为秒        options.setConnectionTimeout(10);        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制        options.setKeepAliveInterval(20);        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。        //options.setWill("willTopic", WILL_DATA, 2, false);        return options;    }    /**     * MQTT客户端     */    @Bean    public MqttPahoClientFactory mqttClientFactory() {        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();        factory.setConnectionOptions(getMqttConnectOptions());        return factory;    }    /**     * MQTT信息通道（生产者）     */    @Bean(name = ChannelName.CHANNEL_NAME_OUT)    public MessageChannel mqttOutboundChannel() {        return new DirectChannel();    }    /**     * MQTT消息处理器（生产者）     */    @Bean    @ServiceActivator(inputChannel = ChannelName.CHANNEL_NAME_OUT)    public MessageHandler mqttOutbound() {        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(                clientid + "_publish", mqttClientFactory());        messageHandler.setAsync(true);        //设置生产者的消息级别        messageHandler.setDefaultQos(2);        //生产消息指定主题        messageHandler.setDefaultTopic(topicWarning);        return messageHandler;    }    /**     * MQTT信息通道（消费者）     */    @Bean(name = ChannelName.CHANNEL_NAME_IN)    public MessageChannel mqttInboundChannel() {        return new DirectChannel();    }    /**     * MQTT消息订阅绑定（消费者）     */    @Bean    public MessageProducer inbound() {        // 可以同时消费（订阅）多个Topic        String[] topics = new String[]{topicWarning};        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(                clientid + "_subscribe", mqttClientFactory(), topics);        adapter.setCompletionTimeout(5000);        adapter.setConverter(new DefaultPahoMessageConverter());        //设置消费者的消息级别        adapter.setQos(2);        // 设置订阅通道        adapter.setOutputChannel(mqttInboundChannel());        return adapter;    }    /**     * MQTT消息处理器（消费者）     * 持久化也是在这里处理     */    @Bean    @ServiceActivator(inputChannel = ChannelName.CHANNEL_NAME_IN)    public MessageHandler handler() {        return new MessageHandler() {            @Override            public void handleMessage(Message message) {                try {                    String topic = message.getHeaders().get("mqtt_receivedTopic").toString();                    String qos = message.getHeaders().get("mqtt_receivedQos").toString();                    String payload = message.getPayload().toString();                    LOGGER.info("主题：" + topic);                    LOGGER.info("内容：" + payload);                    LOGGER.info("级别：" + qos);                } catch (Exception e) {                    LOGGER.error(e.getMessage(), e);                }            }        };    }}

              4.调用生产者像指定通道发布消息

              import com.funfan.autoCodeDemo.common.model.RestResponse;import com.funfan.autoCodeDemo.component.mqtt.service.IMqttMessageGateway;import io.swagger.annotations.ApiOperation;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;@RestController("/test")@Slf4jpublic class TestController {    @Autowired    private IMqttMessageGateway messageGateway;    @GetMapping("/mqttTest")    @ApiOperation(value = "MQTT消息发送测试", notes = "测试")    public Object mqttTest() {        messageGateway.sendToMqtt("topic_warning","test001");        return RestResponse.success("success");    }}

              5.测试

              1.先在客户端测试一下，可以看到mqtt服务是正常的

              2.调用生产者发布消息的接口，可以看到成功发布了

              3.从日志中也可以看到之前的消息都被程序接收到了，只需要在接收位置做业务逻辑即可

              5.其他参考文章

              1. MQTT 客户端库 & SDKs
              2. 在Linux中搭建Mosquitto MQTT协议消息服务端并结合内网穿透工具实现公网访问