说到物联网后端开发, MQTT协议是个好东西, 因为它开销低, 也不怎么占用带宽, 所以设备跟服务器通信的时候, 大家都特别喜欢用它, 把它当成首选方案。不过呢, 很多干后端开发的工程师, 在实际动手做项目的时候, 总会被两个特别核心的问题给卡住脖子,下不去手, 第一个问题是根本感知不到设备的异常离线状况, 第二个问题是网络一旦断开了, 发送的消息就没了, 彻底丢失了。好在的是, 这两个让人头疼的问题, 用遗愿消息Last Will还有心跳保活机制, 就能解决得妥妥的。时间来到了二〇二六年, 物联网项目迎来了爆发式增长的阶段, MQTT协议的落地需求出现了激增的情况, 特别是在工业物联网以及智能家居这些场景里面, 对于设备连接稳定性的要求已经达到了一个新的高度, 本文结合了最新的实战案例, 从原理方面一直延伸到代码实现的部分, 手把手地教大家去实现遗愿消息与心跳保活的功能, 以此来规避掉那百分之九十的常见坑, 这一内容是适合后端开发人员和物联网开发者直接进行套用的。为什么这两个机制是MQTT技术实现落地应用的核心要素?MQTT协议是基于TCP/IP来构建的, 它以轻量级的发布以及订阅为主要模式, 被非常广泛地应用在了像传感器这样的设备领域, 还有智能家居方面, 亦或是车联网等领域, 可是这些场景普遍存在网络不够稳定、设备的计算能力有限的痛点问题, 而遗愿消息以及心跳保活机制这两个技术, 恰恰就是为解决以上这些痛点才会被创造出来的。我们从近期几个平台上面那些阅读量很高的文章的具体数据能够看出来, 凡是跟MQTT这项技术有关系的文章, 大家互动特别频繁的点主要是集中在这两个方面, 第一是实际的动手操作和落地应用, 第二是遇到问题以后的排查处理过程。特别是那些专门讲解遗愿消息还有心跳保活这些内容的文章, 人们收藏它们的数量, 以及转发给别人的比例, 远远超过了普通的、基础的mqtt教程里的内容。这其中的核心原因其实并不是别的, 主要是因为这两个具体的机制直接决定了一个物联网系统到底稳不稳定, 到底可不可靠。对于做后端开发的人来说, 这些知识就是那种必须用到但又非常容易出错的关键知识点。从专业角度去进行分析的话, 二者的核心价值的体现可以归纳为三个方面。第一个方面是关于心跳保活, 它解决了所谓的“TCP假连接”这个难题, 从而避免了服务器产生误判, 不会把设备是否在线的状态搞错了。第二个方面是关于遗愿消息, 它为了解决“设备异常离线时没有通知”这一类问题提供了办法, 这样一来就能确保上下游的系统之间能够正常地进行联动配合。第三个方面是将二者结合在一起考虑, 这种做法可以大幅度地物联网系统的运维成本降低下去, 同时也减少了由于断开连接导致的数据丢失以及业务出现异常这样的一些问题。大家得注意一点, 现在好多开发者都搞错了, 他们觉得TCP那个自带的保活功能能代替MQTT协议层里弄的心跳保活, 其实不是那么回事儿, 因为TCP保活这东西用的资源比较多, 比较费劲, 而且它用在那些带宽窄、延迟又高的物联网环境里头, 根本不合适, 反观MQTT心跳保活只要两个字节的报文就够了, 这样就能把传输时候消耗的东西降到最低了。你如果去看一眼那个叫底层逻辑的东西, 你就会发现关于2.1这个心跳保活机制的解释是非常好理解的, 因为它就像是一个设备和一个服务器在进行定时的说话和互动。MQTT这个心跳保活机制, 它的核心作用在于借助计时器和相应的报文, 去保持设备与服务器端之间的连接状态不被断开, 同时还能用来检测这个连接到底是不是还有效, 下面来说一说它底层所依据的那种逻辑情况是怎么样的。1. 在连接初始化阶段, 客户端也就是所谓的设备, 会在报文里面设置一个字段, 这个字段的单位是秒, 用来指定最长无通信时间间隔。比如把这个时间设置为60秒, 这就意味着设备必须在每60秒的时间段内, 跟服务器完成至少一次的通信操作。2. 如果在规定的期间以内, 设备没有发送任何业务消息例如数据上报之类的操作, 那么该设备就会主动给服务器发送一个报文, 这个报文被叫做心跳请求。3. 当服务器这边把报文接收到了之后, 它会立刻做出反应, 给出去一个心跳响应的那个报文也就是心跳响应, 这样就完成了一整个从开始到结束的心跳交互过程, 接着就把那个计时器重置了, 重新开始计算时间。4. 如果服务器在一点五倍时间这个时间段之内, 既没有收到来自设备的业务消息, 也没有收到报文, 就会判定设备已经和连接断开了, 这个时候要清理这个设备的会话信息。5. 如果在设备发送数据之后, 在合理的等待时间长度内没有收到回应的话, 那么就将这种情况判定为与服务器之间的连接出现了异常情况, 随后就会去启动并且执行那些用于重新建立连接的逻辑处理流程。这里做一个关键的补充说明, 那个字段的取值范围被限定在0到65535秒的区间里面, 当这个取值为0的时候, 它就代表着设备不再发送心跳报文了, 在这种情况下, 服务器就不能主动地去检测设备是否处于离线状态了, 它只能通过TCP连接意外断开这种方式来被动地感知到设备已离线, 所以大家不推荐在实际项目中去使用这样的设置。2.第二条是遗愿消息部分, 这也被称作Last Will, 它就相当于设备在即将停止运行之前发出的最后一句告别留言。所谓的遗愿消息这个概念, 它的本质含义是这样的。当设备处于正常连接状态的时候, 它会预先把一条托付性质的消息发送给服务器。万一设备发生了非自愿的断开连接的情况。比如网络中断、设备断电或者是程序崩溃。那么服务器就会代替设备。把这条消息发布到指定的主题里面。这样其他相关的设备或者系统就能收到通知了。关于这背后的底层逻辑其实如下所示。1. 当设备向服务器发起连接请求的时候, 它需要在传输的报文里面配置有关遗愿相关的多个参数。首先要把那个指示位的字段值设置为1, 然后还要明确指定发送遗嘱消息的目标主题叫什么名字, 同时也要写好这份遗嘱里具体的文本内容, 除此之外还必须选定这份遗嘱消息在传输时的服务质量等级是多少最后再设置一下保留标志的状态。2. 当服务器收到了那个连接请求之后, 它会把跟这个设备的遗愿消息有关的那些参数给保存起来, 但是呢, 它并不会立刻就把这些东西给发布出去。3. 当服务器检测到设备非自愿断开连接, 比如心跳超时, 或者TCP连接异常中断的时候, 并且还没有收到设备主动发送的报文时, 会触发遗愿消息发布逻辑, 将预先保存好的遗愿消息, 发布到指定的那个遗愿主题中。4. 所有订阅了该遗愿主题的客户端, 比如说其他的设备也好, 还有后端的业务系统也罢, 它们都会收到这条属于该遗愿消息的内容, 借此来感知到那个特定的设备已经处于异常情况下的离线状态了。5. 如果设备进行正常的断开连接的状况, 也就是由设备主动发送报文的情况下, 服务器这边就不会发布所谓的遗愿消息, 同时还会把与该设备所关联的遗愿消息参数进行清理的处理操作。核心的区别在于, 遗愿消息这种状态只有在设备非自愿断开连接的时候才会被触发, 如果设备是正常断开的话, 它就不会触发这个机制, 这个特性能够有效地区分出设备是主动选择下线还是发生了异常的掉线情况, 进而避免了业务系统因此产生误判。手把手落地基于EMQ X 本次实战是使用了EMQ X 还有 paho-mqtt客户端这种方式来模拟物联网设备以及服务器之间的连接, 还有心跳交互, 以及遗愿消息触发的场景。这个步骤是非常清晰的, 也是可以复制的, 后端开发者可以直接把它放到实际项目中去使用。3.准备好环境这个步骤, 总共只需要花三步就能够把它搞定。1. 关于安装EMQ X 这个面向服务器端的软件, 我们建议大家去采用那个比较省事的快速部署办法, 具体的操作步骤就是执行下面显示的这段命令代码, 你把它复制过来以后, 可以直接拿去运行。# 拉取EMQ X最新镜像docker pull emqx/emqx:latest# 启动容器映射1883端口MQTT默认端口docker run -d --name emqx -p 1883:1883 emqx/emqx:latest2. 需要执行安装操作, 具体是为那个代表设备端的程序装上这个客户端软件, 而在这套系统里, 它扮演的角色就是模拟一个MQTT客户端。pip install paho-mqtt3. 我们对环境实施了验证, 在启动了EMQ X之后, 前往http://服务器IP:18083登录EMQ X控制台页面。使用用户名admin和密码进行登录, 成功登录后就可以观察到此时客户端所处的连接状态。3.这是第二个实战项目, 它涉及到了心跳保活功能的实现。需要让那台模拟出来的设备去连着EMQ X, 设定成每间隔30秒钟做一次那种心跳方面的交互动作, 然后要把那个模拟的网络中断的情况给弄出来, 好让服务器那边能够发现这设备已经不在线了, 也就是所谓的已经离线的情况。import paho.mqtt.client as mqttimport time# 1. 定义回调函数心跳响应、连接状态等def on_connect(client, userdata, flags, rc):连接成功回调print(f设备连接成功响应码{rc})# 连接成功后订阅心跳状态主题可选用于接收服务器通知client.subscribe(device/heartbeat/status, qos0)def on_pingresp(client, userdata, msg):心跳响应回调print(f收到服务器心跳响应当前时间{time.strftime(%H:%M:%S)})def on_disconnect(client, userdata, rc):断开连接回调if rc ! 0:print(f意外断开连接断开码{rc}将尝试重连...)else:print(正常断开连接)# 2. 初始化MQTT客户端client mqtt.Client(client_idiot_device_001, clean_sessionTrue)# 设置心跳保活时间30秒核心参数client.keepalive 30# 绑定回调函数client.on_connect on_connectclient.on_pingresp on_pingrespclient.on_disconnect on_disconnect# 3. 连接EMQ X Broker替换为自己的服务器IPclient.connect(host127.0.0.1, port1883, keepalive30)# 4. 启动客户端循环维持连接处理心跳、消息等try:client.loop_forever() # 阻塞式循环自动处理心跳except KeyboardInterrupt:# 手动中断主动断开连接不会触发遗愿消息client.disconnect()print(手动中断设备正常下线)具体的实施以及验证的相关操作流程如下。1. 执行以上的代码, 观察控制台输出的情况, 可以发现如果没有业务消息存在的话, 那么每间隔30秒的时间段就会接收一次来自系统的心跳响应。2. 用手动方式按CtrlC键来关闭代码的时候, 控制台会显示出正常断开连接这几个字, 与此同时在EMQ X控制台里面那个客户端的状态就会变成为离线。3. 我们先模拟网络中断的情况, 在运行代码之后, 立即断开与服务器的网络连接, 接着等待45秒钟, 这个等待的时间是标准时间的1.5倍, 当时间到达后, EMQ X控制台就会自动将客户端的状态标记为离线, 在此时, 代码中设置的逻辑会被激活并触发重连操作。3.第3个实战项目, 也就是实战2, 其内容是通过将遗愿消息与心跳保活机制进行组合搭配这种方式来加以实现。在心跳保活的基础上, 还需要设置遗愿消息, 用来模拟设备异常断电这种非自愿断开的情形, 从而触发遗愿消息的发布, 让其他客户端能够接收到这个遗愿消息, 以此感知到该设备已经离线。整个体系划分成了两个不同的角色, 其中一个角色是设备客户端, 它主要的功能包括发布遗愿消息以及维持心跳信号。另一个角色则是监控客户端, 它需要订阅相应的遗愿主题, 并且在过程中接收关于设备离线的通知信息。3.第3.1小节, 关于设备终端软件的源代码部分, 这是最核心的内容。import paho.mqtt.client as mqttimport timedef on_connect(client, userdata, flags, rc):print(f设备连接成功响应码{rc})# 连接成功后发布一条在线通知可选client.publish(device/status/iot_device_001, payloadonline, qos1, retainTrue)def on_pingresp(client, userdata, msg):print(f心跳响应正常当前时间{time.strftime(%H:%M:%S)})# 初始化客户端clean_sessionTrue表示断开连接后清理会话client mqtt.Client(client_idiot_device_001, clean_sessionTrue)# 核心设置遗愿消息参数# Will Flag自动设为1遗愿主题、消息、QoS、保留标志client.will_set(topicdevice/offline/notify, # 遗愿主题payload{device_id: iot_device_001, status: offline, time: %s} % time.strftime(%Y-%m-%d %H:%M:%S), # 遗愿消息内容JSON格式便于后端解析qos1, # 遗愿消息QoS等级推荐1至少一次retainTrue # 保留遗愿消息新订阅该主题的客户端可立即收到)# 设置心跳保活时间30秒client.keepalive 30# 绑定回调函数client.on_connect on_connectclient.on_pingresp on_pingresp# 连接EMQ Xclient.connect(host127.0.0.1, port1883, keepalive30)# 启动循环维持连接try:client.loop_forever()except KeyboardInterrupt:# 正常断开不触发遗愿消息client.publish(device/status/iot_device_001, payloadoffline, qos1, retainTrue)client.disconnect()print(手动中断设备正常下线)3.3.2 监控客户端代码接收遗愿消息import paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc):print(f监控客户端连接成功响应码{rc})# 订阅遗愿消息主题接收所有设备的离线通知client.subscribe(device/offline/notify, qos1)def on_message(client, userdata, msg):接收消息回调核心接收遗愿消息print(f\n收到设备离线通知遗愿消息)print(f主题{msg.topic})print(f消息内容{msg.payload.decode(utf-8)})# 初始化监控客户端monitor_client mqtt.Client(client_idmonitor_client_001, clean_sessionTrue)monitor_client.on_connect on_connectmonitor_client.on_message on_message# 连接EMQ Xmonitor_client.connect(host127.0.0.1, port1883, keepalive60)# 启动循环持续接收消息monitor_client.loop_forever()3.3.3 进行实战方面的验证, 这是其中的一个关键步骤。1. 咱们先把那个监控客户端的代码给跑起来, 然后等着看控制台那边, 等屏幕上面显示出“连接成功”这几个字, 这就代表可以接着往后走, 等着接收消息了。2. 然后呢, 再去运行一下那个设备端的客户端代码, 之后在控制台上会出现表示“连接成功”的情况, 再接着就会发现, 每隔30秒的时间就能收到一个心跳的响应反馈。3. 在进行模拟设备异常断电的操作过程中, 可以直接关闭设备的客户端代码, 但是不要使用CtrlC这个组合键, 而是直接选择关闭终端窗口这样的方式, 那么在这种情况之下, 所对应的设备就会属于非自愿断开的状态。4. 在等待了45秒, 也就是将时间扩大为原来的1.5倍之后, 监控客户端这边肯定会收到那个所谓的遗愿消息, 这个消息里面会明确地显示出设备的ID以及显示设备离线的那个具体时间信息等详细信息。5. 登录EMQ X 控制台以后, 你会发现设备客户端的状态变成了离线。在这个阶段, 遗愿主题里的“//”会把最新的离线消息给保留着。有90%的开发者都曾踩过的坑, 以及相应的规避方案。把近期项目里的实际做法, 加上各个平台那边的搞开发的人说的反馈, 加在一起, 整理出了6个经常出现的大问题, 每个大问题都配上了具体的躲开办法, 可以直接绕过去, 提升开发效率。第一个需要注意的地方是坑, 也就是指把数值设置得过大或者过小。如果你把时间设置得太小, 比如说5秒吧, 那就会导致心跳交互变得特别频繁, 这样的话, 就会增加设备的功耗, 而且也会带来网络方面的开销方面的问题。反过来讲, 如果你把时间设置得太过大了, 例如3600秒这种程度, 就会出现一种状况, 就是服务器没法及时地检测到设备已经离线的情况了, 这就不可避免地会延误后面的业务处理过程。为了规避相关风险, 需要依据具体场景进行合理设置, 推荐的取值范围是三十到一百二十秒, 在工业物联网或者移动设备这些网络不太稳定的情况下, 取值范围应设定为三十秒到六十秒, 而在智能家居这种网络比较稳定的环境中, 取值范围则设定为六十秒到一百二十秒。坑点2使用TCP保活替代MQTT心跳保活TCP的保活机制探测间隔特别长, 默认是2小时, 加上报文开销很大。它不适合用在低带宽和高延迟的物联网场景里。这种情况容易让TCP连接变成假连接, 也就是说, 看起来连接还在维持着, 但是设备实际上已经离线了。规避方案的具体内容是: 要使用MQTT协议层里面那个心跳保活机制, 并且必须把TCP保活给禁用掉, 或者只是把它当作一个兜底的手段, 这样做的目的是为了能够确保设备的离线情况可以被及时发现。在这一点上, 大家可能会遇到一个坑, 那就是遗愿消息所使用的QoS等级设置得并不是很合理, 需要引起注意。把遗愿消息的QoS设置成0, 也就是至多一次这种模式, 这个操作有可能让遗愿消息遭到丢失, 这样一来监控系统就没法察觉到设备已经离线了, 但是要是将其设置成2, 也就是只有一次那种保障机制的话, 这就只会白白增加服务器端的开销压力, 这样做其实完全没有必要。关于规避方案的情况, 遗愿消息的QoS属性被统一设定为数值1, 也就是代表至少发生一次的意思, 这样做的目的一方面能够保证相关消息彻底不会发生丢失现象, 另一方面也能够对服务器端的开销压力实施有效的控制措施, 因此这种设置非常适宜应对绝大多数物联网应用场景下的实际要求。这个坑点第4条, 说的是存在和那个遗愿消息保留功能发生冲突的情况。将设置项调整为真值, 也就是开启断开连接后自动清理会话的功能, 与此同时, 再把遗愿消息的选项也设置为真, 这样的操作组合会引发一个后果, 那就是当设备重新建立连接的时候, 之前在离线期间生成的那些作为最后通知的遗愿消息会被清除掉, 从而导致新注册订阅关系的客户端程序无法接收到关于历史状态下离线的通知信息。为了避免这个问题的发生, 如果你们希望保留遗愿消息的话, 就得把参数设为False, 此外还要确保服务器那边专门为该设备留存相关的会话信息但是呢, 假如说并不需要将这个会话给保留下来, 那么只需把参数设为False就可以了。第5个需要注意的地方是, 在设备重新连接的时候, 没有把之前留下的那些所谓的遗愿消息给重置掉。设备在发生异常离线之后又重新建立了连接, 可是这一过程里面并没有针对遗愿消息进行重新配置, 这就造成了在后续再次出现异常离线的情况时, 服务器端不能够正常地发布该条遗愿消息的问题存在。针对这个问题, 我们要拿出的解决办法是在那个回调函数里面去重新设置一下那个遗愿消息, 哪怕说在前面的第一次设置的时候已经设置过这个事了也要再去设置一遍, 这样做的目的是为了要保证在系统重新连接上以后, 那个遗愿消息还是有效的。第6个需要特别注意的地方在于, 很多人并没有对心跳超时之后的重连策略给予足够的重视。设备的连接状态因出现心跳信号延迟超限的情况而发生了断开行为, 并且由于系统内部没有针对这种状况建立相应的重新进行连接的策略, 直接产生的结果是使得设备没有办法依靠自身的力量去自动地完成对连接的还原与恢复工作, 这就必然需要人工介入来进行处理。为了采取规避方案, 需要在回调函数的地方添加重连逻辑, 建议采用指数退避重连方式, 比如分别在1秒后、3秒后、5秒后依顺序进行重连, 这样做可以避免因频繁重连而占用网络资源。总结在物联网后端开发的范围内, 有一个核心的技术, 这个技术的名字叫作保障设备连接稳定性, MQTT协议的遗愿消息和心跳保活机制就是这个核心, 这二者是相辅相成的关系。心跳保活负责检测连接有效性。遗愿消息负责感知设备异常离线。当这两个东西结合在一起使用的时候, 可以大幅提升物联网系统的可靠性, 可以提升物联网系统的可运维性。这篇文章已经比较彻底地把这两个机制里边的关键信息全都讲清楚了, 那些拿来就能用在真正项目里面的代码示例也可以直接照着写, 能帮大多数人避开不少以前经常遇到的麻烦。到了2026年, 物联网技术的热度还会持续往上走。这时候, 大家对于把MQTT协议真正落地使用的要求, 就会变得越来越多。你如果能把遗愿消息还有心跳保活这两个东西的实战窍门给摸透了, 不光能让你搞项目开发的时候更顺手、效率更高, 还能防止因为设备连不上网、掉线这种破事儿, 导致咱们的业务遭受到不该有的损失和亏损。最后再进行提问互动, 这样有助于提升头条号的互动率, 你在利用MQTT进行方案落地的过程中, 还有没有其他方面遇到了问题呢? 比如心跳保活或者遗愿消息相关的坑。请在评论区里留言, 大家一起交流并探讨解决方案。