什么是MQTTMQTTMessage Queuing Telemetry Transport消息队列遥测传输协议是一种基于发布/订阅publish/subscribe模式的轻量级通讯协议该协议构建于TCP/IP协议上可以以极少的代码和有限的带宽为连接远程设备提供实时可靠的消息服务。服务质量qos0“至多一次”消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。1“至少一次”确保消息到达但消息重复可能会发生。2“只有一次”确保消息到达一次。在一些要求比较严格的计费系统中可以使用此级别。MQTT 发布/订阅架构MQTT 采用发布/订阅Publish/Subscribe模式消息的流转由三个核心角色协作完成。下面这张流程图展示了 Publisher、Broker 与 Subscriber 之间的关系以及消息从发布到分发的完整路径发布消息到主题按主题分发按主题分发按主题分发Publisher 发布者Broker 消息代理Subscriber 订阅者 ASubscriber 订阅者 BSubscriber 订阅者 C各角色职责说明Publisher发布者负责向 Broker 发布消息并指定消息所属的主题Topic。发布者无需关心有哪些订阅者也无需知道订阅者的地址只需将消息交给 Broker 即可。Broker消息代理是整个架构的中枢负责接收发布者发来的消息并根据主题匹配规则将消息分发给所有订阅了该主题的客户端。Broker 还承担会话管理、QoS 保障、离线消息缓存等职责。Subscriber订阅者向 Broker 订阅感兴趣的主题当该主题有新消息到达时Broker 会主动推送给订阅者。一个主题可以被多个订阅者同时订阅从而实现一对多的消息广播。MQTT与HTTP/WebSocket对比对比维度MQTTHTTPWebSocket通信模式发布/订阅一对多请求/响应一对一全双工点对点协议开销极低报文头最小仅 2 字节较高每次请求都携带大量头部信息较低握手后头部开销小实时性高服务端主动推送毫秒级延迟低需客户端轮询存在延迟高双向实时推送适用场景物联网、传感器、移动端低功耗通信传统 Web 页面、RESTful API实时聊天、在线协作、股票行情简要说明MQTT 采用发布/订阅模式天然支持一对多广播且报文头极小、省电省带宽非常适合资源受限的物联网设备HTTP 是经典的请求/响应模型每次通信都要重建连接与头部信息实时性依赖轮询适合普通 Web 接口WebSocket 在握手后建立长连接双向实时推送能力强但连接管理相对复杂适合需要高频双向交互的场景。示例代码publicclassMqttClientTest{publicstaticvoidmain(String[]args)throwsMqttException{MqttClientclientnewMqttClient(tcp://127.0.0.1:1883,client,newMemoryPersistence());MqttConnectOptionsoptionsnewMqttConnectOptions();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){}});// 创建消息MqttMessagemessagenewMqttMessage(hello World!.getBytes());// 设置消息的服务质量// 最多一次(0) 至少一次(1) 恰好一次(2)message.setQos(2);// 发布消息client.publish(test,message);// 断开连接client.disconnect();// 关闭客户端client.close();}}断线重连与自动订阅恢复在实际物联网场景中网络波动是常态客户端与 Broker 之间的连接随时可能中断。为保证消息不丢失、业务不中断我们需要在connectionLost回调中实现自动重连并在重连成功后重新订阅之前订阅的主题。publicclassMqttReconnectClient{// 重连间隔单位毫秒这里设置为 5 秒privatestaticfinallongRECONNECT_INTERVAL5000L;// 最大重连次数-1 表示无限重连privatestaticfinalintMAX_RECONNECT_TIMES-1;privateMqttClientclient;privateMqttConnectOptionsoptions;// 需要恢复订阅的主题privateString[]topics{test};publicvoidconnect()throwsMqttException{clientnewMqttClient(tcp://127.0.0.1:1883,reconnect-client,newMemoryPersistence());optionsnewMqttConnectOptions();options.setUserName(admin);options.setPassword(qwer0103.toCharArray());// 设置是否清空 session// 若设置为 falseBroker 会保留客户端的会话与离线消息// 重连后无需重新订阅即可恢复之前的订阅关系// 若设置为 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(){intattempt0;while(MAX_RECONNECT_TIMES-1||attemptMAX_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连接被拒绝用户名或密码错误。解决代码连接前先校验配置并捕获认证异常给出明确提示。MqttConnectOptionsoptionsnewMqttConnectOptions();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。MqttConnectOptionsoptionsnewMqttConnectOptions();// 持久会话Broker 保留订阅关系与离线消息options.setCleanSession(false);// 设置自动重连断线后由 SDK 自动恢复连接options.setAutomaticReconnect(true);client.connect(options);// 使用 QoS 1 或 2 订阅确保离线消息被 Broker 缓存client.subscribe(sensor/#,1);排查建议若仍收不到离线消息可检查 Broker 是否开启了持久会话支持以及客户端是否在断线后及时重连离线时间过长Broker 可能清理会话。
什么是MQTTMQTTMessage Queuing Telemetry Transport消息队列遥测传输协议是一种基于发布/订阅publish/subscribe模式的轻量级通讯协议该协议构建于TCP/IP协议上可以以极少的代码和有限的带宽为连接远程设备提供实时可靠的消息服务。服务质量qos0“至多一次”消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。1“至少一次”确保消息到达但消息重复可能会发生。2“只有一次”确保消息到达一次。在一些要求比较严格的计费系统中可以使用此级别。MQTT 发布/订阅架构MQTT 采用发布/订阅Publish/Subscribe模式消息的流转由三个核心角色协作完成。下面这张流程图展示了 Publisher、Broker 与 Subscriber 之间的关系以及消息从发布到分发的完整路径发布消息到主题按主题分发按主题分发按主题分发Publisher 发布者Broker 消息代理Subscriber 订阅者 ASubscriber 订阅者 BSubscriber 订阅者 C各角色职责说明Publisher发布者负责向 Broker 发布消息并指定消息所属的主题Topic。发布者无需关心有哪些订阅者也无需知道订阅者的地址只需将消息交给 Broker 即可。Broker消息代理是整个架构的中枢负责接收发布者发来的消息并根据主题匹配规则将消息分发给所有订阅了该主题的客户端。Broker 还承担会话管理、QoS 保障、离线消息缓存等职责。Subscriber订阅者向 Broker 订阅感兴趣的主题当该主题有新消息到达时Broker 会主动推送给订阅者。一个主题可以被多个订阅者同时订阅从而实现一对多的消息广播。MQTT与HTTP/WebSocket对比对比维度MQTTHTTPWebSocket通信模式发布/订阅一对多请求/响应一对一全双工点对点协议开销极低报文头最小仅 2 字节较高每次请求都携带大量头部信息较低握手后头部开销小实时性高服务端主动推送毫秒级延迟低需客户端轮询存在延迟高双向实时推送适用场景物联网、传感器、移动端低功耗通信传统 Web 页面、RESTful API实时聊天、在线协作、股票行情简要说明MQTT 采用发布/订阅模式天然支持一对多广播且报文头极小、省电省带宽非常适合资源受限的物联网设备HTTP 是经典的请求/响应模型每次通信都要重建连接与头部信息实时性依赖轮询适合普通 Web 接口WebSocket 在握手后建立长连接双向实时推送能力强但连接管理相对复杂适合需要高频双向交互的场景。示例代码publicclassMqttClientTest{publicstaticvoidmain(String[]args)throwsMqttException{MqttClientclientnewMqttClient(tcp://127.0.0.1:1883,client,newMemoryPersistence());MqttConnectOptionsoptionsnewMqttConnectOptions();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){}});// 创建消息MqttMessagemessagenewMqttMessage(hello World!.getBytes());// 设置消息的服务质量// 最多一次(0) 至少一次(1) 恰好一次(2)message.setQos(2);// 发布消息client.publish(test,message);// 断开连接client.disconnect();// 关闭客户端client.close();}}断线重连与自动订阅恢复在实际物联网场景中网络波动是常态客户端与 Broker 之间的连接随时可能中断。为保证消息不丢失、业务不中断我们需要在connectionLost回调中实现自动重连并在重连成功后重新订阅之前订阅的主题。publicclassMqttReconnectClient{// 重连间隔单位毫秒这里设置为 5 秒privatestaticfinallongRECONNECT_INTERVAL5000L;// 最大重连次数-1 表示无限重连privatestaticfinalintMAX_RECONNECT_TIMES-1;privateMqttClientclient;privateMqttConnectOptionsoptions;// 需要恢复订阅的主题privateString[]topics{test};publicvoidconnect()throwsMqttException{clientnewMqttClient(tcp://127.0.0.1:1883,reconnect-client,newMemoryPersistence());optionsnewMqttConnectOptions();options.setUserName(admin);options.setPassword(qwer0103.toCharArray());// 设置是否清空 session// 若设置为 falseBroker 会保留客户端的会话与离线消息// 重连后无需重新订阅即可恢复之前的订阅关系// 若设置为 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(){intattempt0;while(MAX_RECONNECT_TIMES-1||attemptMAX_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连接被拒绝用户名或密码错误。解决代码连接前先校验配置并捕获认证异常给出明确提示。MqttConnectOptionsoptionsnewMqttConnectOptions();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。MqttConnectOptionsoptionsnewMqttConnectOptions();// 持久会话Broker 保留订阅关系与离线消息options.setCleanSession(false);// 设置自动重连断线后由 SDK 自动恢复连接options.setAutomaticReconnect(true);client.connect(options);// 使用 QoS 1 或 2 订阅确保离线消息被 Broker 缓存client.subscribe(sensor/#,1);排查建议若仍收不到离线消息可检查 Broker 是否开启了持久会话支持以及客户端是否在断线后及时重连离线时间过长Broker 可能清理会话。
