当前位置：首页 > news >正文

3、Docker搭建MQTT及Spring Boot 3.x集成MQTT

news 2025/8/21 13:11:10

一、前言

本篇主要是围绕着两个点，1、Docker 搭建单机版本 MQTT（EMQX），2、Spring Boot 3.x 集成 MQTT（EMQX）；
而且这里的 MQTT（EMQX）的搭建也只是一个简单的过程，主要 Spring Boot 的集成 MQTT 部分

二、Docker 搭建 MQTT（EMQX）

在这里暂时就不说 Docker 的安装了，如果没有安装可以去 [Docker官网](https://www.docker.com/) 看一下文档即可；

1、emqx 容器初始化

## 下载镜像
docker pull emqx/emqx:latest## 启动容器
docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8081:8081 -p 18083:18083 --network local-net emqx/emqx:latestdocker psdocker logs -f emqx## http://<你的服务器IP>:18083
## 默认用户名：username
## 默认密码：public

2、emqx 配置

在这里插入图片描述
2.1、emqx 默认密码登陆之后会有一个重置登录密码页面，可以自己选择是否跳过，之后就进去主界面

在这里插入图片描述
2.2 访问控制-客户端认证
客户端认证创建一个认证方式，这里使用的是内置数据库的方式，

下一步之后，有多种选择数据源可以选择，每一种数据源的基础新增命令都是有的，可以参考官方提供的命令进行新增即可
在这里插入图片描述
因为已经选择了内置数据库，这里就通过redis给举例一下参考命令的地方

这开始就进行客户端用的创建

到此为止，EMQX的配置基本上结束；至于其他方面配置，各位大神自己摸索吧哈哈哈哈

三、Spring Boot 3.x 集成 MQTT（EMQX）

1、pom.xml

<!-- MQTT 连接版本 -->
<mqtt.version>6.3.4</mqtt.version><!-- EMQX -->
<!-- MQTT Client (Paho) -->
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId><version>${mqtt.version}</version>
</dependency>

2、application.yml

spring:# mqtt配置mqtt:server-uris: ["tcp://localhost:1883"]username: mqttx_user_001password: 123qwe,./clean-session: truedefault-topic: default/topic

2、MqttConfig.java

@Configuration
public class MqttConfig {// 配置MQTT连接选项，serverURIs、userName、password、cleanSession，这几个配置可以配置在application.yml文件中@Beanpublic MqttConnectOptions mqttConnectOptions() {MqttConnectOptions options = new MqttConnectOptions();options.setServerURIs(new String[]{"tcp://localhost:1883"}); 	// EMQX服务器地址、可以集成到application.yml中options.setUserName("mqttx_user_001");  						// 如果需要身份验证、可以集成到application.yml中options.setPassword("123456".toCharArray());  					// 如果需要密码、可以集成到application.yml中options.setCleanSession(true);  								// 是否清除会话、可以集成到application.yml中return options;}// 创建客户端工厂@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();factory.setConnectionOptions(mqttConnectOptions());return factory;}// 配置消息接收通道@Beanpublic MessageChannel mqttInputChannel() {return new DirectChannel();}// 配置MQTT入站通道适配器（订阅消息）// default-topic：默认订阅消息主题@Beanpublic MqttPahoMessageDrivenChannelAdapter inbound() {MqttPahoMessageDrivenChannelAdapter adapter =new MqttPahoMessageDrivenChannelAdapter("client-id", mqttClientFactory(),"default-topic");  										// 指定订阅的主题、可以集成到application.yml中adapter.setCompletionTimeout(5000);adapter.setConverter(new DefaultPahoMessageConverter());adapter.setQos(1);  													// 设置QoS级别、可以集成到application.yml中adapter.setOutputChannel(mqttInputChannel());							// 指定消息通道return adapter;}// 配置处理收到的消息// 接收默认订阅消息主题@Bean@ServiceActivator(inputChannel = "mqttInputChannel")public MessageHandler handler() {return message -> {System.out.println("Received message: " + message.getPayload());};}// 配置MQTT出站通道适配器（发布消息）// default-topic：默认发布消息主题@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")public MessageHandler mqttOutbound() {MqttPahoMessageHandler messageHandler =new MqttPahoMessageHandler("client-id", mqttClientFactory());messageHandler.setAsync(true);  										// 异步发送、可以集成到application.yml中messageHandler.setDefaultTopic("default-topic");  						// 默认发送的主题、可以集成到application.yml中return messageHandler;}// 用于发送消息的通道@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel();}
}

4、MqttPublisherService.java

@Slf4j
@Service
public class MqttPublisherService {/*** 发送消息的通道*/@Resourceprivate MessageChannel mqttOutboundChannel;/*** 默认主题发送消息** @param payload 消息内容*/public void defaultTopicSendMessage(String payload) {log.info("default topic, with payload: {}", payload);mqttOutboundChannel.send(MessageBuilder.withPayload(payload).build());}/*** 发送消息到指定主题** @param topic   主题* @param payload 消息内容*/public void withTopicSendMessage(String topic, String payload) {log.info("topic: {}, with payload: {}", topic, payload);mqttOutboundChannel.send(MessageBuilder.withPayload(payload).setHeader("mqtt_topic", topic).build());}
}

5、MqttPublisherController.java

@RestController
@RequestMapping("/mqtt/publisher")
public class MqttPublisherController {/*** MQTT 发布消息服务组件*/@Resourceprivate MqttPublisherService mqttPublisherService;/*** 发送消息到默认主题** @param payload 消息内容*/@PostMapping("/default/topic")public void defaultTopicSendMessage(@RequestParam String payload) {mqttPublisherService.defaultTopicSendMessage(payload);}/*** 发送消息到指定主题** @param topic   主题* @param payload 消息内容*/@PostMapping("/with/topic")public void withTopicSendMessage(@RequestParam String topic, @RequestParam String payload) {mqttPublisherService.withTopicSendMessage(topic, payload);}
}