物联网实战|Spring Boot MQTT平台 |emqx broker实战

EMQX 是一款「无限连接，任意集成，随处运行」的大规模分布式物联网接入平台基于版本6.2.2

emqx官网

https://docs.emqx.com/zh/emqx/latest/

docker部署emqx

其中18083 为网页访问端口 (http://127.0.0.1:28084/#/dashboard/overview)

账号

默认为 admin/public

前期准备和权限开通

mkdir -p /data/emqx/
chmod -R 777 /data/

docker构建emqx broker应用

  docker run -d   \
  --name emqx   \
  --restart always   \
  -p 1883:1883   \
  -p 8083:8083   \
  -p 8084:8084   \
  -p 8883:8883   \
  -p 28084:18083  \
  -p 18085:18085   \
  -v /data/emqx/data:/opt/emqx/data   \
  -v /data/emqx/log:/opt/emqx/log   \
  -e EMQX_COPY_CONF=true   \
  emqx/emqx:latest

spring-boot接入emqx broker

pom依赖

<dependencies>
    <!-- HiveMQ MQTT5 客户端（同时兼容 MQTT3.1.1） -->
    <dependency>
        <groupId>com.hivemq</groupId>
        <artifactId>hivemq-mqtt-client</artifactId>
        <version>1.3.15</version>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <!-- Spring Boot Starter -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>3.5.11</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

源码

https://gitee.com/kcnf-iot/iot-sample/tree/master/emqx

客户端注册部分代码

package com.jysemel.iot.client;

import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;

@Slf4j
@Configuration
public class MqttConfig {

    @Bean
    public Mqtt5AsyncClient mqtt5AsyncClient() {
        // 使用 Builder 模式创建异步客户端
        Mqtt5AsyncClient client = Mqtt5Client.builder()
                .identifier("client" + UUID.randomUUID()) // 设置客户端ID，需唯一
                .serverHost("192.168.0.19") // Broker 地址
                .serverPort(1883) // 默认非加密端口
                .buildAsync(); // 构建异步客户端

        // 连接到 Broker
        client.connect()
                .whenComplete((connAck, throwable) -> {
                    if (throwable != null) {
                        System.err.println("MQTT 连接失败: " + throwable.getMessage());
                        // 可以在这里添加重试逻辑
                    } else {
                        System.out.println("MQTT 连接成功!");
                    }
                });
        return client;
    }
}

客户端注册信息

订阅部分代码

package com.jysemel.iot.client;

import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import jakarta.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

@Service
public class MqttSubscribeService {

    @Autowired
    private Mqtt5AsyncClient mqttClient;

    // 使用 @PostConstruct 在 Bean 初始化后自动订阅
    @PostConstruct
    public void subscribe() {
        String topic = "server/v1/test";
        mqttClient.subscribeWith()
                .topicFilter(topic) // 订阅的主题
                .qos(MqttQos.AT_LEAST_ONCE)
                .callback(publish -> {
                    // 处理收到的消息
                    String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
                    System.out.println("收到消息，主题: " + publish.getTopic() + ", 内容: " + payload);
                })
                .send()
                .whenComplete((subAck, throwable) -> {
                    if (throwable != null) {
                        System.err.println("订阅失败: " + throwable.getMessage());
                    } else {
                        System.out.println("订阅主题 'test/topic' 成功!");
                    }
                });
    }
}

emqx 模拟下发

{ "msg": "hello" }