MQTT Stream 设计与实现

本页说明 MQTT Stream 的内部工作机制，涵盖端到端数据管道、Stream 插件架构、自定义插件开发以及查询通道协议。

本文面向需要理解内部设计或扩展 MQTT Stream 默认行为的开发者和系统集成者。

端到端数据管道

启用 MQTT Stream 后，发布到配置主题的消息会进入独立的数据处理管道。整个过程中，正常 MQTT 转发不受影响。

启动：基于配置初始化

MQTT Stream 数据路径在启动时完全根据配置初始化。系统不会在运行时创建或修改管道。

启动期间：

• 从配置中解析 Exchange 定义、主题过滤器、Stream 插件类型和 Ringbus 参数。
• 为每条配置的数据路径创建一个 Exchange 实例和一个 Ringbus 缓冲区。
• 建立 Broker 与 Exchange 之间的本地通信通道。
• 将每条数据路径绑定到其配置的 Stream 插件。

该模型可保持运行时管道结构稳定，避免资源频繁变动，并确保高吞吐场景下的性能可预测。

消息捕获：Broker 与 Exchange 分离

Broker 和 Exchange 具有严格分离的职责。

Broker 负责 MQTT 协议处理和实时消息转发。当 PUBLISH 消息匹配配置主题时，Broker 会将其旁路复制到 MQTT Stream 管道中，不会修改原始消息或转发路径。

每条旁路复制的消息会在捕获时分配一个单调递增的毫秒级时间戳。该时间戳作为消息在整个管道中的全局 key（用于缓冲、持久化和查询），后续无需再进行排序或去重。

Exchange 接收旁路复制的消息，并仅负责缓冲、批处理、持久化和历史查询。Broker 不感知这些操作。

Ringbus 缓冲和批处理触发

逐条将高频消息写入磁盘会严重限制吞吐量。MQTT Stream 使用 Ringbus 作为内存中的第一级缓冲，用于吸收写入压力。

Ringbus 按时间戳顺序接收消息，直到达到配置容量。它有意不感知编码和存储语义：其唯一职责是快速接收数据，并按批次交给上层处理。

当缓冲区达到容量时，fullOp 配置决定后续行为。推荐模式是 FULL_RETURN（2 - Return to AIO）：

• 整个批次一次性交给上层处理层。
• Ringbus 立即清空并继续接收新消息。
• 所有编码和持久化逻辑都保留在上层，使 Ringbus 保持简单，并便于支持多个后端。

Stream 插件编码和 Parquet 持久化

当 Ringbus 返回一个批次后，管道会执行以下步骤：

1. 将批次规范化为 stream_data_in 结构，其中每条消息包含 payload、长度和时间戳 key。
2. 调用 stream_encode(streamType, ...) 分发到配置的 Stream 插件。
3. 插件将批次编码为可存储的表示形式，例如列式 Parquet 结构。
4. 编码结果会异步写入本地 Parquet 文件。

使用 Parquet 作为存储格式时，面向批次的写入可最大化列式压缩效率，减少文件碎片，并支持快速按时间范围扫描。

查询路径

历史查询通过 Exchange 的本地查询接口处理，不会经过 MQTT 转发路径。一次查询会同时覆盖仍在内存中（Ringbus）的数据和已经持久化到磁盘的数据。Exchange 合并这些来源的数据，并将其交给 Stream 插件解码，然后再向调用方返回结果。

Stream 插件架构

Stream 插件是 MQTT Stream 中数据语义的唯一来源。核心管道不假定固定的存储格式或查询协议；这些能力都委托给插件实现。

每个插件实现三个接口：

Interface	Invocation point	Purpose
encode	Ringbus 满后、写入前	将一批消息编码为可持久化结构
cmd_parser	查询开始时	将查询命令字符串解析为结构化参数
decode	组装查询结果时	将持久化记录解码为最终结果格式

MQTT Stream 包含两个内置插件：

• RAW Stream (streamType = 0)：通用类型。以最少转换传递 MQTT Payload。适用于没有复杂语义要求的高频数据。
• SPI Stream (streamType = 1)：面向汽车和 SDV 场景。支持自定义 key 结构和查询协议。

Exchange 配置中的 streamType 字段决定整条数据路径使用哪个插件，包括编码、持久化、查询解析和结果解码。

编码流程

当 Ringbus 触发 FULL_RETURN，或显式调用 hook_last_flush 刷新剩余数据时，批次会传递给 flush_smsg_to_disk()，该函数会：

1. 根据返回的消息构建 stream_data_in，并使用 nni_msg_get_timestamp() 作为每条记录的 key。
2. 调用 stream_encode(streamType, sdata)，分发到插件的 encode 实现。对于 RAW，raw_encode 会将 stream_data_in 转换为 stream_data_out，并调用 parquet_data_alloc 构建列式结构。
3. 编码结果（parquet_data*）会传递给 parquet_object_alloc + parquet_write_batch_async，用于异步写入磁盘。

解码流程

当查询命令到达 Exchange 时：

1. stream_cmd_parser(streamType, keystr) 将命令字符串解析为 cmd_data 结构，包含 is_sync、start_key、end_key 和 schema。对于 RAW Stream，默认 schema 为 {"ts", "data"}（参见 raw_stream.c::parse_input_cmd）。
2. Exchange 通过 exchange_client_get_msgs_fuzz 从 Ringbus 内存中检索匹配记录，并通过 parquet_get_data_packets_in_range_by_column 从 Parquet 文件中检索匹配记录。
3. 对每个 parquet_data_ret*，调用 stream_decode(streamType, ...) 将列式数据转换为 stream_decoded_data，即适合调用方使用的连续缓冲区。
4. 结果会组装为 NNG 消息，并以增量方式（async）或聚合方式（sync）返回。

自定义 Stream 插件开发

自定义 Stream 插件无需修改核心管道即可实现。插件需要实现三个函数，并在启动时注册。

接口结构体

/* Input to encode: one batch of messages from Ringbus */
struct stream_data_in {
    void     **datas;  /* payload pointer for each message */
    uint64_t  *keys;   /* timestamp key for each message */
    uint32_t  *lens;   /* payload length for each message */
    uint32_t   len;    /* number of messages in the batch */
};

/* Output of encode: columnar structure for Parquet persistence */
struct stream_data_out {
    uint32_t                col_len;      /* number of columns */
    uint32_t                row_len;      /* number of rows */
    uint64_t               *ts;           /* timestamp column */
    char                  **schema;       /* column names */
    parquet_data_packet ***payload_arr;   /* per-column payload */
};

/* Output of decode: continuous buffer returned to the caller */
struct stream_decoded_data {
    void     *data;  /* decoded buffer */
    uint32_t  len;   /* buffer length */
};

/* Output of cmd_parser: structured query parameters */
struct cmd_data {
    bool      is_sync;     /* true = sync, false = async */
    uint64_t  start_key;   /* query start timestamp (ms) */
    uint64_t  end_key;     /* query end timestamp (ms) */
    uint32_t  schema_len;  /* number of requested columns */
    char    **schema;      /* column names to return */
};

插件实现和注册

实现三个接口函数，并在启动时使用 stream_register 注册：

void *my_encode(void *data);      /* data: stream_data_in*   -> return: parquet_data* */
void *my_decode(void *data);      /* data: parquet_data_ret* -> return: stream_decoded_data* */
void *my_cmd_parser(void *data);  /* data: const char*       -> return: cmd_data* */

int my_stream_init() {
    int   ret  = 0;
    char *name = malloc(strlen("my_stream") + 1);
    if (name == NULL) {
        return -1;
    }
    strcpy(name, "my_stream");
    /* Use an ID that does not conflict with built-ins: 0 = RAW, 0x1 = SPI */
    ret = stream_register(name, 0x2, my_decode, my_encode, my_cmd_parser);
    if (ret != 0) {
        free(name);
        return -1;
    }
    return 0;
}

注册后，在对应 Exchange 配置中设置 streamType = 2，即可为该数据路径启用该插件。

各接口调用位置

• encode：在 webhook_post.c::send_exchange_cb / hook_last_flush 中调用，调用时机是 RB_FULL_RETURN 触发后，或显式调用 hook_last_flush 刷新剩余数据时。它接收一个 stream_data_in 批次，并必须返回可用于 Parquet 持久化的编码结构。
• cmd_parser：在 exchange_server.c::query_cb 中调用，调用时机是 NNG pair0 socket 收到查询命令时。它接收原始命令字符串，并必须返回 cmd_data。
• decode：在 query_send_sync / query_send_async 中针对每个从 Parquet 检索到的 parquet_data_ret* 调用。它必须返回带有连续结果缓冲区的 stream_decoded_data。

查询通道和协议

EMQX Edge 暴露一个本地 NNG 通道，用于查询 Exchange 数据。源码仓库中包含一个参考客户端：nng/demo/exchange_consumer/exchange_consumer.c。

该客户端会：

• 通过 nng_pair0 连接到 Exchange Server，地址由 exchange_url 指定，默认值为 tcp://127.0.0.1:10000
• 将单个命令字符串作为请求 payload 发送
• 持续读取响应帧，直到收到 2 字节 EOF 标记（0x0B 0xAD，参见 exchange_server.c::query_send_eof）

EOF 之前的每一帧都是一批解码后的查询结果。

RAW Stream 数据布局

使用 RAW Stream（streamType = 0）时，数据以 key/value 形式存储：

• key：64 位单调递增的毫秒级时间戳，在捕获时通过 nng_msg_set_timestamp 设置
• value：原始 MQTT Payload，存储在 Parquet 文件的 data 列中

查询命令格式

RAW Stream 使用以下命令格式：

sync-<start_key>-<end_key>
async-<start_key>-<end_key>

• sync：在连接关闭前返回 key 范围内的所有结果
• async：Exchange 拆分并调度结果增量返回，适合大时间范围查询
• start_key / end_key：毫秒级时间戳；end_key 必须大于等于 start_key

示例：查询 5 秒窗口内的记录：

$ ./demo/exchange_consumer/exchange_consumer "sync-1700000000000-1700000005000"
Received 1234 bytes
Received 5678 bytes
...

raw_cmd_parser 会将其解析为 is_sync = true、start_key = 1700000000000、end_key = 1700000005000。Exchange 会从 Parquet 和 Ringbus 中读取所有匹配记录，解码后返回响应帧，直到 EOF。

要检索所有持久化数据：

$ ./demo/exchange_consumer/exchange_consumer "sync-0-9223372036854775807"

自定义插件可以通过实现自己的 cmd_parser 来扩展该机制，以支持复合条件、自定义 schema 和分页。