Spring Boot + Spring Integration整合MQTT打造双向通信客户端

1. 概述

本文分两个章节讲解MQTT相关的知识，第一部份主要讲解MQTT的原理和相关配置，第二个章节主要讲和Spring boot的integration相结合代码的具体实现，如果想快速实现功能，可直接跳过第一章节查看第二章讲。

1.1 MQTT搭建

为了实现MQTT通讯，服务端可以使用市面上常见的软件进行安装，推荐EMQX，他有配套的web页面，并且有跨平台的客户端MQTTX方便测试，当然本文主要内容就是讲解如何利用Spring boot自己写一个MQTT的客户端进行消息发布和消息订阅。这里所说的客户端MQTTX只是用来进行MQTT消息发布订阅测试，并不能和业务代码结合一起使用，结合业务内容的MQTT客户端还需自己编写，下个大章节有介绍。 

1.1.1 MQTT服务端部署

EMQX是一款实现了MQTT协议的，开源的MQTT消息代理软件。MQTT定义了消息通讯的规则和流程，而EMQX则是遵循这些规则的软件，使得设备能够依据MQTT协议进行有效通讯。

官网：https://www.emqx.com/zh

安装步骤：

1）. 下载 

wget https://www.emqx.com/zh/downloads/enterprise/5.8.5/emqx-enterprise-5.8.5-ubuntu24.04-amd64.deb 

2）安装

sudo apt install ./emqx-enterprise-5.8.5-ubuntu24.04-amd64.deb

3）启动

sudo systemctl start emqx

4）查看

sudo systemctl status emqx 

常用端口如下：

1883 TCP端口

8083 WebSocket端口

8084 WebSocket Secure端口

8883 SSL/TLS端口

18083 Broker的Dashboard访问端口

EMQX提供了一个内置的管理控制台，即EMQX Dashboard 方便用户通过web进行管理和监控EMQX集群，并配置和使用各项功能。

dashboard网址：localhost:18083  admin/public

1.1.2 MQTT客户端部署

MQTTX是EMQX开源的一款跨平台MQTT客户端工具，包含三种类型工具：

MQTTX Desktop：MQTTX Desktop 是一款跨平台的MQTT桌面客户端工具

MQTTX CLI：MQTTX CLI是EMQ开源的一款MQTT5.0命令行客户端工具

MQTT Web: MQTTX Web是一款基于浏览器访问客户端工具

官网：https://mqttx.app/zh

后续用Spring boot自己写客户端，和1.1.1章节安装的服务端进行连接通讯测试，最好安装一下这个客户端，这样比较好测试，用这个客户端作为发布者发数据，用java写代码作为订阅端收数据，反之亦然。

1.2 MQTT报文

剩余长度：剩余长度指示了当前报文剩余部分的字节数，也就是可变报头和有效载荷这两部份的长度。

报文的总长度：固定报头的长度 + 剩余长度

有效载荷：在publish报文中，payload用于承载具体的应用消息内容，这也是publish报文最核心的功能。在subscribe报文中，payload包含了想要订阅的主题以及对应的订阅选项，这也是subscribe报文最主要的功能。

1.3 Qos介绍

Qos 三个常见取值的应用场景：

0：可能会丢数据，消息丢失的频率依赖所处的网络环境，传递效率最高，传输一些高频且不那么重要的数据，比如周期性更新传感器数据。

即发即弃，不需要等待确认，不需要存储和重传，接收端永远不会接收到重复报文。

1: 保证消息到达不丢失，但可能会导致消息重复，传输一些较为重要的数据，比如下达关键指令。

如果发送报文失败或者应答报文失败，都会导致报文重传，应答报文失败会导致重复数据。packetID是本次报文的唯一标识，DUP为是否重传的标识。

如果客户端收到 DUP=1 的消息，但第一次接收的消息已经处理完毕并删除了 Packet ID 记录，那么 DUP 这个标志本身就没法帮助客户端去重。
 

2: 既可以保证消息到达，也可以保证消息不会重复，但传输成本最高，在金融、航空等行业场景下使用。通过对packet ID进行标记，保证了消息传输不重复。

packet ID需要进行锁定和释放放回，而不是一直生成新的：

防止 Packet ID 过快消耗：MQTT 规定 Packet ID 范围为 1~65535，如果不控制使用，短时间内可能会用尽。
保证 ID 唯一性：如果 ID 生成过快，可能会导致不同的 PUBLISH 消息误用相同的 ID，从而导致数据错误。
确保 PUBACK 逻辑正确：QoS 1 需要等到 PUBACK 确认 后才能释放 ID，避免 ID 污染。

1.4 主题

1.4.1 主题通配符

MQTT主题通配符包括单层通配符 + 以及多层通配符 #，主要用于客户端一次订阅多个主题

单层通配符：加号（"+"）是用于单个主题层级匹配的通配符，在使用单层通配符时，单层通配符必须占据整个层级。

test/+ : test/1 , test/2, test/any

test/+/topic : test/1/topic , test/2/topic ,test/any/topic

多层通配符: 符号（"#"）用于匹配主题中任意层级的通配符，多层通配符表示它的父级和任意数量的子层级，在使用多层通配符时，它必须占据整个层级并且必须时主题的最后一个字符。

# ： 匹配所有主题，用于搭建MQTT服务端集群用到

test/topic/# : test/topic/1 , test/topic/1/2

1.4.2 系统主题

以$SYS/ 开头的主题为系统主题，系统主题主要是用于获取MQTT服务器自身运行状态，消息统计，客户端上下线事件等数据。

集群状态信息

主题	说明
$SYS/brokers	集群节点列表
$SYS/brokers/${node}/version	EMQX 版本
$SYS/brokers/${node}/uptime	EMQX 运行时间
$SYS/brokers/${node}/datetime	EMQX 系统时间
$SYS/brokers/${node}/sysdescr	EMQX 系统信息

客户端上下线事件

$SYS 主题前缀：$SYS/brokers/${node}/clients/ 

主题 (Topic)	说明
${clientid}/connected	上线事件。当任意客户端上线时，EMQX 就会发布该主题的消息
${clientid}/disconnected	下线事件。当任意客户端下线时，EMQX 就会发布该主题的消息

 更多内容请参考官方文档：

系统主题 | EMQX 5.0 文档

1.5 会话介绍

        MQTT客户端和MQTT服务器之间的连接被称为会话，每个MQTT客户端都可以启动一个或多个会话，通过会话可以实现客户端和服务器之间的消息传递。服务端使用client ID来唯一标识每个会话，如果客户端想要在连接时复用之前的会话，那么必须使用与此前一致的client ID

1.5.1 clean start 参数配置

clean start ：用于指示客户端在和服务器建立连接的时候应该尝试恢复之前的会话还是直接创建全新的会话。

等于0: 服务端存在一个关联此客户端标识符client ID的会话，服务端必须基于此会话的状态恢复与客户端的通信，之前的订阅信息会再次绑定，并且会接收到客户端断开时发布者发布的消息。如果不存在任何关联此客户端标识符的会话，服务端必须创建一个新的会话。

等于1: 客户端和服务端必须丢弃任何一存在的会话，并开始一个新的会话。

1.5.2 session expiry interval 参数配置

session expiry interval: 决定了会话状态数据在服务端的存储时长。

没有指定此属性或设置为0： 表示会话将在网络连接断开时立即结束。

设置为一个大于0的值：则表示会话将在网络连接断开的多少秒之后过期。

设置为0xFFFFFFF：session expiry interval 属性能够设置的最大值时，表示会话数据永不过期。

1.6 消息详解

1.6.1 保留消息

普通消息：普通消息在发送之前所对应的主题如果不存在订阅者，普通消息MQTT服务器会直接将其丢弃。

保留消息：保留消息可以保留MQTT服务器中，任何新的订阅者订阅与该保留消息中的主题匹配的主题时，都会立即接收到该消息，即使这个消息是在他们订阅主题之前发布的。

常用场景：

a. 智能家居设备的状态只有在变更时才会上报，但是控制端需要在上线后就能获取到设备的状态；

b. 传感器上报数据的间隔时间太长，但是订阅者需要在订阅后立即获取到最新的数据；

c. 传感器的版本号，序列号等不经常变更的属性，可在上线后发布一条保留消息告知后续的所有订阅者。

注意：

a. 发布者发布保留消息，保留消息针对某一个主题，最多只能发布一个保留消息，即最后的一条保留消息。

b. 保留消息存储在服务端的默认存储方式是内存存储，可以设置成磁盘存储。

c. 通过发送一条空的保留消息进行覆盖上一条保留消息，达成保留消息删除的效果

d. 通过dashboard页面进行删除按钮删除

1.6.2 消息过期时间

 设置消息过期时间后，如果订阅者在消息过期前连接并订阅了相关主题，则能够收到消息；否则，将无法收到消息。当订阅者收到设置过期时间的消息时，消息会携带过期时间信息，根据接收数据的当前时间事实计算而得。

1.6.3 遗嘱消息

在MQTT 中，客户端可以在连接时在服务端注册一个遗嘱消息，当客户端意外断开连接，服务端会向其他订阅了相应主题的客户端发送此遗嘱消息。这些订阅者可以向用户发送通知，切换备用设备等。正常关闭断开不会触发遗嘱消息。

will delay interval 这个属性决定了服务端与发布者网络连接断开后多久发布遗嘱消息，单位秒。如果发布者在延迟时间内恢复连接，遗嘱消息将不会被发布。这是为了避免发布者连续短暂网络反复中断而频繁发布遗嘱消息。

如果会话时间快要过期了，但遗嘱消息延迟时间还未到达，此时也会发布遗嘱消息。

1.6.4 延迟发布

延迟发布主题格式 $delayed/{DelayInterval}/{TopicName} 

$delayed：使用$delayed作为主题的前缀的消息都将被视为需要延迟发布的消息

DelayInterval：延迟发布的时间间隔，单位秒，允许最大值4294967秒（50天）

TopicName：MQTT消息主题名称

1.7 订阅详解 

1.7.1 订阅配置

1）Qos

服务端在向订阅端发送消息时可以使用的最大Qos等级

情况1: 服务端支持的最大Qos < 客户端订阅时请求的最大Qos

服务端将无法满足客户端的要求，这时服务端就会通过订阅的响应报文SUBACK告知订阅端最终授予的最大Qos等级，订阅端可以自行评估是否接受并继续通信，此情况订阅端依然能收到消息，只不过Qos不是订阅端配置的等级，而是发布端配置的等级，降级处理。

情况2：订阅时请求的最大Qos < 消息发布时的Qos

为了尽可能地投递消息，服务端不会忽略这些消息，而是会在转发时对这些消息对Qos进行降级处理。

2）no local

被用在桥接场景中，桥接的本质是两个MQTT server建立一个MQTT连接，然后相互订阅一些主题，server将客户端的消息转发给另一个server，而另一个server则可以将消息转发给它的客户端。为了避免server相互无限循环转发的转发风暴，两个server均将此选项设置为1即可避免。

等于0（默认）: 服务端可以将消息转发给发布这个消息的客户端

等于1: 服务端不可以将消息转发给发布这个消息的客户端，禁止本地转发 

3）retain as published （RAP）

为了解决桥接场景下的问题，当server A将保留消息转发给server B时，由于消息中的retain标识被删除，server B将不会知道这原本是一条保留消息，自然不会再存储它，这就导致了保留消息无法跨桥接使用。

等于0（默认）：服务端在向此订阅转发应用消息时需要清楚消息中的retain标识不变

等于1: 服务端在向此订阅转发应用消息时需要保持消息中的retain标识不变

4）retain handling

虽然发布者有配置保留消息，订阅者可以选择是否接受这个保留消息，这个配置是被用来向服务端指示当订阅建立时，是否需要发送保留消息。

等于0（默认）：表示只要订阅建立，就发送保留消息

等于1：表示只有建立全新的订阅而不是重复订阅时，才发送保留消息

等于2：表示订阅建立时不要发送保留消息

1.7.2 共享订阅

普通订阅者：发布者每发布一条消息时，所有匹配的订阅端都会收到该消息的副本，当某个订阅者的消费速度无法跟上消息的生产速度时，broker没办法将其中一部份消息分流到其他订阅端中来分担压力，这使得订阅端容易成为整个消息系统的性能瓶颈。

共享订阅者：共享订阅者可以均衡的分配消息负载，各个客户端共享一个订阅，每个匹配该订阅的消息都会有一个副本投递给其中一个客户端。提高了吞吐量，带来了高可用，即使共享订阅组中的一个客户端断开连接或发生故障，其他客户端仍然可以继续处理消息。

带群组共享订阅：通过在原始主题前添加 $share/<group-name> 前缀为分组的订阅者启用共享订阅。组名可以是任意字符串。broker同时将消息转发给不同的组，属于同一组的订阅者可以使用负载均衡接收消息。

 不带群组共享订阅：以$queue/为前缀的共享订阅是不带群组的共享订阅，它是带群组共享订阅的特例，可以理解为所有共享订阅者都在一个订阅群组中。

负载均衡算法：

1. 随机（random）：在共享订阅组内随机选择一个会话发送消息（推荐）

2. 轮询（round robin）：在共享订阅组内按顺序选择一个会话发送消息，循环往复

3. 哈希（hash）：基于某个字段的哈希结果来分配

4.粘性（sticky）：在共享订阅组内随机选择一个会话发送消息，此后保持这一选择，直到该会话结束再重复这一过程，有点像是备用机的感觉。

5. 本地优先（local）：随机选择，但优先选择与消费的发布者同处于同一节点的会话，如果不存在这样的会话，则退化为普通的随机策略。 

1.7.3 排它订阅

排它订阅允许对主题进行互斥订阅，一个主题同一时刻仅被允许存在一个订阅者，在当前订阅者未取消订阅前，其他订阅者都将无法订阅对应主题。要进行排它订阅，需要为主题名称添加：

$exclusive/ 前缀

注意：

1. broker端需要开启排它订阅

2. 针对订阅者进行设置，发布者正常配置topic即可

3. 即使有订阅者对topic进行了排它订阅，其他订阅者依然可以对此topic进行正常订阅，排它订阅仅针对另一个排它订阅生效排它。

1.7.4 自动订阅

通过dashboard设置自动订阅主题，所有客户端包括发布者都会收到这个主题的消息

2.  Spring boot 实战

2.1 准备工作

2.1.1 添加依赖

<!-- Spring Integration Core --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-core</artifactId>
<!--            <version>5.5.10</version>--></dependency>
<!-- Spring Integration MQTT -->
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId><version>5.5.10</version>
</dependency>

注意：这里的spring-integration-core千万别指定版本号，不然在开发中，会遇到接口加了@MessagingGateway注解，接口旁边没有bean的标志，无法被Spring管理和注入使用。

 2.1.2 添加配置

spring:# Mqtt配置mqtt:username:password:url: tcp://192.168.1.214:1883subClientId: wuLang_subClient_01subTopic: worker/alert # 订阅多个主题 用逗号隔开pubTopic: worker/locationpubClientId: wuLang_pubClient_01

2.2 创建配置类

2.2.1 参数配置类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;/*** @Author: HarryLin* @Date: 2025/3/20  14:32* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Data
@ConfigurationProperties(prefix = "spring.mqtt")
public class MqttPropertiesConfig {private String username;private String password;private String url;private String subClientId;private String subTopic;private String pubClientId;private String pubTopic;
}

2.2.2 创建连接工厂 

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;/*** @Author: HarryLin* @Date: 2025/3/20  14:40* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Configuration
public class MqttConfiguration {@Autowiredprivate MqttPropertiesConfig mqttPropertiesConfig;/**  创建连接工厂 **/@Beanpublic MqttPahoClientFactory mqttPahoClientFactory(){DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();options.setCleanSession(true); //设置新会话options.setUserName(mqttPropertiesConfig.getUsername());options.setPassword(mqttPropertiesConfig.getPassword().toCharArray());options.setServerURIs(new String[]{mqttPropertiesConfig.getUrl()});factory.setConnectionOptions(options);return factory;}}

2.2.3 配置入站适配器

import com.wulang.pnt.handler.mqttservice.ReceiverMessageHandler;
import org.springframework.beans.factory.annotation.Autowired;
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.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;/*** @Author: HarryLin* @Date: 2025/3/20  14:54* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Configuration
public class MqttInboundConfiguration {@Autowiredprivate MqttPropertiesConfig mqttPropertiesConfig;@Autowiredprivate MqttPahoClientFactory mqttPahoClientFactory;@Autowiredprivate ReceiverMessageHandler receiverMessageHandler;//消息通道@Beanpublic MessageChannel messageInboundChannel(){return new DirectChannel();}/*** 配置入站适配器* 作用： 设置订阅主题，以及指定消息的通道 等相关属性* */@Beanpublic MessageProducer messageProducer(){MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter = new MqttPahoMessageDrivenChannelAdapter(mqttPropertiesConfig.getUrl(),mqttPropertiesConfig.getSubClientId(),mqttPahoClientFactory,mqttPropertiesConfig.getSubTopic().split(","));mqttPahoMessageDrivenChannelAdapter.setQos(1);mqttPahoMessageDrivenChannelAdapter.setConverter(new DefaultPahoMessageConverter());mqttPahoMessageDrivenChannelAdapter.setOutputChannel(messageInboundChannel());return mqttPahoMessageDrivenChannelAdapter;}/**  指定处理消息来自哪个通道 */@Bean@ServiceActivator(inputChannel = "messageInboundChannel")public MessageHandler messageHandler(){return receiverMessageHandler;}
}

2.2.4 配置出站适配器

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
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.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;/*** @Author: HarryLin* @Date: 2025/3/20  15:46* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Configuration
@Slf4j
public class MqttOutboundConfiguration {@Autowiredprivate MqttPropertiesConfig mqttPropertiesConfig;@Autowiredprivate MqttPahoClientFactory mqttPahoClientFactory;// 消息通道@Beanpublic MessageChannel mqttOutboundChannel(){return new DirectChannel();}/** 配置出站消息处理器 */@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")  // 指定处理器针对哪个通道的消息进行处理public MessageHandler mqttOutboundMessageHandler(){MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(mqttPropertiesConfig.getUrl(),mqttPropertiesConfig.getPubClientId(),mqttPahoClientFactory);mqttPahoMessageHandler.setDefaultQos(1);mqttPahoMessageHandler.setDefaultTopic("worker/location");mqttPahoMessageHandler.setAsync(true);return mqttPahoMessageHandler;}}

 2.2.5 配置出站网关

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;/*** @Author: HarryLin* @Date: 2025/3/20  17:06* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, String payload);public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, @Header(value = MqttHeaders.QOS) int qos, String payload );
}

2.3 收发数据

2.3.1 订阅数据

在入站适配器中配置了订阅主题，MQTT服务端等信息后，并将以下类注入到入站适配器到消息处理方法中（2.2.3 最后一个方法），在这个方法中就能得到订阅消息。只需用户在此方法中写消息处理方法即可。

import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Service;import java.util.Objects;/*** @Author: HarryLin* @Date: 2025/3/20  15:24* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Service
@Slf4j
public class ReceiverMessageHandler implements MessageHandler {@Overridepublic void handleMessage(Message<?> message) throws MessagingException{Object payload = message.getPayload();MessageHeaders headers = message.getHeaders();String receivedTopic = Objects.requireNonNull(headers.get("mqtt_receivedTopic")).toString();String receivedQos = Objects.requireNonNull(headers.get("mqtt_receivedQos")).toString();String timestamp = Objects.requireNonNull(headers.get("timestamp")).toString();log.info("MQTT payload= {} \n receivedTopic = {} \n receivedQos = {} \n timestamp = {}",payload,receivedTopic,receivedQos,timestamp);}
}

2.3.2 发布消息

封装出站网关，调用如下方法即可发送消息

import com.wulang.pnt.config.mqtt.MqttGateway;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;/*** @Author: HarryLin* @Date: 2025/3/20  16:16* @Company: 北京红山信息科技研究院有限公司* @Email: linyun@***.com.cn**/
@Service
public class MqttMessageSender{@Autowiredprivate MqttGateway mqttGateway;public void sendMsg(@Header(value = MqttHeaders.TOPIC) String topic, String payload) {mqttGateway.sendMsgToMqtt(topic,payload);}public void sendMsg(@Header(value = MqttHeaders.TOPIC) String topic, @Header(value = MqttHeaders.QOS) int qos, String payload) {mqttGateway.sendMsgToMqtt(topic,qos,payload);}
}

@SpringBootTest
@Slf4j
public class MqttClientTest {@Autowiredprivate MqttGateway mqttGateway;@Testpublic void sendMsg(){mqttGateway.sendMsgToMqtt("worker/location","hello mqtt spring boot");log.info("message is send");}