AI 助理

你好，我是AI助理，可以解答问题、推荐解决方案等

开发者社区 数据库 文章 正文
数据库
内容
活动
关注

Java 开发中基于 Spring Boot 3.2 框架集成 MQTT 5.0 协议实现消息推送与订阅功能的技术方案解析

2025-06-11 835
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
啦啦啦191
简介： 本文介绍基于Spring Boot 3.2集成MQTT 5.0的消息推送与订阅技术方案，涵盖核心技术栈选型（Spring Boot、Eclipse Paho、HiveMQ）、项目搭建与配置、消息发布与订阅服务实现，以及在智能家居控制系统中的应用实例。同时，详细探讨了安全增强（TLS/SSL）、性能优化（异步处理与背压控制）、测试监控及生产环境部署方案，为构建高可用、高性能的消息通信系统提供全面指导。附资源下载链接：[https://pan.quark.cn/s/14fcf913bae6](https://pan.quark.cn/s/14fcf913bae6)。

Spring Boot 3.2集成MQTT 5.0实现消息推送与订阅技术方案

一、技术选型与架构设计

1. 核心技术栈

  • Spring Boot 3.2.0 (基于Java 17)
  • Eclipse Paho MQTT Client 1.2.5
  • MQTT 5.0 协议 (支持属性扩展、增强的错误处理)
  • HiveMQ (开源MQTT Broker)
  • WebSocket 支持 (可选)

2. 架构图

+------------------+ +------------------+ +------------------+ | 设备/前端应用 |<--->| MQTT Broker |<--->| Spring Boot应用 | +------------------+ +------------------+ +------------------+ 发布/订阅 消息路由 业务处理

二、项目搭建与配置

1. 创建Spring Boot项目

使用Spring Initializr创建项目，添加以下依赖：

<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-integration</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency> <dependency> <groupId>com.hivemq</groupId> <artifactId>hivemq-mqtt-client</artifactId> <version>1.3.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency> </dependencies>

2. 配置MQTT连接

使用HiveMQ客户端实现MQTT 5.0支持：

@Configuration public class MqttConfig {  @Value("${mqtt.broker-url}") private String brokerUrl; @Value("${mqtt.client-id}") private String clientId; @Value("${mqtt.username}") private String username; @Value("${mqtt.password}") private String password; @Bean public Mqtt5AsyncClient mqttClient() {  Mqtt5AsyncClient client = MqttClient.builder() .useMqttVersion5() .identifier(clientId) .serverHost(brokerUrl) .serverPort(1883) .buildAsync(); // 添加认证信息 Mqtt5ConnectBuilder.Mqtt5ConnectWithUserPropertiesBuilder connectBuilder = Mqtt5ClientConnectionConfig.builder() .automaticReconnectWithDefaultConfig() .build(); return client; } // 配置消息转换器和质量服务 @Bean public MqttPahoClientFactory mqttClientFactory() {  DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(new String[]{ brokerUrl}); options.setUserName(username); options.setPassword(password.toCharArray()); options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_5); factory.setConnectionOptions(options); return factory; } }

3. 配置文件示例

# MQTT配置 mqtt.broker-url=tcp://localhost:1883 mqtt.client-id=spring-boot-mqtt-client mqtt.username=admin mqtt.password=password mqtt.default-qos=1 mqtt.keep-alive-interval=30

三、消息发布服务实现

1. 通用消息发布服务

@Service public class MqttPublisherService {  private static final Logger logger = LoggerFactory.getLogger(MqttPublisherService.class); private final Mqtt5AsyncClient mqttClient; @Autowired public MqttPublisherService(Mqtt5AsyncClient mqttClient) {  this.mqttClient = mqttClient; } /** * 发布MQTT消息 * @param topic 主题 * @param payload 消息内容 * @param qos 服务质量等级 * @param retained 是否保留消息 */ public CompletableFuture<Void> publish(String topic, String payload, int qos, boolean retained) {  Mqtt5Publish publishMessage = Mqtt5Publish.builder() .topic(topic) .payload(ByteBuffer.wrap(payload.getBytes(StandardCharsets.UTF_8))) .qos(MqttQos.fromCode(qos)) .retain(retained) .build(); return mqttClient.publish(publishMessage) .thenAccept(publishResult -> logger.info("消息发布成功: {}", publishResult)) .exceptionally(ex -> {  logger.error("消息发布失败: {}", ex.getMessage(), ex); return null; }); } // 重载方法，使用默认QoS和retained设置 public CompletableFuture<Void> publish(String topic, String payload) {  return publish(topic, payload, 1, false); } }

2. 领域特定消息发布示例

@Service public class DeviceMessageService {  private static final String DEVICE_DATA_TOPIC = "v1/devices/me/telemetry"; @Autowired private MqttPublisherService publisherService; public CompletableFuture<Void> sendDeviceData(String deviceId, Map<String, Object> data) {  try {  ObjectMapper mapper = new ObjectMapper(); String payload = mapper.writeValueAsString(data); String topic = String.format("%s/%s", DEVICE_DATA_TOPIC, deviceId); return publisherService.publish(topic, payload); } catch (JsonProcessingException e) {  logger.error("序列化设备数据失败: {}", e.getMessage(), e); return CompletableFuture.failedFuture(e); } } }

四、消息订阅服务实现

1. 基于注解的消息处理

@Component public class MqttMessageListener {  private static final Logger logger = LoggerFactory.getLogger(MqttMessageListener.class); @Autowired private DeviceService deviceService; @ServiceActivator(inputChannel = "mqttInputChannel") public void handleMessage(Message<?> message) {  String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC); String payload = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8); logger.info("收到MQTT消息 - 主题: {}, 内容: {}", topic, payload); // 根据主题路由消息处理 if (topic.startsWith("v1/devices/")) {  handleDeviceMessage(topic, payload); } else if (topic.startsWith("system/")) {  handleSystemMessage(topic, payload); } } private void handleDeviceMessage(String topic, String payload) {  try {  // 解析设备ID String deviceId = topic.split("/")[2]; // 解析JSON数据 ObjectMapper mapper = new ObjectMapper(); JsonNode data = mapper.readTree(payload); // 处理设备数据 deviceService.processDeviceData(deviceId, data); } catch (Exception e) {  logger.error("处理设备消息失败: {}", e.getMessage(), e); } } private void handleSystemMessage(String topic, String payload) {  // 处理系统消息逻辑 } }

2. 配置消息订阅

@Configuration public class MqttSubscriberConfig {  @Value("${mqtt.client-id}") private String clientId; @Autowired private MqttPahoClientFactory mqttClientFactory; @Bean public MessageChannel mqttInputChannel() {  return new DirectChannel(); } @Bean public MessageProducer inbound() {  MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter( clientId + "-subscriber", mqttClientFactory, "v1/devices/#", // 订阅设备相关主题 "system/#" // 订阅系统相关主题 ); adapter.setQos(1); adapter.setOutputChannel(mqttInputChannel()); return adapter; } @Bean @ServiceActivator(inputChannel = "mqttInputChannel") public MessageHandler handler() {  return new MessageHandler() {  @Override public void handleMessage(Message<?> message) throws MessagingException {  // 消息将被路由到MqttMessageListener } }; } }

五、应用实例：智能家居控制系统

1. 设备状态监控与控制

设备数据模型

public record DeviceData( String deviceId, double temperature, double humidity, boolean powerStatus, LocalDateTime timestamp ) { }

设备服务实现

@Service public class SmartHomeService {  private static final String CONTROL_TOPIC = "v1/devices/me/control"; @Autowired private MqttPublisherService publisherService; @Autowired private DeviceRepository deviceRepository; // 处理设备上报数据 public void processDeviceData(String deviceId, JsonNode data) {  // 解析数据 double temperature = data.path("temperature").asDouble(); double humidity = data.path("humidity").asDouble(); boolean powerStatus = data.path("powerStatus").asBoolean(); // 创建设备数据对象 DeviceData deviceData = new DeviceData( deviceId, temperature, humidity, powerStatus, LocalDateTime.now() ); // 保存数据 deviceRepository.save(deviceData); // 检查自动化规则 checkAutomationRules(deviceData); } // 发送控制命令到设备 public CompletableFuture<Void> sendDeviceCommand(String deviceId, String command) {  String topic = String.format("%s/%s", CONTROL_TOPIC, deviceId); return publisherService.publish(topic, command); } // 自动化规则检查 private void checkAutomationRules(DeviceData data) {  // 示例：温度超过30度时自动打开空调 if (data.deviceId().endsWith("thermostat") && data.temperature() > 30) {  sendDeviceCommand("air-conditioner-01", "{\"command\":\"ON\"}"); } } }

2. REST API实现

@RestController @RequestMapping("/api/v1/devices") public class DeviceController {  @Autowired private SmartHomeService smartHomeService; @PostMapping("/{deviceId}/command") public ResponseEntity<?> sendCommand( @PathVariable String deviceId, @RequestBody String command ) {  smartHomeService.sendDeviceCommand(deviceId, command) .thenAccept(v -> ResponseEntity.ok().build()) .exceptionally(ex -> ResponseEntity.status(500).body(ex.getMessage())); return ResponseEntity.accepted().build(); } @GetMapping("/{deviceId}/data") public ResponseEntity<DeviceData> getDeviceData(@PathVariable String deviceId) {  Optional<DeviceData> deviceData = smartHomeService.getLatestData(deviceId); return deviceData.map(ResponseEntity::ok) .orElse(ResponseEntity.notFound().build()); } }

六、安全增强与性能优化

1. TLS/SSL配置

@Bean public MqttConnectOptions mqttConnectOptions() {  MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(new String[]{ brokerUrl}); options.setUserName(username); options.setPassword(password.toCharArray()); // 配置TLS if (useTls) {  try {  SSLSocketFactory sslSocketFactory = createSSLSocketFactory(); options.setSocketFactory(sslSocketFactory); } catch (Exception e) {  logger.error("配置TLS连接失败: {}", e.getMessage(), e); } } return options; } private SSLSocketFactory createSSLSocketFactory() throws Exception {  // 加载证书 KeyStore keyStore = KeyStore.getInstance("JKS"); InputStream inputStream = getClass().getResourceAsStream("/client.jks"); keyStore.load(inputStream, "password".toCharArray()); // 初始化SSL上下文 SSLContext sslContext = SSLContext.getInstance("TLS"); TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance( TrustManagerFactory.getDefaultAlgorithm()); trustManagerFactory.init(keyStore); sslContext.init(null, trustManagerFactory.getTrustManagers(), new SecureRandom()); return sslContext.getSocketFactory(); }

2. 异步处理与背压控制

@Service public class AsyncMessageProcessor {  private final ExecutorService threadPool = Executors.newFixedThreadPool(10); public void processMessageAsync(String topic, String payload) {  CompletableFuture.runAsync(() -> {  try {  // 处理消息的业务逻辑 processMessage(topic, payload); } catch (Exception e) {  logger.error("异步处理消息失败: {}", e.getMessage(), e); } }, threadPool); } private void processMessage(String topic, String payload) {  // 消息处理逻辑 } }

七、测试与监控

1. 单元测试示例

@SpringBootTest @ActiveProfiles("test") class MqttIntegrationTest {  @Autowired private MqttPublisherService publisherService; @Autowired private TestMessageCollector messageCollector; @Test void testPublishAndSubscribe() throws Exception {  String testTopic = "test/unit/" + UUID.randomUUID().toString(); String testPayload = "Test Message"; // 设置预期消息 messageCollector.expectMessage(testTopic, testPayload); // 发布消息 publisherService.publish(testTopic, testPayload).get(5, TimeUnit.SECONDS); // 验证消息是否收到 assertTrue(messageCollector.waitForMessage(5, TimeUnit.SECONDS)); } }

2. 监控指标

使用Micrometer添加MQTT相关监控指标：

@Bean public MeterRegistryCustomizer<MeterRegistry> configurer( @Value("${spring.application.name}") String applicationName) {  return (registry) -> registry.config().commonTags("application", applicationName); } // 在消息处理中添加计数器 @Service public class MqttMetricsService {  private final Counter publishCounter; private final Counter subscribeCounter; private final Timer messageProcessingTimer; public MqttMetricsService(MeterRegistry registry) {  publishCounter = registry.counter("mqtt.publish.count"); subscribeCounter = registry.counter("mqtt.subscribe.count"); messageProcessingTimer = registry.timer("mqtt.message.processing.time"); } public void incrementPublishCount() {  publishCounter.increment(); } public void incrementSubscribeCount() {  subscribeCounter.increment(); } public <T> T recordProcessingTime(Supplier<T> operation) {  return messageProcessingTimer.record(operation); } }

八、生产环境部署

1. MQTT Broker选型

  • HiveMQ CE：开源、高性能、支持MQTT 5.0
  • Mosquitto：轻量级、易于部署
  • EMQ X：企业级、支持百万级连接

2. Docker部署示例

version: '3' services: mqtt-broker: image: hivemq/hivemq-ce ports: - "1883:1883" # MQTT - "8080:8080" # HiveMQ Web UI volumes: - ./hivemq/config:/opt/hivemq/conf - ./hivemq/data:/opt/hivemq/data - ./hivemq/log:/opt/hivemq/log restart: always spring-boot-app: build: . ports: - "8081:8081" environment: - MQTT_BROKER_URL=mqtt-broker - MQTT_USERNAME=admin - MQTT_PASSWORD=password depends_on: - mqtt-broker restart: always

九、总结与扩展

本文展示了如何使用Spring Boot 3.2集成MQTT 5.0实现消息推送与订阅，通过实际案例演示了智能家居控制系统的实现。可以根据需求进一步扩展：

  1. 添加消息持久化存储（如Redis、MongoDB）
  2. 实现消息重试机制和幂等性保障
  3. 集成WebSocket支持Web客户端实时通信
  4. 添加分布式追踪（如Zipkin、Jaeger）
  5. 实现多租户隔离和权限控制

通过合理的架构设计和技术选型，可以构建出高可用、高性能、安全可靠的消息通信系统。

Java 开发，Spring Boot 3.2,MQTT 5.0, 消息推送，消息订阅，Spring 框架集成，MQTT 协议应用，实时通信技术，微服务消息传递，物联网消息交互，Spring Boot 集成 MQTT, 异步消息处理，分布式消息系统，Java 后端开发，消息中间件集成

资源地址：
https://pan.quark.cn/s/14fcf913bae6

文章标签：
云解析DNS
云消息队列 MQ
Java
Spring
监控
智能硬件
安全
关键词：
Java开发
Java框架
Spring解析
Spring Boot解析
Spring框架
来 源：
开发者社区 > 数据库 > 文章 > 正文

数据库
啦啦啦191
目录
相关文章
木流牛马
|
2月前
|
安全 Java Ruby
我尝试了所有后端框架 — — 这就是为什么只有 Spring Boot 幸存下来
作者回顾后端开发历程，指出多数框架在生产环境中难堪重负。相比之下，Spring Boot凭借内置安全、稳定扩展、完善生态和企业级支持，成为构建高可用系统的首选，真正经受住了时间与规模的考验。
木流牛马
223 2
阿里云云原生
|
25天前
|
人工智能 运维 Java
Spring AI Alibaba Admin 开源！以数据为中心的 Agent 开发平台
Spring AI Alibaba Admin 正式发布！一站式实现 Prompt 管理、动态热更新、评测集构建、自动化评估与全链路可观测，助力企业高效构建可信赖的 AI Agent 应用。开源共建，现已上线！
阿里云云原生
2284 41
一枚后端工程狮
|
28天前
|
安全 前端开发 Java
《深入理解Spring》：现代Java开发的核心框架
Spring自2003年诞生以来，已成为Java企业级开发的基石，凭借IoC、AOP、声明式编程等核心特性，极大简化了开发复杂度。本系列将深入解析Spring框架核心原理及Spring Boot、Cloud、Security等生态组件，助力开发者构建高效、可扩展的应用体系。（238字）
一枚后端工程狮
356 55
一枚后端工程狮
|
28天前
|
XML Java 数据格式
《深入理解Spring》：AOP面向切面编程深度解析
Spring AOP通过代理模式实现面向切面编程，将日志、事务等横切关注点与业务逻辑分离。支持注解、XML和编程式配置，提供五种通知类型及丰富切点表达式，助力构建高内聚、低耦合的可维护系统。
一枚后端工程狮
387 7
一枚后端工程狮
|
28天前
|
前端开发 Java 微服务
《深入理解Spring》：Spring、Spring MVC与Spring Boot的深度解析
Spring Framework是Java生态的基石，提供IoC、AOP等核心功能；Spring MVC基于其构建，实现Web层MVC架构；Spring Boot则通过自动配置和内嵌服务器，极大简化了开发与部署。三者层层演进，Spring Boot并非替代，而是对前者的高效封装与增强，适用于微服务与快速开发，而深入理解Spring Framework有助于更好驾驭整体技术栈。
一枚后端工程狮
310 4
蓝易云
|
1月前
|
消息中间件 缓存 Java
Spring框架优化：提高Java应用的性能与适应性
以上方法均旨在综合考虑Java Spring 应该程序设计原则, 数据库交互, 编码实践和系统架构布局等多角度因素, 旨在达到高效稳定运转目标同时也易于未来扩展.
蓝易云
110 8
凉凉心.
|
1月前
|
XML JSON Java
【SpringBoot（三）】从请求到响应再到视图解析与模板引擎，本文带你领悟SpringBoot请求接收全流程！
Springboot专栏第三章，从请求的接收到视图解析，再到thymeleaf模板引擎的使用！ 本文带你领悟SpringBoot请求接收到渲染的使用全流程！
凉凉心.
163 3
varin
|
2月前
|
消息中间件 Ubuntu Java
SpringBoot整合MQTT实战：基于EMQX实现双向设备通信
本教程指导在Ubuntu上部署EMQX 5.9.0并集成Spring Boot实现MQTT双向通信，涵盖服务器搭建、客户端配置及生产实践，助您快速构建企业级物联网消息系统。
varin
929 1
JJLIN距离
|
2月前
|
人工智能 Java API
Java与大模型集成实战：构建智能Java应用的新范式
随着大型语言模型（LLM）的API化，将其强大的自然语言处理能力集成到现有Java应用中已成为提升应用智能水平的关键路径。本文旨在为Java开发者提供一份实用的集成指南。我们将深入探讨如何使用Spring Boot 3框架，通过HTTP客户端与OpenAI GPT（或兼容API）进行高效、安全的交互。内容涵盖项目依赖配置、异步非阻塞的API调用、请求与响应的结构化处理、异常管理以及一些面向生产环境的最佳实践，并附带完整的代码示例，助您快速将AI能力融入Java生态。
JJLIN距离
390 12
varin
|
2月前
|
Java 数据库 数据安全/隐私保护
Spring Boot四层架构深度解析
本文详解Spring Boot四层架构（Controller-Service-DAO-Database）的核心思想与实战应用，涵盖职责划分、代码结构、依赖注入、事务管理及常见问题解决方案，助力构建高内聚、低耦合的企业级应用。
varin
683 1

热门文章

最新文章

  • 1
    ETCD系列之一：简介
  • 2
    阿里云分析引擎Spark On 多数据源介绍
  • 3
    德哥的PostgreSQL私房菜 - 史上最屌PG资料合集
  • 4
    PostgreSQL 如何实现批量更新、删除、插入
  • 5
    MongoDB Sharded cluster架构原理
  • 6
    Redis Stream——作为消息队列的典型应用场景
  • 7
    MongoDB 生态 - 可视化管理工具
  • 8
    在Docker上玩转PostgreSQL -- Mac篇
  • 9
    DRDS 数据恢复重磅发布，全方位保障您的数据安全
  • 10
    PostgreSQL upsert功能(insert on conflict do)的用法
  • 1
    【Java架构师体系课 | MySQL篇】④ 索引失效问题总结
    27
  • 2
    Mysql如何往表中插入10万条测试数据
    25
  • 3
    【Java架构师体系课 | MySQL篇】③ Explain执行计划详解
    27
  • 4
    【赵渝强老师】OceanBase的连接与路由管理
    26
  • 5
    StarRocks 4.0：让 Apache Iceberg 数据真正 Query-Ready
    30
  • 6
    从 ClickHouse、Druid、Kylin 到 Doris：网易云音乐 PB 级实时分析平台降本增效
    36
  • 7
    专为 Apache Doris 打造的可视化数据管理工具 SelectDB Studio
    31
  • 8
    什么是脏读、幻读、不可重复读？Mysql的隔离级别是什么？
    32
  • 9
    CMP7（类Cloudera CDP 7 404版华为Kunpeng）用开源软件Label Studio做数据标注
    55
  • 10
    【Java架构师体系课 | MySQL篇】② 深入理解MySQL索引底层数据结构与算法
    53

    • 相关课程

    查看更多
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计
  • Java编程入门
  • Java面向对象编程
  • Java高级编程

    • 相关电子书

    查看更多
  • Spring Cloud Alibaba - 重新定义 Java Cloud-Native
  • The Reactive Cloud Native Arch
  • JAVA开发手册1.5.0
    • 下一篇