初步认识mqtt

什么是MQTT

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。

• 服务质量（qos）
  0：“至多一次”，消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。
  1：“至少一次”，确保消息到达，但消息重复可能会发生。
  2：“只有一次”，确保消息到达一次。在一些要求比较严格的计费系统中，可以使用此级别。

MQTT 发布/订阅架构

MQTT 采用发布/订阅（Publish/Subscribe）模式，消息的流转由三个核心角色协作完成。下面这张流程图展示了 Publisher、Broker 与 Subscriber 之间的关系，以及消息从发布到分发的完整路径：

各角色职责说明：

• Publisher（发布者）：负责向 Broker 发布消息，并指定消息所属的主题（Topic）。发布者无需关心有哪些订阅者，也无需知道订阅者的地址，只需将消息交给 Broker 即可。
• Broker（消息代理）：是整个架构的中枢，负责接收发布者发来的消息，并根据主题匹配规则，将消息分发给所有订阅了该主题的客户端。Broker 还承担会话管理、QoS 保障、离线消息缓存等职责。
• Subscriber（订阅者）：向 Broker 订阅感兴趣的主题，当该主题有新消息到达时，Broker 会主动推送给订阅者。一个主题可以被多个订阅者同时订阅，从而实现一对多的消息广播。

MQTT与HTTP/WebSocket对比

简要说明：MQTT 采用发布/订阅模式，天然支持一对多广播，且报文头极小、省电省带宽，非常适合资源受限的物联网设备；HTTP 是经典的请求/响应模型，每次通信都要重建连接与头部信息，实时性依赖轮询，适合普通 Web 接口；WebSocket 在握手后建立长连接，双向实时推送能力强，但连接管理相对复杂，适合需要高频双向交互的场景。

示例代码

publicclassMqttClientTest{publicstaticvoidmain(String[]args)throwsMqttException{MqttClientclient=newMqttClient("tcp://127.0.0.1:1883","client",newMemoryPersistence());MqttConnectOptionsoptions=newMqttConnectOptions();options.setUserName("admin");options.setPassword("qwer0103".toCharArray());// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，// 这里设置为true表示每次连接到服务器都以新的身份连接options.setCleanSession(true);// 设置超时时间 单位为秒options.setConnectionTimeout(10);// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制options.setKeepAliveInterval(20);// 连接client.connect(options);// 订阅client.subscribe("test");client.setCallback(newMqttCallback(){@OverridepublicvoidconnectionLost(Throwablethrowable){System.out.println("connection success .....");}@OverridepublicvoidmessageArrived(Strings,MqttMessagemqttMessage)throwsException{System.out.println("topic is "+s+",Message is "+mqttMessage.toString());}@OverridepublicvoiddeliveryComplete(IMqttDeliveryTokeniMqttDeliveryToken){}});// 创建消息MqttMessagemessage=newMqttMessage("hello World!".getBytes());// 设置消息的服务质量// 最多一次(0) 至少一次(1) 恰好一次(2)message.setQos(2);// 发布消息client.publish("test",message);// 断开连接client.disconnect();// 关闭客户端client.close();}}

断线重连与自动订阅恢复

在实际物联网场景中，网络波动是常态，客户端与 Broker 之间的连接随时可能中断。为保证消息不丢失、业务不中断，我们需要在connectionLost回调中实现自动重连，并在重连成功后重新订阅之前订阅的主题。

publicclassMqttReconnectClient{// 重连间隔：单位毫秒，这里设置为 5 秒privatestaticfinallongRECONNECT_INTERVAL=5000L;// 最大重连次数，-1 表示无限重连privatestaticfinalintMAX_RECONNECT_TIMES=-1;privateMqttClientclient;privateMqttConnectOptionsoptions;// 需要恢复订阅的主题privateString[]topics={"test"};publicvoidconnect()throwsMqttException{client=newMqttClient("tcp://127.0.0.1:1883","reconnect-client",newMemoryPersistence());options=newMqttConnectOptions();options.setUserName("admin");options.setPassword("qwer0103".toCharArray());// 设置是否清空 session：// 若设置为 false，Broker 会保留客户端的会话与离线消息，// 重连后无需重新订阅即可恢复之前的订阅关系；// 若设置为 true，每次连接都以全新身份接入，重连后必须手动重新订阅。options.setCleanSession(false);options.setConnectionTimeout(10);options.setKeepAliveInterval(20);// 设置自动重连：Broker 断开后由客户端自动尝试重连options.setAutomaticReconnect(true);client.setCallback(newMqttCallback(){@OverridepublicvoidconnectionLost(Throwablethrowable){System.out.println("连接丢失："+throwable.getMessage());// 手动重连逻辑：按固定间隔重试，直到成功reconnect();}@OverridepublicvoidmessageArrived(Stringtopic,MqttMessagemessage)throwsException{System.out.println("topic is "+topic+",Message is "+message.toString());}@OverridepublicvoiddeliveryComplete(IMqttDeliveryTokentoken){}});client.connect(options);// 首次连接成功后订阅主题subscribeTopics();}/** * 手动重连：循环尝试，直到连接成功或达到最大次数 */privatevoidreconnect(){intattempt=0;while(MAX_RECONNECT_TIMES==-1||attempt<MAX_RECONNECT_TIMES){attempt++;try{Thread.sleep(RECONNECT_INTERVAL);if(!client.isConnected()){client.connect(options);}// 重连成功后恢复订阅subscribeTopics();System.out.println("重连成功，已恢复订阅");break;}catch(Exceptione){System.out.println("第 "+attempt+" 次重连失败："+e.getMessage());}}}/** * 订阅主题（首次连接与重连成功后都会调用） */privatevoidsubscribeTopics()throwsMqttException{client.subscribe(topics);}}

重连与 session 恢复策略说明：

• 重连间隔：示例中设置为 5 秒（RECONNECT_INTERVAL），避免网络抖动时频繁重连给 Broker 造成压力。实际生产环境可根据业务容忍度调整，通常建议 3~10 秒。
• session 恢复：setCleanSession(false)是关键。Broker 会持久化客户端的会话状态，包括订阅关系和离线期间的消息（QoS 1/2），重连后无需重新订阅即可自动恢复；若使用setCleanSession(true)，则每次重连都必须手动重新订阅，且离线消息会丢失。
• 自动重连 vs 手动重连：setAutomaticReconnect(true)由客户端 SDK 内置机制处理重连，但重连成功后不会自动恢复订阅，因此仍需在connectionLost中配合手动重连或重连成功后的订阅恢复逻辑，确保订阅关系不丢失。

常见问题与排查

在实际接入 MQTT 的过程中，除了断线重连，还会遇到一些典型的连接与消息问题。下面列出 3 个最常见的问题，并给出原因分析与对应的解决代码片段。

问题 1：连接被 Broker 拒绝（用户名/密码错误）

原因分析：Broker 开启了认证校验，客户端传入的用户名或密码与 Broker 配置不一致，导致连接握手阶段被拒绝，抛出MqttSecurityException或连接返回码为 5（连接被拒绝，用户名或密码错误）。

解决代码：连接前先校验配置，并捕获认证异常给出明确提示。

MqttConnectOptionsoptions=newMqttConnectOptions();options.setUserName("admin");options.setPassword("qwer0103".toCharArray());options.setConnectionTimeout(10);try{client.connect(options);System.out.println("连接成功");}catch(MqttSecurityExceptione){// 认证失败：检查用户名/密码是否与 Broker 配置一致System.err.println("认证失败，请检查用户名或密码："+e.getMessage());}catch(MqttExceptione){// 其他连接异常：检查 Broker 地址、端口、网络连通性System.err.println("连接异常，返回码："+e.getReasonCode()+"，原因："+e.getMessage());}

问题 2：订阅后收不到消息（主题通配符不匹配）

原因分析：订阅主题与发布主题不匹配。MQTT 主题支持通配符：+匹配单层、#匹配多层。若订阅使用sensor/+/temp，而发布到sensor/room1/humidity，则因层级不匹配而收不到消息；此外，#必须位于主题末尾。

解决代码：订阅前先确认主题层级，必要时用通配符覆盖多个层级。

// 订阅单层通配：匹配 sensor/room1/temp、sensor/room2/tempclient.subscribe("sensor/+/temp");// 订阅多层通配：匹配 sensor/ 下所有主题client.subscribe("sensor/#");// 精确订阅：只匹配 sensor/room1/tempclient.subscribe("sensor/room1/temp");

问题 3：重连后消息丢失（cleanSession 设置不当）

原因分析：setCleanSession(true)时，Broker 不保留客户端会话，断线期间的离线消息（QoS 1/2）会被丢弃，重连后也无法恢复订阅关系，导致消息丢失。若业务要求断线期间消息不丢，必须使用setCleanSession(false)并配合 QoS 1/2。

解决代码：使用持久会话并设置合理的 QoS。

MqttConnectOptionsoptions=newMqttConnectOptions();// 持久会话：Broker 保留订阅关系与离线消息options.setCleanSession(false);// 设置自动重连，断线后由 SDK 自动恢复连接options.setAutomaticReconnect(true);client.connect(options);// 使用 QoS 1 或 2 订阅，确保离线消息被 Broker 缓存client.subscribe("sensor/#",1);

排查建议：若仍收不到离线消息，可检查 Broker 是否开启了持久会话支持，以及客户端是否在断线后及时重连（离线时间过长，Broker 可能清理会话）。