让多台机器狗跳同一支舞、做同步表演，靠的就是 MQTT 通信。本章讲清群组表演的通信原理与时间同步机制，并给出一段可直接接入的自定义 MQTT 示例代码，方便你扩展多机协作玩法。
本章包含：
小节 | 内容 | 难度 |
|---|---|---|
5.1 群组表演原理 | 主题协议、消息格式、NTP 时间同步与 UDP 回退 | ★★☆ |
5.2 自定义 MQTT 开发 | 用 paho-mqtt 接入同一套主题协议（完整示例代码） | ★★★ |
🧭 阅读路径：只想理解多机同步怎么做到的，看 5.1；要自己写代码接入，照 5.2 的示例改即可。
前置条件：
完成第一章，已能 SSH 登录机器狗；
至少两台机器狗（单机也可，只是看不到同步效果）；
机器狗能访问外网（群组表演用公共 MQTT Broker 与 NTP 服务器）。
5.1 群组表演原理
为什么用 MQTT 而不是点对点 / 单机同步：多机同步要解决「同一时刻做同一动作」。用中心逐个发指令延迟不可控，用局域网广播又无法跨网段。MQTT 走公共 Broker（云中转），每台狗都订阅同一个房间号下的消息，收到带绝对时间戳的指令后按各自时钟到点执行——这样不同网络的狗也能同步。谁是发言方不重要，大家都在协同同一份时间表。
群组表演（apps/group_perform/main.py）基于公共 MQTT Broker 实现多机同步：
┌─────────────┐
│ MQTT Broker │ broker.emqx.io:1883 (MQTTv311, keepalive=60)
└──┬──┬──┬──┬──┘
│ │ │ │
▼ ▼ ▼ ▼
狗1 狗2 狗3 狗N
每台设备以自身的外部公网 IP 作为房间号 room_id（get_external_ip()），同一房间内的设备在三个主题上互通：
主题 | 消息 type | 说明 |
|---|---|---|
|
| 设备上下线与心跳（并设置遗嘱 will） |
|
| 控制指令：停止表演 / 退出房间 |
|
| 动作计划（带绝对时间戳） |
消息格式（JSON）：
// presence
{"type": "join", "ip": "192.168.1.100", "dog_type": "xgomini"}
{"type": "heartbeat", "ip": "192.168.1.100", "dog_type": "xgomini"}
// command
{"type": "stop"}
{"type": "exit"}
// plan
{
"type": "start",
"actions": [
{"id": 13, "name": "招手", "start_at": 1710000000.500, "duration": 7},
{"id": 16, "name": "摇摆", "start_at": 1710000008.000, "duration": 6}
],
"music_start": 1710000000.500,
"dog_type": "xgomini"
}
时间同步（多机同步执行的关键）：
各设备通过 SNTP（UDP 123）向
ntp.aliyun.com/ntp1.aliyun.com/cn.pool.ntp.org/pool.ntp.org请求时钟偏移，维护统一时间synced_time() = time.time() + offset（成功每 60s 刷新，失败 5s 重试）发起方以
synced_time() + ACTION_PREP_DELAY(3.0s)为基准，为每个动作计算绝对开始时间start_at并下发；订阅方按绝对时间等待执行（与本地时钟偏差 >30s 视为时钟未同步，拒绝执行）无外网时自动回退 UDP 广播模式（端口 5005，以本地 IP 前 3 段作为房间号），消息格式与 MQTT 相同
通信细节：client_id = f"xgo-{local_ip}-{time.time()%100000}"；will_set(presence, {"type":"leave","ip":local_ip}) 保证设备掉线时通知其他设备；断线自动重连。
一次同步表演的时间线（记住它，写代码才有全局观）：
各狗开机后每 60s 从 NTP 同步一次时钟（失败 5s 重试），维护
offset；发起方选好动作、算好每个动作的绝对开始时间
start_at，向plan主题发start；每台订阅狗校验自己与发起方时钟偏差 <30s，然后本地到点执行，动作之间靠时间戳对齐；
整个过程中各狗周期发
heartbeat上报在线，掉线由will遗嘱广播leave。
怎么确认同步生效：让每台狗打印自己的 synced_time() - 本地时间（即 offset），应稳定在几毫秒内；若偏差 >30s 会被判定「时钟未同步」而拒绝执行 start，这时先检查 NTP 能否连通 ntp.aliyun.com、是否被防火墙拦截。可在 group_perform/main.py 里临时加 print(offset) 观察。
5.2 自定义 MQTT 开发
参考 group_perform/main.py 的写法，使用 paho-mqtt 接入同一套主题协议：
import paho.mqtt.client as mqtt
import json
# paho-mqtt 2.x 需传 callback_api_version；1.x 可直接 mqtt.Client(client_id=..., protocol=...)（源码用 try/except 兼容两者）
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id="xgo-dev-1", protocol=mqtt.MQTTv311)
client.connect("broker.emqx.io", 1883, 60)
room_id = "ROOM_ID" # 群组表演中以外部公网 IP 作为房间号
base = f"xgo/group/{room_id}"
# 1) 发布 presence 心跳
client.publish(f"{base}/presence", json.dumps({
"type": "join",
"ip": "192.168.1.100",
"dog_type": "xgomini",
}))
# 2) 发布动作计划（start_at/music_start 为 Unix 秒，须先做 NTP 时钟同步）
client.publish(f"{base}/plan", json.dumps({
"type": "start",
"actions": [
{"id": 13, "name": "招手", "start_at": 1710000000.5, "duration": 7},
],
"music_start": 1710000000.5,
"dog_type": "xgomini",
}))
# 3) 发送控制指令
client.publish(f"{base}/command", json.dumps({"type": "stop"}))
# 4) 订阅三个主题
client.subscribe(f"{base}/presence")
client.subscribe(f"{base}/command")
client.subscribe(f"{base}/plan")
# 5) 处理消息
client.on_message = lambda c, u, msg: print(f"{msg.topic}: {msg.payload.decode()}")
client.loop_forever()
💡 需先
pip3 install --break-system-packages paho-mqtt>=2.0（新镜像已预装）。
5.2.1 先跑一个最小验证
写完上面的订阅端，先单独打印消息而不去驱动动作，验证链路是否通畅：
python3 my_listener.py # 只订阅并打印三个主题
✅ 预期：先用另一台设备（或 MQTTX 客户端）往某主题发一条消息，订阅端能实时打印
主题: 消息体，说明 MQTT 链路正常。⚠️ 常见问题： - 连接失败 / 报
ssl错误 → 先ping broker.emqx.io，确认能访问外网； - 收到消息但本狗不动 →start_at用了本地时间而非synced_time()，或动作 id 不在本机型动作表里； - 多狗不同步 → 各狗时钟偏差 >30s（检查 NTP），或room_id不相等（外部 IP 不同导致不在同一房间）； - 想离线 / 内网测试 → 群组表演会自动回退 UDP 广播模式（端口 5005，以本地 IP 前 3 段为房间号）。