断电、断网、Broker 掉线：用 iotspool 给 MQTT 设备加一层掉电安全消息队列

声明： iotspool 是一个开源项目，作者为 Vanderhell。本文是阅读该项目源码和文档后整理的学习笔记，用于理解嵌入式 MQTT 场景下持久化 store-and-forward 队列的工程实现方式。本文作者不是该项目的开发者，未参与该项目的任何代码贡献。 文中所有工程细节均来自对开源代码的分析，不代表本文作者的设计决策。

项目仓库：github.com/Vanderhell/iotspool（MQTT publish 的持久化 store-and-forward 队列，掉电不丢消息）

做物联网设备的人迟早要踩同一个坑：数据到了发送这一刻，网络断了、Broker 重启了、或者设备自己掉电了。直接 mqtt_publish() 就赌它成功，赌输了就是数据默默消失；自己在 RAM 里攒 FIFO，一掉电全没；往 Flash 里随便写，写一半断电文件就坏。

iotspool 专门填这个坑：一个零依赖的 C99 库，把 MQTT PUBLISH 消息写进 append-only 日志，CRC 校验，掉电自动截断尾部半截记录，Broker 不可达时指数退避重发，成功后才把 ACK 落到磁盘。支持 Linux、ESP-IDF、STM32 等裸机环境，coreMQTT/Paho/mosquitto/自研 MQTT 客户端都能对接。

iotspool 位置与数据流

它解决什么问题

嵌入式 MQTT 上报的常见故障模式：

瞬时断网：路由器重启、基站切换，几秒到几十秒恢复。RAM FIFO 能顶一阵，但满了之后丢谁、留谁要自己定策略。
Broker 不可达：云端升级、证书轮换，可能持续几分钟到几小时。RAM 顶不住，必须落盘。
设备意外重启：看门狗复位、掉电、HardFault。RAM 里未发送的消息全部丢失。
写入中途断电：消息写了一半电没了，下次启动不能读到半截脏数据。

iotspool 给每一条待发消息分配一个单调递增的 msg_id，把 ENQ（入队）记录追加到存储文件，写完后 fsync（或等价操作）；发送成功后再追加一条 ACK 记录标记该 msg_id 已确认。下次启动时 iotspool_recover() 扫一遍日志，尾部 CRC 不完整的半截记录直接丢弃，其余未被 ACK 的 ENQ 重新进入待发队列。

五分钟在 Linux 上跑通 Demo

下面是最小可运行示例，逻辑对应仓库里的 examples/rpi/main.c，做了精简。

1. 编译

git clone https://github.com/Vanderhell/iotspool.git
cd iotspool
cmake -S . -B build -DIOTSPOOL_BUILD_TESTS=ON
cmake --build build -j

仓库自带 POSIX 存储后端 src/store_posix.c，Linux/macOS/ESP-IDF VFS 都能直接用。STM32 等裸机环境需要自己实现 iotspool_store_t 的 5 个回调（append / read_at / sync / size_bytes / truncate_to / replace），对接到 raw Flash 或 LittleFS。

2. 最小发送循环

#include "iotspool.h"
#include "../../src/store_posix.h"
#include <stdio.h>
#include <string.h>
#include <time.h>

static uint32_t now_ms(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (uint32_t)(ts.tv_sec * 1000 + ts.tv_nsec / 1000000);
}

/* 替换成你自己的 MQTT 客户端发布函数，返回 0 表示成功 */
static int fake_mqtt_publish(const char *topic, uint32_t topic_len,
                             const uint8_t *payload, uint32_t payload_len,
                             uint8_t qos) {
    printf("  [MQTT] %.*s -> %.*s (qos=%d)\n",
           (int)topic_len, topic,
           (int)payload_len, (const char *)payload, qos);
    return 0;
}

int main(void) {
    /* 1. 打开存储文件 */
    iotspool_store_t store = {0};
    if (store_posix_open("/tmp/spool.bin", &store) != IOTSPOOL_OK) {
        fprintf(stderr, "open store failed\n");
        return 1;
    }

    /* 2. 初始化 + 从磁盘恢复未发送队列 */
    iotspool_cfg_t cfg = iotspool_cfg_default();
    cfg.max_store_bytes    = 512 * 1024;  /* 最多占 512KB */
    cfg.drop_oldest_on_full = true;      /* 满了丢最老的，保新数据 */
    cfg.min_retry_ms = 1000;
    cfg.max_retry_ms = 60000;

    iotspool_t *spool = NULL;
    iotspool_init(&spool, &cfg, &store);
    iotspool_recover(spool);   /* 空文件上调用也是安全的 */

    /* 3. 业务侧入队：传感器采样后直接 enqueue，不等网络 */
    const char *payload = "{\"temp\":23.7,\"humi\":58}";
    iotspool_msg_t m = {
        .topic       = "factory/line1/env",
        .topic_len   = (uint32_t)strlen("factory/line1/env"),
        .payload     = (const uint8_t *)payload,
        .payload_len = (uint32_t)strlen(payload),
        .qos         = 1,
        .retain      = false,
    };
    iotspool_msg_id_t id;
    iotspool_enqueue(spool, &m, &id);

    /* 4. 网络任务里的发送循环 */
    iotspool_msg_t out = {0};
    iotspool_msg_id_t out_id = IOTSPOOL_MSG_ID_INVALID;

    if (iotspool_peek_ready(spool, now_ms(), &out, &out_id) == IOTSPOOL_OK) {
        if (fake_mqtt_publish(out.topic, out.topic_len,
                              out.payload, out.payload_len, out.qos) == 0) {
            iotspool_ack(spool, out_id);          /* QoS0 在发送成功后调 */
        } else {
            iotspool_on_publish_fail(spool, now_ms());  /* 触发退避 */
        }
    }

    iotspool_stats_t st;
    iotspool_stats(spool, &st);
    printf("pending=%u acked=%u store=%u bytes\n",
           st.pending_count, st.acked_total, st.store_bytes);

    iotspool_deinit(spool);
    store_posix_close(&store);
    return 0;
}

运行仓库自带的完整示例：

./build/example_rpi

输出会显示 5 条传感器读数入队、随后被逐个取出、ACK 后统计归零。

3. 对接真实 MQTT 客户端的关键约定

QoS	`iotspool_ack()` 调用时机
0	传输层确认报文已发出去之后（例如 Paho `MQTTClient_publishMessage` 返回成功）
1	收到 Broker 的 `PUBACK` 之后

QoS 2 当前不在库的支持范围内。库保证 QoS 1 跨重启的 at-least-once 语义，重复送达由上层去重。

配置项速查

iotspool_cfg_default() 返回默认值，按需改：

max_pending_msgs：RAM 里索引的最大消息条数（默认 64）。超过这个值再 enqueue 会返回 IOTSPOOL_EFULL，除非开了 drop_oldest_on_full。
max_store_bytes：存储文件最大字节数（默认 256KB）。
max_payload_bytes / max_topic_bytes：单条消息 payload/topic 的上限，用于内部分配 scratch。
min_retry_ms / max_retry_ms：退避区间，默认 1s 到 60s。
drop_oldest_on_full：队列满时是否丢最老消息。设为 false 时 enqueue 直接返回 IOTSPOOL_EFULL，由调用方决定丢还是等。
enable_sha256：是否给每条记录加 SHA-256 校验（默认关，CRC32 已够检测掉电半截）。
lock / unlock：多任务环境下传入互斥锁回调，核心 API 执行前自动加锁。ISR 不可直接调库，除非锁回调能在 ISR 中返回 IOTSPOOL_EBUSY。

掉电恢复的启动顺序

推荐的开机流程：

挂载文件系统（或初始化 Flash 驱动）。
打开 store 后端（store_posix_open 或自定义 vtable）。
iotspool_init / iotspool_init_inplace。
iotspool_recover。
启动传感器采样任务和 MQTT 发送任务。

这一步必须放在 MQTT 连接之前，否则连接一建立、publish 任务开始跑，但队列状态还没重建，可能漏发重启前积压的消息。

嵌入式无堆环境的用法

默认 iotspool_init 会 malloc 内部索引和 scratch 空间。STM32 等不想用堆的场合，用静态内存版本：

iotspool_t spool;
iotspool_entry_t entries[64];                     /* RAM 索引 */
uint8_t scratch[iotspool_required_scratch_bytes(&cfg)];

iotspool_init_inplace(&spool, entries, 64,
                      scratch, sizeof(scratch),
                      &cfg, &store);

大小可以用 iotspool_required_index_bytes(&cfg) 和 iotspool_required_scratch_bytes(&cfg) 在编译期或启动时精确计算，避免多分配。

适用边界

iotspool 只负责"把 publish 消息可靠地送到 MQTT 客户端手里"，不做以下事情：

不做 MQTT 协议本身。连接、保活、重连、TLS、订阅都由你现有的 MQTT 栈负责。
不做加密。CRC32/SHA-256 只检测意外损坏，不防篡改；需要机密性请用文件系统级加密或在 payload 层加密。
不做 QoS 2。只保证 QoS 1 at-least-once，重复消息去重在 Broker 或业务层。
不跨设备同步。日志文件是单设备本地的。