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

springboot集成mqtt的订阅端

news 2026/6/14 13:44:14

环境：JDK21

springboot：3.3.5

EMQX：5.8.5

MQTT协议：5.0

引入的mqtt jar包：

<dependency> <groupId>com.hivemq</groupId> <artifactId>hivemq-mqtt-client</artifactId> <version>1.3.0</version> </dependency>

yml中mqtt的配置：

$share/my-group/test/topic：共享订阅，也就是负载均衡，$share是固定写法，my-group共享组，共享组内的消费者只能有一个消费，

mqtt: host: 192.168.1.2 port: 1883 client-id: springboot-listener-001 username: admin password: xxyy1212@ qos: 1 topics: - device/up - device/data - $share/my-group/test/topic

读取yml的配置文件

import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; import java.util.List; @Component @ConfigurationProperties(prefix = "mqtt") public class MqttProperties { private String host; private int port; private String clientId; private String username; private String password; private int qos; private List<String> topics; // ✅ 列表正确读取 // 自动生成 getter setter public String getHost() {return host;} public void setHost(String host) {this.host = host;} public int getPort() {return port;} public void setPort(int port) {this.port = port;} public String getClientId() {return clientId;} public void setClientId(String clientId) {this.clientId = clientId;} public String getUsername() {return username;} public void setUsername(String username) {this.username = username;} public String getPassword() {return password;} public void setPassword(String password) {this.password = password;} public int getQos() {return qos;} public void setQos(int qos) {this.qos = qos;} public List<String> getTopics() {return topics;} public void setTopics(List<String> topics) {this.topics = topics;} }

接收消息：

1、自动检测链接状态，

2、连接中断后自动连接

package com.crazy.shopping.listener; import com.crazy.shopping.config.MqttProperties; import com.hivemq.client.mqtt.MqttClientState; import com.hivemq.client.mqtt.MqttGlobalPublishFilter; import com.hivemq.client.mqtt.datatypes.MqttQos; import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; import com.hivemq.client.mqtt.mqtt5.Mqtt5Client; import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck; import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5DisconnectReasonCode; import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.nio.charset.StandardCharsets; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.*; @Component public class Mqtt5MessageListener { @Autowired private MqttProperties mqttProperties; private MqttQos mqttQos; private Mqtt5AsyncClient client; private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); private final ExecutorService messageExecutor = new ThreadPoolExecutor( 3, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy() ); @PostConstruct public void start() { mqttQos = MqttQos.fromCode(mqttProperties.getQos()); // 每3秒检查重连 + 强制绑定监听 scheduler.scheduleAtFixedRate(this::checkAndConnect, 0, 3, TimeUnit.SECONDS); } private void checkAndConnect() { if (client != null && client.getState() == MqttClientState.CONNECTED) { return; } System.out.println("⏳ 重连中..."); // 🔥 每次重连都新建客户端（彻底解决监听丢失） client = Mqtt5Client.builder() .serverHost(mqttProperties.getHost()) .serverPort(mqttProperties.getPort()) .identifier(mqttProperties.getClientId()) .buildAsync(); // ========================================== // 🔥 关键：连接之前 先注册监听！！！ // ========================================== registerMessageListener(); // 连接 client.connectWith() .cleanStart(false) .sessionExpiryInterval(86400L) .keepAlive(60) .simpleAuth() .username(mqttProperties.getUsername()) .password(mqttProperties.getPassword().getBytes(StandardCharsets.UTF_8)) .applySimpleAuth() .send() .whenComplete((ack, t) -> { if (t == null) { System.out.println("✅ 连接成功！"); subscribeAllTopics(); } }); } // ========================================== // 🔥 每次重连都重新注册监听（绝杀） // ========================================== private void registerMessageListener() { client.toAsync().publishes(MqttGlobalPublishFilter.ALL, this::handleMessage); } // 消息处理：可以进行幂等性校验结合redis--TODO private void handleMessage(Mqtt5Publish publish) { messageExecutor.submit(() -> { try { String topic = publish.getTopic().toString(); String msg = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8); System.out.println("========================================"); System.out.println("📥 收到消息 >>> " + topic); System.out.println("内容 >>> " + msg); System.out.println("========================================"); } catch (Exception e) { e.printStackTrace(); } }); } // 订阅 private void subscribeAllTopics() { for (String topic : mqttProperties.getTopics()) { client.subscribeWith() .topicFilter(topic) .qos(mqttQos) .send() .whenComplete((sub, t) -> { if (t == null) { System.out.println("✅ 已订阅：" + topic); } }); } } @PreDestroy public void close() { try { scheduler.shutdownNow(); messageExecutor.shutdown(); if (client != null && client.getState().isConnected()) { client.disconnect(); } } catch (Exception ignored) {} } }

效果：

查看全文

http://www.jsqmd.com/news/603595/