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

news2024/10/9 11:25:44

一、前言

本篇主要是围绕着两个点，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:latest

docker ps

docker 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_001
    password: 123qwe,./
    clean-session: true
    default-topic: default/topic

2、MqttConfig.java

@Configuration
public class MqttConfig {

    // 配置MQTT连接选项，serverURIs、userName、password、cleanSession，这几个配置可以配置在application.yml文件中
    @Bean
    public 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;
    }

    // 创建客户端工厂
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    // 配置消息接收通道
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    // 配置MQTT入站通道适配器（订阅消息）
    // default-topic：默认订阅消息主题
    @Bean
    public 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;
    }

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

4、MqttPublisherService.java

@Slf4j
@Service
public class MqttPublisherService {

    /**
     * 发送消息的通道
     */
    @Resource
    private 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 发布消息服务组件
     */
    @Resource
    private 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);
    }
}