嵌入式MQTT开发中，通信稳定性往往不取决于协议本身，而取决于数据流如何在不同线程间流动。本文从数据流视角，梳理MQTT语音导航模块的设计思路。

---

一、整体数据流

云端MQTT Broker
↓
设备MQTT客户端（tcpip线程回调）
↓
消息队列
↓
工作线程
↓
地图引擎接口
↓
JSON构造
↓
tcpip_callback
↓
MQTT发布
↓
云端

text

整个链路是单向异步的：下行指令从云端到达设备，在工作线程中完成业务处理后，上行结果再通过MQTT发布出去。

核心模块有三个：

• MQTT回调（中断上下文）
• 消息队列（缓冲层）
• 工作线程（执行层）

---

二、为什么不直接在MQTT回调里执行业务逻辑？

MQTT回调运行在lwIP的tcpip线程中，该线程负责整个网络协议栈的运行。如果在这里调用地图引擎的HTTP请求（如map_search_poi），tcpip线程会被阻塞，MQTT心跳包发不出去，服务器会在1.5 × keepalive超时后断开连接。——这是早期版本连接频繁掉线的根本原因。

解决方案：把回调拆成「接收」和「处理」两步——回调只做数据接收和任务封装，然后投递到消息队列，立刻返回释放tcpip线程。

---

三、消息队列与工作线程

使用RT-Thread的消息队列rt_mq_t作为缓冲层：

```c
static rt_mq_t work_mq = RT_NULL;
#define WORK_MQ_DEPTH 8
#define WORK_MQ_SIZE (sizeof(navigation_task_t) * 8)
投递时，MQTT回调把JSON解析后的参数打包成navigation_task_t结构体，用rt_mq_send非阻塞投递；如果队列满了（8个任务），rt_mq_send会立即返回错误，不会阻塞tcpip线程。

工作线程work_thd的优先级设为10，栈大小32KB。它用rt_mq_recv阻塞等待，队列空时线程挂起，不消耗CPU：

c
static void work_thread_entry(void *parameter) {
while (1) {
navigation_task_t task = {0};
if (rt_mq_recv(work_mq, &task, sizeof(task), RT_WAITING_FOREVER) != RT_EOK) {
continue;
}
// 执行业务逻辑：搜索POI、规划路线、构造JSON
// 最终调用 tcpip_callback 发送结果
}
}

四、保活线程与自动恢复

另一个独立线程mqtt_ka每5秒检查连接状态。它的职责是：

WiFi断开时重置订阅状态

非地图主题时断开MQTT

连接丢失时自动重连

订阅失败时重试

它不负责业务数据，只负责维护MQTT连接的长期稳定性。

五、数据流的完整闭环

云端下发search指令后，经过以下步骤：

tcpip线程收到JSON，在mqtt_incoming_data_cb中解析action_id为search，提取keyword和location。

构造navigation_task_t，type = TASK_TYPE_SEARCH，复制会话元数据，rt_mq_send投递到工作队列。

work_thd从队列取出任务，调用map_search_poi执行同步HTTP请求（可能耗时数秒）。

获取POI列表后，构造interaction.result格式的JSON，调用tcpip_callback((tcpip_callback_fn)mqtt_publish_callback, response_str)，把发布任务交还给tcpip线程。

tcpip线程执行mqtt_publish，QoS 1保证消息至少送达一次，回调释放JSON内存。

云端收到poi_list，用户语音选择POI后下发select_poi，重复上述流程，最终到达select_route启动导航。

六、设计要点总结

组件 职责 关键约束
MQTT回调（tcpip线程） 接收、解析、投递任务 不能阻塞，不能执行耗时操作
消息队列 缓冲任务 容量8，满则丢弃新任务（可接受）
工作线程（work_thd） 执行业务逻辑 可阻塞，32KB栈，优先级10
保活线程（mqtt_ka） 维护连接状态 优先级15，4KB栈
核心设计原则
耗时操作离开网络线程。

地图引擎的HTTP请求、JSON序列化、文件I/O都放在工作线程，tcpip线程只做MQTT收发和协议栈处理。

目前该模块已稳定运行，主流程（搜索 → 选POI → 选路线 → 导航）能够完整跑通。