3. 心跳协议设计：基于MQTT的心跳包格式设计（Topic、Payload）

好，咱们接着聊。设备离线检测的核心，说白了就是「心跳」。那心跳包长什么样？怎么发？发到哪？这就是本节要解决的问题。

我个人习惯，在设计心跳协议时，先定Topic，再定Payload。顺序不能乱。Topic决定了消息的路由，Payload决定了消息的内容。两者缺一不可。

3.1 Topic设计：分层、清晰、可扩展

Topic是MQTT的灵魂。设计不好，后期维护就是灾难。我在项目中见过太多「扁平化」的Topic设计，比如：

device/heartbeat/device001
device/heartbeat/device002
device/heartbeat/device003

这种设计，设备一多，管理就乱套了。你想想看，如果有一万台设备，你需要在服务端订阅一万个Topic？不现实。

我建议采用分层Topic结构。举个例子：

v1/{productKey}/{deviceName}/heartbeat

这里：

• v1：协议版本号。方便以后升级。
• {productKey}：产品标识。同一类设备共享一个Key。
• {deviceName}：设备唯一标识。比如SN号或MAC地址。
• heartbeat：固定后缀，表示这是心跳消息。

这样做的好处很明显：

• 服务端可以订阅 v1/{productKey}/+/heartbeat，一个通配符搞定所有设备。
• 如果某个产品线出问题，可以单独订阅该产品的所有心跳，方便排查。
• 版本号的存在，让协议演进变得平滑。

3.2 Payload设计：轻量、自描述、可扩展

Payload是心跳包的实际内容。嗯，这里要注意：心跳包不是数据上报包，不要塞太多东西。我见过有人把设备的所有传感器数据都塞进心跳包，结果一个包几十KB，完全违背了心跳的初衷。

心跳包应该只包含「证明设备活着」的最小信息。我个人常用的格式是JSON，因为它自描述、易解析。但如果你对性能有极致要求，可以考虑Protobuf或MessagePack。

一个典型的心跳包Payload如下：

{
    "id": "msg_20240301120000_001",
    "type": "heartbeat",
    "timestamp": 1709270400000,
    "status": {
        "battery": 85,
        "rssi": -65,
        "uptime": 3600
    },
    "version": "1.0.0"
}

字段说明：

字段	类型	说明
id	string	消息唯一ID，用于去重和追踪
type	string	固定为"heartbeat"，方便服务端快速识别
timestamp	int64	Unix毫秒时间戳，服务端据此判断是否超时
status.battery	int	电量百分比（0-100），可选字段
status.rssi	int	信号强度（dBm），可选字段
status.uptime	int	设备运行时长（秒），可选字段
version	string	固件版本号，方便OTA管理

3.3 心跳频率与QoS选择

心跳频率怎么定？这没有标准答案。我一般遵循「3倍超时原则」：

• 假设服务端判定设备离线的时间窗口是60秒。
• 那么心跳间隔建议设为20秒（60/3）。
• 这样即使连续丢两个包，第三个包也能在超时前到达。

QoS选择上，我建议用QoS 1。为什么？

• QoS 0：发完即忘，网络抖动时容易丢包，导致误判离线。
• QoS 2：四次握手，开销太大，心跳包没必要这么可靠。
• QoS 1：至少一次送达，性能与可靠性的平衡点。

3.4 代码示例：Python实现心跳发送

说了这么多，来点实际的。下面是一个简单的Python心跳发送示例：

import json
import time
import random
from paho.mqtt import client as mqtt_client

class HeartbeatSender:
    def __init__(self, broker, port, product_key, device_name):
        self.broker = broker
        self.port = port
        self.product_key = product_key
        self.device_name = device_name
        self.client = mqtt_client.Client(client_id=device_name)
        self.client.on_connect = self.on_connect
        
    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print("连接成功")
        else:
            print(f"连接失败，返回码：{rc}")
            
    def build_heartbeat(self):
        """构建心跳包Payload"""
        return {
            "id": f"hb_{int(time.time()*1000)}_{random.randint(1000,9999)}",
            "type": "heartbeat",
            "timestamp": int(time.time() * 1000),
            "status": {
                "battery": random.randint(60, 100),
                "rssi": random.randint(-80, -40),
                "uptime": int(time.time())
            },
            "version": "1.0.0"
        }
        
    def send_heartbeat(self):
        """发送心跳包"""
        topic = f"v1/{self.product_key}/{self.device_name}/heartbeat"
        payload = json.dumps(self.build_heartbeat())
        
        # 使用QoS 1发送
        result = self.client.publish(topic, payload, qos=1)
        
        if result.rc == mqtt_client.MQTT_ERR_SUCCESS:
            print(f"心跳发送成功：{payload}")
        else:
            print(f"心跳发送失败，错误码：{result.rc}")
            
    def start(self, interval=20):
        """启动心跳循环"""
        self.client.connect(self.broker, self.port)
        self.client.loop_start()
        
        while True:
            self.send_heartbeat()
            time.sleep(interval)

# 使用示例
sender = HeartbeatSender(
    broker="mqtt.example.com",
    port=1883,
    product_key="pk_123456",
    device_name="device_001"
)
sender.start(interval=20)

这段代码里，我特意加了个随机数到消息ID里，就是为了防止重复。你想想看，如果只用时间戳，同一毫秒内发两个包，ID就撞了。加个随机数，基本不会重复。

3.5 服务端校验逻辑

心跳包发出去，服务端怎么处理？简单说三步：

1. 解析Topic：提取productKey和deviceName。
2. 校验Payload：检查type是否为heartbeat，timestamp是否合法。
3. 更新状态：将设备最后心跳时间写入缓存（Redis或内存）。

这里有个细节：服务端不要直接拿当前时间减去心跳时间来判断离线。因为网络延迟会导致时间偏差。我建议用心跳到达时间作为基准，而不是心跳包里的timestamp。

下一节，咱们聊聊服务端如何高效处理海量设备的心跳数据。嗯，那才是真正的挑战。