SpringBoot集成MQTT客户端

📅 发布时间：2026/7/6 22:47:30 👁️ 浏览次数：

SpringBoot集成MQTT客户端SpringBoot集成MQTT客户端一、消息中间件1.1、下载中间件1.2、启动服务1.3、基本设置1.3.1、配置客户端认证1.3.2、通用配置二、客户端工具2.1、配置连接2.2、配置订阅主题2.3、测试三、代码演示3.1、导入依赖3.2、配置文件3.3、核心配置类3.4、常量类3.5、消费者配置类3.4、生产者配置类3.6、消费者处理类3.7、生产者处理类四、测试4.1、测试接受消息4.2、测试发送消息五、2024-12-06新增想法实现关于配置使用配置文件配置多个订阅主题避免硬编码5.1、修改yml配置文件5.2、修改核心配置类5.3、修改消费者配置类5.4、启动项目测试修改本文运行环境win10、jdk11、springboot 2.7.18一、消息中间件下载支持mqtt协议的中间件市面上许多支持mqtt协议的中间件比如Mosquitto、RabbitMQ、EMQ X、HiveMQ、ActiveMQ等本文选择EMQ X 5.3.2。ps5.3.2是EMQ X最后一个支持windows的版本。1.1、下载中间件下载链接https://packages.emqx.io/emqx-ce/v5.3.2/emqx-5.3.2-windows-amd64.zip1.2、启动服务①下载解压后进入安装目录.emqx-5.3.2-windows-amd64in在地址栏输入cmd执行命令emqx start启动服务。为了以后方便启动EMQ X服务可以使用NSSM将EMQ X封装成windows服务。NSSM是一个服务封装程序它可以将普通exe程序或者.bat文件等封装成windows服务操作起来也比较简单有兴趣可以去了解下这里就不演示了。②服务启动后打开浏览器输入http://127.0.0.1:18083进入EMQ X后台管理页面。默认登录账号admin默认登录密码public③这里就进入EMQ X后台管理页面了1.3、基本设置1.3.1、配置客户端认证因为EMQ X默认是没有配置连接用户的也就是说任何人都可以通过tcp去访问当我们EMQ X服务①创建客户端认证访问控制——客户端任务——创建②选择认证方式为了演示方便这里我选择密码认证的方式选择后点击下一步③选择存储认证数据的数据源这里也有挺多数据源选择为了演示方便这里我选择EMQ X内部的数据库④配置参数账号类型可以选择clientId或者username两种方式都可以用其他的默认就行⑤新增用户点击完成后会出现一条配置信息点击用户管理新建连接用户⑥添加用户填写用户名和密码我把成为超级管理员勾上了方便我展示嘻嘻⑦这样我们把连接用户建好了咯1.3.2、通用配置这个我就不管了各位根据需要自行设置到此消息中间件部分就结束咯二、客户端工具既然选择了EMQ的EMQ X那么客户端工具也选择EMQ的MQTTX咯。下载链接https://mqttx.app/zh/downloads?oswindows下载安装完后打开MQTTX2.1、配置连接①这里可以设置中文②配置连接1、服务器地址因为是本地服务所以用127.0.0.12、端口默认是1883我没改所以是18833、Client ID这个是客户端的唯一标识就是说这个id是区分连接用户的4、用户名密码就是刚刚1.3.1配置的认证用户5、把自动重连勾掉因为如果你的连接信息填错他会一直重连很烦。6、填写完后点击右上角连接③这样就连接上咯再打开EMQ X 后台管理页面出现了我们配置的连接客户端id就是刚刚配置的Client ID2.2、配置订阅主题关于topic的配置1、主题层级分隔符—“/”用于分割主题层级2、单层通配符—“”单层通配符只能匹配一层主题3、多层通配符—“#”多层通配符多层通配符可以匹配于多层主题关于Qos配置这个配置是保证消息的可靠性的Qos 0我把一条消息丢给你我就不管你收到还没没收到。Qos 1我把一条消息丢给你你就要给我回复收到不然我就一直发。Qos 2我把一条消息丢给你你一定要给我回复一次收到。这里跟Qos 1的区别是Qos 1可能会让你接收到重复的消息而Qos 2确保一条消息只被你接收到一次。描述比较抽象总之Qos等级越高对于资源消耗就越高数据的可靠性也越高我这里就不过多赘述了。订阅好后我们的EMQ X后台管理页面也出现哪个客户端订阅了哪个主题。2.3、测试第一步填写我们要发送消息的主题这里填的主题要和我们刚刚订阅的主题要能对应上第二步填写我们要发送的消息这里我选择的json格式和Qos 1第三步点击发送对话框右边背景纯绿色的就是我们刚刚发送的消息而左边黄色线圈起来的就是我们的订阅主题收到的消息到此EMQ X服务准备工作就完成啦进入代码模块咯。三、代码演示3.1、导入依赖!-- Springboot父依赖 -- parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version relativePath/ /parent dependencies !-- mqtt相关依赖 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-stream/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency !-- springBoot web -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- springBoot test -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId /dependency !-- Apache Lang3 -- dependency groupIdorg.apache.commons/groupId artifactIdcommons-lang3/artifactId /dependency !-- JSON -- dependency groupIdcom.alibaba.fastjson2/groupId artifactIdfastjson2/artifactId version2.0.43/version /dependency !-- lombok -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId version1.18.22/version /dependency /dependencies3.2、配置文件我这里是用的.yml格式的--- # mqtt配置 spring: mqtt: # 账号 username: 请填写自己配置的账号 # 密码 password: 请填写自己配置的密码 # mqtt连接tcp地址 host-url: tcp://127.0.0.1:1883 # 客户端Id每个启动的id要不同 client-id: jyy # 默认主题 default-topic: jyy/# # 超时时间 timeout: 100 # 保持连接数 keepalive: 1003.3、核心配置类这个代码注入配置信息构建客户端工厂创建连接的package com.jyy.common.mqtt.config; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; /** * mqtt相关配置信息 * * author jyy * date 2024-11-26 */ Slf4j Setter Getter Configuration // 识别配置文件中spring.mqtt下面的属性 ConfigurationProperties(spring.mqtt) public class MqttConfiguration { /** * 用户名 */ private String username; /** * 密码 */ private String password; /** * 连接地址 */ private String hostUrl; /** * 客户Id */ private String clientId; /** * 默认连接话题 */ private String defaultTopic; /** * 超时时间 */ private int timeout; /** * 保持连接数 */ private int keepalive; /** * 客户端与服务器之间的连接意外中断服务器将发布客户端的“遗嘱”消息 */ private static final byte[] WILL_DATA offline.getBytes(); /** * 注册MQTT客户端工厂 * * return MqttPahoClientFactory */ Bean public MqttPahoClientFactory mqttClientFactory() { // 客户端工厂 DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); // 连接配置 MqttConnectOptions options new MqttConnectOptions(); // 设置连接的用户名 options.setUserName(username); // 设置连接的密码 options.setPassword(password.toCharArray()); // 设置连接的地址 options.setServerURIs(new String[]{hostUrl}); // 如果设置为 false客户端和服务器将在客户端、服务器和连接重新启动时保持状态。随着状态的保持 // 即使客户端、服务器或连接重新启动消息传递也将可靠地满足指定的 QOS。服务器将订阅视为持久的。 // 如果设置为 true客户端和服务器将不会在客户端、服务器或连接重新启动时保持状态。 options.setCleanSession(true); // 设置超时时间该值以秒为单位必须0定义了客户端等待与 MQTT 服务器建立网络连接的最大时间间隔。 // 默认超时为 30 秒。值 0 禁用超时处理这意味着客户端将等待直到网络连接成功或失败。 options.setConnectionTimeout(10); // 设置会话心跳时间单位为秒服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线 // 此值以秒为单位定义发送或接收消息之间的最大时间间隔必须0 // 但这个方法并没有重连的机制 options.setKeepAliveInterval(20); // 设置“遗嘱”消息的话题若客户端与服务器之间的连接意外中断服务器将发布客户端的“遗嘱”消息。 options.setWill(willTopic, WILL_DATA, 2, false); //自动重新连接 options.setAutomaticReconnect(true); factory.setConnectionOptions(options); log.info(初始化 MQTT 配置); return factory; } }3.4、常量类package com.jyy.common.mqtt.constants; /** * 常量 * * author jyy * date 2024-11-26 */ public class MqttConstant { /** * 客户端id消费者后缀 */ public static final String CLIENT_SUFFIX_CONSUMERS _consumers; /** * 客户端id生产者后缀 */ public static final String CLIENT_SUFFIX_PRODUCERS _producers; }3.5、消费者配置类订阅主题在这里配置package com.jyy.common.mqtt.config; import com.jyy.common.mqtt.constants.MqttConstant; import com.jyy.common.mqtt.handler.MqttMessageReceiver; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; 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.endpoint.MessageProducerSupport; 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; import javax.annotation.Resource; /** * 消费者配置 * * author jyy * date 2024-11-26 */ Slf4j AllArgsConstructor Configuration IntegrationComponentScan public class MqttInboundConfiguration { /** * 注入核心配置类 */ Resource private MqttConfiguration mqttConfiguration; private MqttMessageReceiver mqttMessageReceiver; /** * 此处可以使用其他消息通道 * MQTT信息通道消费者 * Spring Integration默认的消息通道它允许将消息发送给一个订阅者然后阻碍发送直到消息被接收。 */ Bean public MessageChannel mqttInBoundChannel() { return new DirectChannel(); } /** * mqtt入站消息处理工具对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。 */ Bean ServiceActivator(inputChannel mqttInBoundChannel) public MessageHandler mqttMessageHandler() { return this.mqttMessageReceiver; } /** * MQTT消息订阅绑定消费者 * 适配器, 两个topic共用一个adapter * 客户端作为消费者订阅主题消费消息 */ Bean public MessageProducerSupport mqttInbound() { String clientId mqttConfiguration.getClientId(); String defaultTopic mqttConfiguration.getDefaultTopic(); MqttPahoClientFactory mqttPahoClientFactory mqttConfiguration.mqttClientFactory(); // Paho客户端消息驱动通道适配器主要用来订阅主题 MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter( clientId MqttConstant.CLIENT_SUFFIX_CONSUMERS, mqttPahoClientFactory, // TODO 这后面填写需要订阅的主题这里是一个可变参数可以填写多个用逗号隔开。例如我再订阅一个yy/的主题 defaultTopic, yy/ ); adapter.setCompletionTimeout(5000); // Paho消息转换器 DefaultPahoMessageConverter defaultPahoMessageConverter new DefaultPahoMessageConverter(); // 按字节接收消息 // defaultPahoMessageConverter.setPayloadAsBytes(true); adapter.setConverter(defaultPahoMessageConverter); // 设置QoS adapter.setQos(1); // 设置订阅通道 adapter.setOutputChannel(mqttInBoundChannel()); return adapter; } }3.4、生产者配置类package com.jyy.common.mqtt.config; import com.jyy.common.mqtt.constants.MqttConstant; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; 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.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import javax.annotation.Resource; /** * 生产者配置 * * author jyy * date 2024-11-26 */ Slf4j Configuration AllArgsConstructor public class MqttOutboundConfiguration { Resource private MqttConfiguration mqttConfiguration; /** * MQTT信息通道生产者 */ Bean public MessageChannel mqttOutboundChannel() { return new DirectChannel(); } /** * MQTT消息处理器生产者 */ Bean ServiceActivator(inputChannel mqttOutboundChannel) public MessageHandler mqttOutbound() { // 客户端id String clientId mqttConfiguration.getClientId(); // 默认主题 String defaultTopic mqttConfiguration.getDefaultTopic(); MqttPahoClientFactory mqttPahoClientFactory mqttConfiguration.mqttClientFactory(); // 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory MqttPahoMessageHandler messageHandler new MqttPahoMessageHandler(clientId MqttConstant.CLIENT_PREFIX_PRODUCERS, mqttPahoClientFactory); // true异步发送消息时将不会阻塞。 messageHandler.setAsync(true); messageHandler.setDefaultTopic(defaultTopic); // 默认QoS messageHandler.setDefaultQos(1); // Paho消息转换器 DefaultPahoMessageConverter defaultPahoMessageConverter new DefaultPahoMessageConverter(); // defaultPahoMessageConverter.setPayloadAsBytes(true); // 发送默认按字节类型发送消息 messageHandler.setConverter(defaultPahoMessageConverter); return messageHandler; } }注意我们只配置了一个clientId所以生产者和消费者用的同一个clientId接收消息的时候不会报错但是发送消息的时候就会报错虽然消息是发出来但是毕竟报错看着不舒服是不。所以在配置生产者和消费者配置类的时候这里我用了常量类来区分两个clientId。在消费者配置类中配置了clientId MqttConstant.CLIENT_SUFFIX_CONSUMERS在生产者配置类中配置了clientId MqttConstant.CLIENT_SUFFIX_PRODUCERS这样就避免这个报错了。3.6、消费者处理类这里是处理接收到的消息的处理类可以根据不同主题分别处理package com.jyy.common.mqtt.handler; import lombok.extern.slf4j.Slf4j; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.stereotype.Component; /** * 消费者处理器 * * author jyy * date 2024-11-26 */ Slf4j Component public class MqttMessageReceiver implements MessageHandler { /** * 消息处理 * * param message 消息 * throws MessagingException 消息异常 */ Override public void handleMessage(Message? message) throws MessagingException { try { // 获取消息Topic MessageHeaders headers message.getHeaders(); String topic headers.get(MqttHeaders.RECEIVED_TOPIC).toString(); log.info([获取到的消息的topic]{} , topic); // 获取消息体 String payload (String) message.getPayload(); log.info([获取到的消息的payload]{} , payload); // 根据主题分别进行消息处理 if (topic.contains(jyy/)) { log.info(接收到jyy/的消息啦快去处理); // TODO 数据处理 payload } // ...... // TODO else if匹配其他题主为了演示方便这里用的String的contains()方法匹配主题还可以用String的matches()方法匹配正则表达式 } catch (Exception e) { log.error(e.toString()); } } }3.7、生产者处理类这里有两个class一个是接了生产者配置类的一个是封装了生产者发送消息的方法。package com.jyy.common.mqtt.handler; 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; /** * 生产者处理器 * * author jyy * date 2024-11-26 */ Component MessagingGateway(defaultRequestChannel mqttOutboundChannel) public interface MqttGateway { /** * 发送mqtt消息 * * param topic 主题 * param payload 内容 */ void sendToMqtt(Header(MqttHeaders.TOPIC) String topic, String payload); /** * 发送包含qos的消息 * * param topic 主题 * param qos 对消息处理的几种机制。 * * 0 表示的是订阅者没收到消息不会再次发送消息会丢失。br * * 1 表示的是会尝试重试一直到接收到消息但这种情况可能导致订阅者收到多次重复消息。br * * 2 多了一次去重的动作确保订阅者收到的消息有一次。 * param payload 消息体 */ void sendToMqtt(Header(MqttHeaders.TOPIC) String topic, Header(MqttHeaders.QOS) int qos, String payload); /** * 发送包含qos的消息 * * param topic 主题 * param qos 对消息处理的几种机制。 * * 0 表示的是订阅者没收到消息不会再次发送消息会丢失。br * * 1 表示的是会尝试重试一直到接收到消息但这种情况可能导致订阅者收到多次重复消息。br * * 2 多了一次去重的动作确保订阅者收到的消息有一次。 * param payload 消息体 */ void sendToMqtt(Header(MqttHeaders.TOPIC) String topic, Header(MqttHeaders.QOS) int qos, byte[] payload); } package com.jyy.common.mqtt.handler; import com.alibaba.fastjson2.JSONObject; import org.springframework.stereotype.Component; import javax.annotation.Resource; /** * 生产者处理器 * * author jyy * date 2024-11-26 */ Component public class MqttMessageSender { Resource private MqttGateway mqttGateway; /** * 发送mqtt消息 * * param topic 主题 * param message 内容 * return void */ public void send(String topic, String message) { mqttGateway.sendToMqtt(topic, message); } /** * 发送包含qos的消息 * * param topic 主题 * param qos 质量 * param messageBody 消息体 * return void */ public void send(String topic, int qos, JSONObject messageBody) { mqttGateway.sendToMqtt(topic, qos, messageBody.toString()); } /** * 发送包含qos的消息 * * param topic 主题 * param qos 质量 * param message 消息体 * return void */ public void send(String topic, int qos, byte[] message) { mqttGateway.sendToMqtt(topic, qos, message); } }四、测试4.1、测试接受消息①启动项目发现我们的初始化日志在控制台打印了我们在核心代码类中写的初始化mqtt配置日志。②再看EMQ X后台管理页面的客户端发现我们的消费者也准备就绪咯③再看订阅管理也有我们的消费者④使用MQTTX向订阅的主题发送消息⑤在回到idea中看控制台发现我们能接收到订阅的主题’jyy/#其他客户端向这个主题发送的消息至此消费者功能完成4.2、测试发送消息①写一个测试类向主题jyy/1发送消息这里写了个死循环因为测试代码执行完后会结束进程不方便我演示。package com.jyy.common; import com.jyy.common.mqtt.handler.MqttMessageSender; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; import javax.annotation.Resource; /** * 测试mqtt * * author jyy * date 2024-11-28 */ SpringBootTest public class MqttTest { Resource private MqttMessageSender mqttMessageSender; Test public void test() { mqttMessageSender.send(jyy/1, 1, hello world.getBytes()); // TODO while (true){ } } }②执行测试代码后发现我们的生产者也准备就绪咯③再看MQTTX的会话窗口我们订阅的jyy/#主题成功接收到了我们写的测试类发送的消息证明我们的生产者功能也是没问题的。至此我们的演示就到这里咯相信聪明的你肯定学会了springboot集成mqtt。五、2024-12-06新增想法实现关于配置使用配置文件配置多个订阅主题避免硬编码5.1、修改yml配置文件相比于之前的配置文件新增了 topics 属性--- # mqtt配置 spring: mqtt: # 账号 username: jyy # 密码 password: 请填写自己配置的密码 # mqtt连接tcp地址 host-url: tcp://127.0.0.1:1883 # 客户端Id每个启动的id要不同 client-id: jyy # 默认主题 default-topic: jyy/# # 超时时间 timeout: 100 # 保持连接数 keepalive: 100 # 2024-12-06新增要订阅的其他主题 topics: - yy/# - y/5.2、修改核心配置类相比于之前的核心配置类这里新增了 private List topics; 属性和 getAllTopics()方法。package com.jyy.common.mqtt.config; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.util.CollectionUtils; import java.util.ArrayList; import java.util.List; /** * mqtt相关配置信息 * * author jyy * date 2024-11-26 */ Slf4j Setter Getter Configuration ConfigurationProperties(spring.mqtt) public class MqttConfiguration { /** * 用户名 */ private String username; /** * 密码 */ private String password; /** * 连接地址 */ private String hostUrl; /** * 客户Id */ private String clientId; /** * 默认连接话题 */ private String defaultTopic; /** * 超时时间 */ private int timeout; /** * 保持连接数 */ private int keepalive; /** * 2024-12-06新增 * 要订阅的其他主题 */ private ListString topics; /** * 客户端与服务器之间的连接意外中断服务器将发布客户端的“遗嘱”消息 */ private static final byte[] WILL_DATA offline.getBytes(); /** * 2024-12-06新增 * 获取所有订阅的主题 * * return 所有订阅的主题 */ public String[] getAllTopics() { // 校验配置文件是否配置 if (CollectionUtils.isEmpty(topics)) { this.topics new ArrayList(); } // 将默认主题条件到其他主题里 this.topics.add(defaultTopic); // 返回主题数组 return topics.toArray(new String[0]); } /** * 注册MQTT客户端工厂 * * return MqttPahoClientFactory */ Bean public MqttPahoClientFactory mqttClientFactory() { // 客户端工厂 DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); // 设置连接的用户名 options.setUserName(username); // 设置连接的密码 options.setPassword(password.toCharArray()); // 设置连接的地址 options.setServerURIs(new String[]{hostUrl}); // 如果设置为 false客户端和服务器将在客户端、服务器和连接重新启动时保持状态。随着状态的保持 // 即使客户端、服务器或连接重新启动消息传递也将可靠地满足指定的 QOS。服务器将订阅视为持久的。 // 如果设置为 true客户端和服务器将不会在客户端、服务器或连接重新启动时保持状态。 options.setCleanSession(true); // 设置超时时间该值以秒为单位必须0定义了客户端等待与 MQTT 服务器建立网络连接的最大时间间隔。 // 默认超时为 30 秒。值 0 禁用超时处理这意味着客户端将等待直到网络连接成功或失败。 options.setConnectionTimeout(10); // 设置会话心跳时间单位为秒服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线 // 此值以秒为单位定义发送或接收消息之间的最大时间间隔必须0 // 但这个方法并没有重连的机制 options.setKeepAliveInterval(20); // 设置“遗嘱”消息的话题若客户端与服务器之间的连接意外中断服务器将发布客户端的“遗嘱”消息。 options.setWill(willTopic, WILL_DATA, 2, false); //自动重新连接 options.setAutomaticReconnect(true); factory.setConnectionOptions(options); log.info(初始化 MQTT 配置); return factory; } }5.3、修改消费者配置类在消费者配置类中将原来的String defaultTopic mqttConfiguration.getDefaultTopic();替换成了String[] topics mqttConfiguration.getAllTopics();package com.jyy.common.mqtt.config; import com.jyy.common.mqtt.constants.MqttConstant; import com.jyy.common.mqtt.handler.MqttMessageReceiver; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; 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.endpoint.MessageProducerSupport; 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; import javax.annotation.Resource; /** * 消费者配置 * * author jyy * date 2024-11-26 */ Slf4j AllArgsConstructor Configuration IntegrationComponentScan public class MqttInboundConfiguration { Resource private MqttConfiguration mqttConfiguration; private MqttMessageReceiver mqttMessageReceiver; /** * 此处可以使用其他消息通道 * MQTT信息通道消费者 * Spring Integration默认的消息通道它允许将消息发送给一个订阅者然后阻碍发送直到消息被接收。 */ Bean public MessageChannel mqttInBoundChannel() { return new DirectChannel(); } /** * mqtt入站消息处理工具对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。 */ Bean ServiceActivator(inputChannel mqttInBoundChannel) public MessageHandler mqttMessageHandler() { return this.mqttMessageReceiver; } /** * MQTT消息订阅绑定消费者 * 适配器, 两个topic共用一个adapter * 客户端作为消费者订阅主题消费消息 */ Bean public MessageProducerSupport mqttInbound() { // 获取客户端id String clientId mqttConfiguration.getClientId(); // 获取默认主题 // String defaultTopic mqttConfiguration.getDefaultTopic(); // 2024-12-06新增获取所有配置的主题 String[] topics mqttConfiguration.getAllTopics(); // 获取客户端工厂 MqttPahoClientFactory mqttPahoClientFactory mqttConfiguration.mqttClientFactory(); // Paho客户端消息驱动通道适配器主要用来订阅主题 MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter( clientId MqttConstant.CLIENT_SUFFIX_CONSUMERS, mqttPahoClientFactory, // TODO 这后面填写需要订阅的主题这里是一个可变参数可以填写多个用逗号隔开。例如我再订阅一个yy/的主题 // defaultTopic, // yy/, // TODO 2024-12-06新增把所有订阅主题都放在一起方便管理 topics ); adapter.setCompletionTimeout(5000); // Paho消息转换器 DefaultPahoMessageConverter defaultPahoMessageConverter new DefaultPahoMessageConverter(); // 按字节接收消息 // defaultPahoMessageConverter.setPayloadAsBytes(true); adapter.setConverter(defaultPahoMessageConverter); // 设置QoS adapter.setQos(1); // 设置订阅通道 adapter.setOutputChannel(mqttInBoundChannel()); return adapter; } }5.4、启动项目测试修改①项目启动成功②再看EMQ X后台管理页面也出现了我们在配置文件中配置的 default-topic默认主题和 topics要订阅的其他主题③使用MQTTX向 topics要订阅的其他主题发送消息看到后台打印了我们用MQTTX向 yy/1 发送的消息至此使用配置文件配置多个主题避免硬编码的功能就实现咯。我就一个代码低手欢迎大家跟我交流学习喔。

相关新闻

最新新闻

日新闻

周新闻

月新闻