1. 项目概述为什么 MQTT 是 SpringBoot 做物联网通信的“默认答案”你有没有遇到过这种场景用 SpringBoot 写了个设备管理后台前端页面要实时显示温湿度传感器每秒上报的数据后端得同时支撑几百台设备长连接或者要做一个智能灌溉系统主控箱需要接收云端下发的浇水指令还要把土壤含水量、阀门状态主动推给服务器——这时候如果还用 HTTP 轮询服务器 CPU 30% 花在建立/断开连接上响应延迟动辄 2 秒设备电量三天就耗光。我去年在某高校实验室带一个农业物联网模拟项目X时就踩过这个坑最初用 RestTemplate 每 5 秒拉一次数据结果 87 台 LoRa 节点一上线Nginx 就开始报 502日志里全是upstream prematurely closed connection。后来切到 MQTT连接数从 87 个 TCP 长连接压到 1 个复用连接单台 2 核 4G 的云服务器轻松扛住 300 设备并发消息端到端延迟稳定在 80ms 以内。这就是 MQTT 协议不可替代的价值它不是“另一个通信协议”而是专为低带宽、高延迟、不稳定的网络环境比如 NB-IoT、4G Cat.1、LoRa 终端和海量轻量级客户端设计的发布/订阅模型。它和 HTTP 的本质区别在于HTTP 是“你问我答”的请求-响应式每次交互都要握手、发头、等回包MQTT 是“你发我收”的事件驱动式客户端只管连上 Broker消息代理想发就 publish想收就 subscribeBroker 自动完成路由分发。SpringBoot 对它的支持不是简单封装而是通过spring-integration-mqtt和spring-boot-starter-integration这套成熟生态把 MQTT 的连接管理、序列化、异常重连、QoS 策略全部交由框架托管开发者真正只需要关注“发什么”和“收什么”。所以这个标题里的“主题消息发布与订阅”绝不是教你怎么写两行代码调 API。它背后是一整套工程实践如何设计主题层级让设备可扩展比如sensor/{location}/{type}/{id}/status而不是sensor_status怎么选 QoS 级别平衡可靠性与性能QoS 0 快但可能丢QoS 2 不丢但三步握手开销大为什么必须配置cleanSessionfalse才能让离线设备收到积压消息还有 SpringBoot 启动时如何优雅关闭 MQTT 连接避免 Broker 报错……这些细节才是决定项目能不能从 Demo 跑进真实产线的关键。接下来我会带你从零搭起一个可直接用于工业现场的 MQTT 通信模块所有配置参数都附实测依据所有坑我都替你踩过了。2. 整体架构设计与技术选型逻辑2.1 为什么选 MQTT 而不是 WebSocket 或 CoAP先说结论WebSocket 适合浏览器到服务端的双向通信CoAP 适合资源极度受限的微控制器如 STM32L0而 MQTT 是 SpringBoot 接入中低功耗物联网终端的黄金平衡点。这不是拍脑袋决定的而是基于三组硬指标对比对比维度MQTT v3.1.1WebSocketCoAP连接开销TCP 连接 2 字节固定头TCP 连接 HTTP 握手头UDP 包 4 字节头最小包大小Publish 最小 4 字节文本帧至少 6 字节Confirmable 请求 18 字节QoS 支持原生 QoS 0/1/2无内置重传机制Confirmable/Non-confirmableSpringBoot 生态spring-integration-mqtt 官方维护自动重连、线程池管理需手动集成 netty-websocket重连逻辑需自研eclipse-californium 依赖复杂SpringBoot 无原生 starter我实测过同一台树莓派 4B4GB跑三种协议MQTT 在 200 设备并发下 CPU 占用率 12%内存稳定在 380MBWebSocket 因为每个连接要维持 Session 状态200 并发时内存飙升到 1.2GBGC 频繁导致消息延迟抖动超 500msCoAP 虽然包小但 Californium 的异步回调模型和 SpringBoot 的响应式编程风格冲突严重调试一次消息丢失要翻 3 层源码。最终我们放弃 WebSocket 和 CoAP坚定选 MQTT。2.2 Broker 选型Eclipse Mosquitto vs EMQX vs HiveMQBroker 是 MQTT 的心脏选错等于给系统埋雷。我们对比了三款主流开源 BrokerEclipse MosquittoC 语言编写内存占用极低1000 连接仅 15MB但集群功能弱官方只支持主从复制横向扩展靠第三方插件。EMQXErlang 开发单节点轻松支撑 10 万连接集群自动发现但内存消耗大1000 连接约 300MB对 Java 开发者不友好配置文件是 HOCON 格式。HiveMQ商业版功能强社区版阉割严重比如没有 Web 控制台文档更新慢。我们最终选Mosquitto理由很实在项目 X 的设备规模预估峰值 5000 台Mosquitto 单节点 8GB 内存能稳撑 8000 连接官方 benchmark 数据且它的配置极其简单——整个mosquitto.conf文件核心参数就 5 行listener 1883 allow_anonymous false password_file /etc/mosquitto/passwd persistence true persistence_location /var/lib/mosquitto/更重要的是Mosquitto 的 Docker 镜像只有 5MB启动命令一行搞定docker run -d --name mqtt-broker -p 1883:1883 -v $(pwd)/mosquitto.conf:/mosquitto/config/mosquitto.conf -v $(pwd)/passwd:/mosquitto/config/passwd eclipse-mosquitto而 EMQX 的镜像 200MB启动要配 12 个环境变量。在快速验证阶段轻量即正义。2.3 SpringBoot 集成方案为什么不用 Paho Client 原生 SDK很多教程教你直接用org.eclipse.paho:org.eclipse.paho.client.mqttv3自己 new MqttClient、setCallback、connect。这就像造车时自己炼钢——理论上可行但实际会掉进三个深坑连接生命周期管理缺失Paho Client 的connect()是阻塞方法SpringBoot 启动时若 Broker 未就绪应用直接启动失败而spring-integration-mqtt提供MqttPahoClientFactory支持自动重连、连接池、优雅关闭。消息序列化耦合严重Paho 只认byte[]你要自己写 JSON 序列化/反序列化逻辑且无法统一处理空指针、类型转换异常。缺乏 Spring 生态集成比如无法用EventListener监听连接断开事件不能用Scheduled触发定时发布更没法和Transactional结合做事务性消息。所以我们采用Spring Integration MQTT方案它把 MQTT 客户端包装成 Spring 的MessageChannel和MessageHandler所有操作都变成 Spring Bean 的声明式配置。比如订阅一个主题只需定义一个ServiceActivator方法Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler mqttMessageHandler() { return message - { String payload new String((byte[]) message.getPayload()); System.out.println(收到消息 payload); // 这里可以调用 Service 层业务逻辑 }; }这种设计让 MQTT 通信彻底融入 Spring 的 IoC 和 AOP 体系后续加日志、监控、熔断都水到渠成。3. 核心细节解析与实操要点3.1 主题Topic设计规范别再用test和hello了主题是 MQTT 的路由键设计好坏直接决定系统可维护性。我见过太多项目初期用device/status后期加新设备类型时被迫改代码——因为device/status无法区分不同设备。正确的做法是按业务维度分层用斜杠/划分层级且每层含义明确、可预测。我们为项目 X 设计的主题模板是{domain}/{location}/{deviceType}/{deviceId}/{action}domain业务域如sensor传感器、actuator执行器、gateway网关location物理位置编码如farm-001农场001号、greenhouse-AA号温室deviceType设备类型如temp-humi温湿度、soil-moisture土壤湿度、valve电磁阀deviceId设备唯一 ID如SN20230001action动作类型如status状态上报、command控制指令、config配置下发举个真实例子一台部署在农场001号温室A区的温湿度传感器ID 为 SN20230001它上报状态的主题是sensor/farm-001/greenhouse-A/temp-humi/SN20230001/status而服务器下发校准指令的主题是sensor/farm-001/greenhouse-A/temp-humi/SN20230001/command这样设计的好处是支持通配符订阅。比如运维人员想监控整个农场001的所有传感器状态只需订阅sensor/farm-001////status匹配单层#匹配多层。而sensor/#就能捕获所有传感器消息。注意通配符只能在订阅时用发布时主题必须是完整字符串。提示主题中禁止使用空格、$、#、等特殊字符$开头的主题是系统主题如$SYS/broker/clients/connected普通应用不能用。我们曾因设备固件 bug 把主题拼成sensor/farm 001/temp-humi/SN20230001/status带空格导致 Mosquitto 直接拒绝连接错误日志只显示Invalid topic排查了 3 小时才发现是空格问题。3.2 QoS 级别选择QoS 0 不是“不保证”而是“不重试”QoSQuality of Service是 MQTT 的灵魂但很多人误解为“越高越好”。实际上QoS 0最多一次发送方发完就删不存也不等 ACK。适合传感器上报温度丢一次无所谓下次再报或日志推送少量丢失不影响分析。QoS 1至少一次发送方存消息等接收方发 PUBACK没收到就重发。适合控制指令如“打开阀门”但可能重复比如 PUBACK 丢了发送方重发接收方执行两次。QoS 2恰好一次四步握手PUBLISH → PUBREC → PUBREL → PUBCOMP确保不重不丢。适合金融交易、固件升级等关键场景但延迟高、开销大。我们项目 X 的策略是分场景配置设备上报状态QoS 0省电、低延迟服务器下发控制指令QoS 1允许重复但指令本身幂等比如“设置阀门开度50%”重复执行结果一致固件升级包分片QoS 2必须不丢不重在 SpringBoot 中QoS 通过MqttMessage构造指定MqttMessage message new MqttMessage(payload.getBytes()); message.setQos(1); // QoS 1 mqttTemplate.send(sensor/farm-001/valve/SN20230002/command, message);注意QoS 是发布方和订阅方协商的结果取双方 QoS 的较小值。比如你用 QoS 2 发布但订阅方只支持 QoS 1Broker 会降级为 QoS 1 投递。Mosquitto 默认支持全部 QoS但某些嵌入式 Broker如 PubSubClient for Arduino只支持 QoS 0/1务必提前确认终端能力。3.3 连接参数调优为什么cleanSessionfalse是工业现场的刚需cleanSession参数决定客户端断开后Broker 是否保留其订阅关系和未投递消息。默认true清理会话意味着设备重启后要重新订阅且离线期间发给它的消息全丢。这对手机 App 没问题但对农田里的传感器就是灾难——设备可能因电池耗尽断连 12 小时醒来发现所有控制指令都没收到。我们设为false持久会话并配合sessionExpiryInterval会话过期时间MqttConnectOptions options new MqttConnectOptions(); options.setCleanSession(false); options.setConnectionTimeout(30); // 连接超时 30 秒 options.setKeepAliveInterval(60); // 心跳间隔 60 秒 options.setSessionExpiryInterval(24 * 60 * 60L); // 会话保留 24 小时这里有个关键细节sessionExpiryInterval必须在 CONNECT 报文中显式设置Mosquitto 5.0 才支持老版本无效。我们测试时用的 Mosquitto 2.0设了也没用直到升级到 5.1 才生效。所以务必检查 Broker 版本docker exec mqtt-broker mosquitto -h | head -n 1 # 输出mosquitto version 5.1另外keepAliveInterval不是心跳频率而是“最大无消息间隔”。如果设备 60 秒内没发任何报文包括 PINGREQBroker 就认为它掉线并断开连接。我们设为 60 秒但设备固件实际每 45 秒发一次 PINGREQ留出 15 秒缓冲。4. 实操过程与核心环节实现4.1 环境准备5 分钟搭好本地开发环境步骤 1安装并启动 Mosquitto Broker# 创建配置目录 mkdir -p ~/mqtt-demo/{config,data} # 生成密码文件用户名 admin密码 123456 sudo apt-get install mosquitto mosquitto-clients sudo mosquitto_passwd -c ~/mqtt-demo/config/passwd admin # 编写 mosquitto.conf cat ~/mqtt-demo/config/mosquitto.conf EOF listener 1883 allow_anonymous false password_file /mosquitto/config/passwd persistence true persistence_location /mosquitto/data/ log_dest file /mosquitto/log/mosquitto.log log_type all connection_messages true EOF # 启动容器挂载配置和数据卷 docker run -d \ --name mqtt-broker \ -p 1883:1883 \ -v ~/mqtt-demo/config:/mosquitto/config \ -v ~/mqtt-demo/data:/mosquitto/data \ -v ~/mqtt-demo/log:/mosquitto/log \ eclipse-mosquitto步骤 2创建 SpringBoot 项目2.7.x 版本用 Spring Initializr 选中Spring WebSpring IntegrationLombokSpring Boot DevTools添加 MQTT 依赖到pom.xmldependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency步骤 3配置 MQTT 连接工厂在application.yml中spring: integration: mqtt: client: id: springboot-mqtt-client connection: url: tcp://localhost:1883 username: admin password: 123456 clean-session: false keep-alive: 60 connection-timeout: 30 session-expiry-interval: 86400然后在Configuration类中定义 BeanConfiguration public class MqttConfig { Value(${spring.integration.mqtt.connection.url}) private String url; Value(${spring.integration.mqtt.connection.username}) private String username; Value(${spring.integration.mqtt.connection.password}) private String password; Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[]{url}); options.setUserName(username); options.setPassword(password.toCharArray()); options.setCleanSession(false); options.setConnectionTimeout(30); options.setKeepAliveInterval(60); options.setSessionExpiryInterval(86400L); factory.setConnectionOptions(options); return factory; } Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } Bean public MessageChannel mqttOutputChannel() { return new DirectChannel(); } }4.2 发布消息从 Controller 到 Broker 的完整链路假设我们要实现一个 REST 接口接收 HTTP POST 请求然后转发为 MQTT 消息。例如curl -X POST http://localhost:8080/api/v1/command \ -H Content-Type: application/json \ -d {deviceId:SN20230002,command:OPEN,value:100}步骤 1定义消息实体Data NoArgsConstructor AllArgsConstructor public class CommandRequest { private String deviceId; private String command; private Integer value; }步骤 2Controller 接收并触发发布RestController RequestMapping(/api/v1) public class MqttController { Autowired private MqttPublisher mqttPublisher; PostMapping(/command) public ResponseEntityString sendCommand(RequestBody CommandRequest request) { try { // 构建主题sensor/farm-001/valve/{deviceId}/command String topic String.format(sensor/farm-001/valve/%s/command, request.getDeviceId()); // 构建消息体JSON 字符串 String payload String.format( {\command\:\%s\,\value\:%d,\timestamp\:%d}, request.getCommand(), request.getValue(), System.currentTimeMillis() ); mqttPublisher.publish(topic, payload, 1); // QoS 1 return ResponseEntity.ok(指令已发布); } catch (Exception e) { return ResponseEntity.status(500).body(发布失败 e.getMessage()); } } }步骤 3MqttPublisher 封装发布逻辑Component public class MqttPublisher { Autowired private MqttTemplate mqttTemplate; /** * 发布消息 * param topic 主题 * param payload 负载字符串 * param qos QoS 级别 */ public void publish(String topic, String payload, int qos) { MqttMessage message new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); message.setId(System.nanoTime()); // 用于日志追踪 try { mqttTemplate.send(topic, message); System.out.printf([MQTT-PUB] %s - %s (QoS%d)%n, topic, payload, qos); } catch (MessagingException e) { // 这里要记录详细错误比如 Broker 连接断开、认证失败 System.err.printf([MQTT-PUB-ERROR] %s - %s : %s%n, topic, payload, e.getMessage()); throw new RuntimeException(MQTT 发布失败, e); } } }步骤 4配置 MqttTemplate BeanBean public MqttTemplate mqttTemplate(MqttPahoClientFactory factory) { MqttTemplate template new MqttTemplate(); template.setConnectionFactory(factory); template.setDefaultTopic(default-topic); // 默认主题publish 时可不指定 template.setAsync(true); // 异步发送不阻塞主线程 return template; }实测心得MqttTemplate.send()是同步方法但底层调用 Paho 的publish()是异步的消息先入队列由独立线程发送。所以即使 Broker 暂时不可达send()也不会抛异常而是等重连成功后自动补发。这正是MqttPahoClientFactory的价值——它把复杂的重连逻辑封装掉了。4.3 订阅消息如何让 SpringBoot 自动监听主题订阅的核心是定义MessageChannel和MessageHandler并绑定到主题。步骤 1配置 MQTT 入站通道适配器Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(tcp://localhost:1883, springboot-subscriber); adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); adapter.setQos(1); // 订阅时的 QoS影响消息投递级别 adapter.setOutputChannel(mqttInputChannel()); // 投递到这个 Channel adapter.setClientFactory(mqttClientFactory()); // 订阅多个主题支持通配符 MapString, Integer topics new HashMap(); topics.put(sensor////status, 1); // 所有状态上报 topics.put(sensor////command, 1); // 所有控制指令 adapter.setTopicNames(topics.keySet().toArray(new String[0])); adapter.setTopicQos(new int[]{1, 1}); return adapter; }步骤 2定义消息处理器Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler handler() { return message - { // 解析消息 String topic (String) message.getHeaders().get(mqtt_receivedTopic); byte[] payload (byte[]) message.getPayload(); String content new String(payload, StandardCharsets.UTF_8); System.out.printf([MQTT-SUB] 收到 %s : %s%n, topic, content); // 根据主题路由到不同业务逻辑 if (topic.contains(/status)) { handleStatusMessage(topic, content); } else if (topic.contains(/command)) { handleCommandMessage(topic, content); } }; } private void handleStatusMessage(String topic, String content) { // 解析 JSON存入数据库触发告警等 try { JSONObject json new JSONObject(content); String deviceId extractDeviceId(topic); double temp json.optDouble(temperature, 0.0); double humi json.optDouble(humidity, 0.0); // 保存到数据库此处省略 DAO 调用 System.out.printf(设备 %s 温度 %.1f℃湿度 %.1f%%%n, deviceId, temp, humi); } catch (Exception e) { System.err.println(解析状态消息失败 e.getMessage()); } } private void handleCommandMessage(String topic, String content) { // 处理控制指令比如下发给下位机 try { JSONObject json new JSONObject(content); String deviceId extractDeviceId(topic); String command json.optString(command); int value json.optInt(value, 0); System.out.printf(执行指令设备 %s %s %d%n, deviceId, command, value); // 这里调用串口通信或 Modbus 协议发给 PLC } catch (Exception e) { System.err.println(解析指令消息失败 e.getMessage()); } } private String extractDeviceId(String topic) { // 从 sensor/farm-001/valve/SN20230002/command 提取 SN20230002 String[] parts topic.split(/); return parts.length 5 ? parts[4] : unknown; }步骤 3添加连接状态监听可选但强烈推荐Component public class MqttConnectionListener { EventListener public void handleMqttConnectionEvent(ContextRefreshedEvent event) { System.out.println(MQTT 客户端已连接到 Broker); } EventListener public void handleMqttConnectionLostEvent(MqttConnectionLostEvent event) { System.err.println(MQTT 连接丢失原因 event.getCause().getMessage()); // 这里可以发企业微信告警或触发降级逻辑 } }5. 常见问题与排查技巧实录5.1 连接被拒绝Connection refused的 5 种原因及定位方法这是新手最常遇到的错误org.eclipse.paho.client.mqttv3.MqttException: Connection refused看似简单实则原因多样。我整理了真实排障记录现象可能原因定位命令解决方案启动时报错Connection refused: connectBroker 未运行或端口未暴露netstat -tuln | grep 1883docker ps | grep mqtt启动 Broker 容器检查-p 1883:1883是否正确日志显示Connection refused: Not authorized用户名密码错误或allow_anonymous false但没配密码docker logs mqtt-broker | grep -i auth检查mosquitto.conf的password_file路径用mosquitto_passwd -c重置密码连接成功但立即断开cleanSessiontrue且 Broker 已存在同名 clientIDdocker logs mqtt-broker | grep -i clientid改clientID或设cleanSessionfalse连接超时30秒后报错防火墙拦截 1883 端口telnet localhost 1883本地nc -zv host 1883远程开放防火墙sudo ufw allow 1883连接后无法收发消息主题权限未配置Mosquitto ACL查看mosquitto.conf是否有acl_file配置添加 ACL 文件授权用户读写对应主题实操心得永远先用mosquitto_sub和mosquitto_pub命令行工具验证 Broker。比如# 订阅所有状态 mosquitto_sub -h localhost -p 1883 -u admin -P 123456 -t sensor/# -v # 另开终端发布测试消息 mosquitto_pub -h localhost -p 1883 -u admin -P 123456 -t sensor/test/status -m {temp:25.5}如果命令行能通说明 Broker 没问题问题一定在 SpringBoot 配置如果命令行也不通就专注修 Broker。5.2 消息丢失的典型场景与根因分析消息丢失不是 Bug而是对 MQTT 特性理解不足导致的误用。以下是三个血泪教训场景 1QoS 0 发布但设备要求不丢现象传感器上报温湿度后台偶尔收不到某次数据。根因QoS 0 本就不保证送达网络抖动时消息直接丢弃。解法将上报主题的 QoS 改为 1并在设备端实现 PUBACK 确认逻辑设备收到 PUBACK 才删除本地缓存。场景 2订阅方重启后收不到离线消息现象订阅服务停机 10 分钟期间有 20 条状态消息发到主题重启后一条没收到。根因cleanSessiontrue默认重启后 Broker 当作新会话不投递积压消息。解法订阅时设cleanSessionfalse且 Broker 配置persistence true。场景 3通配符订阅不生效现象订阅sensor//#但sensor/farm-001/temp-humi/SN20230001/status消息收不到。根因只匹配一层sensor//#实际匹配sensor/A/B/C但sensor/farm-001/temp-humi/SN20230001/status有 5 层只占一层#匹配剩余 4 层语法没错。真正原因是Mosquitto 默认禁用通配符订阅安全考虑。解法在mosquitto.conf中添加per_listener_settings true listener 1883 allow_anonymous false password_file /mosquitto/config/passwd # 允许通配符订阅 subscription_pattern sensor/#5.3 性能瓶颈排查当消息延迟突然升高到 2 秒以上我们曾遇到生产环境消息延迟从 80ms 暴涨到 2s排查过程如下第一步确认是网络还是 Broker 问题# 测试 Broker 响应时间用 mosquitto_pub 的 -d 参数看详细日志 time mosquitto_pub -h broker-ip -p 1883 -u admin -P 123456 -t test -m ping -d # 如果 time 显示 2s说明 Broker 或网络有问题第二步检查 Broker 资源# 进入容器看 CPU 和内存 docker exec -it mqtt-broker top -b -n 1 | head -20 # 查看连接数 docker exec mqtt-broker mosquitto_ctrl -u admin -P 123456 connection list | wc -l发现连接数 4200CPU 98%内存 7.8GB8GB 机器。根因是未限制最大连接数恶意扫描器建立了大量空连接。第三步加固 Broker在mosquitto.conf中添加max_connections 5000 # 限制最大连接数 connection_messages false # 关闭连接日志减少 IO log_type error # 只记错误日志第四步优化 SpringBoot 端发现MqttPahoMessageDrivenChannelAdapter的taskExecutor未配置所有消息在一个线程处理。添加Bean public AsyncTaskExecutor mqttTaskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(50); executor.setQueueCapacity(1000); executor.setThreadNamePrefix(mqtt-sub-); executor.initialize(); return executor; } // 在 inbound() Bean 中设置 adapter.setTaskExecutor(mqttTaskExecutor());优化后延迟恢复至 120msCPU 降至 35%。5.4 安全加固 checklist从开发到上线必做的 7 件事MQTT 默认明文传输生产环境必须加固启用 TLS 加密生成证书修改mosquitto.conflistener 8883 cafile /mosquitto/config/ca.crt certfile /mosquitto/config/server.crt keyfile /mosquitto/config/server.key require_certificate falseSpringBoot 连接 URL 改为ssl://broker:8883。禁用匿名访问allow_anonymous false且密码文件用mosquitto_passwd生成SHA-512 加密。最小权限原则为不同角色创建不同用户用 ACL 限制主题访问# acl_file /mosquitto/config/acl user admin topic readwrite sensor/# topic readwrite gateway/# user device1 topic read sensor/farm-001////status topic write sensor/farm-001////command限制连接速率防暴力破解在mosquitto.conf中max_inflight_messages 20 max_queued_messages 100开启连接日志log_type all但上线后改为log_type error防止日志爆炸。SpringBoot 端敏感信息外置密码不要写在application.yml用环境变量docker run -e SPRING_INTEGRATION_MQTT_CONNECTION_USERNAMEadmin \ -e