C 上的流式数据处理：架构与示例

C 语言中的流数据处理：架构与实现

流处理要求最小延迟、可预测的执行时间，以及对配置变更的灵活性。与批处理不同，批处理中数据会在分析前积累，导致固有延迟，而流系统会在数据到达时立即响应。这使得它们对于视频分析、网络检查和其他实时应用至关重要。C 语言非常适合这些系统，因为它没有垃圾回收机制，并提供对内存和执行时序的完全控制。

流处理的架构抽象

将流系统建模为计算图很方便，其中节点是数据处理器，边表示数据流方向。每个节点都有严格的类型：

• INPUT_NODE — 数据源（网络、磁盘、传感器）；
• PROCESSING_NODE — 转换器（分析、过滤、编码）；
• OUTPUT_NODE — 消费者（存储、传输、日志记录）。

图的拓扑结构决定了系统的整体行为。最简单的是用于线性管道的单链表。更复杂的任务需要树结构（按数据类型分支）或任意图（循环依赖、连接状态）。

为任务选择合适的拓扑结构

选择合适的图结构直接影响性能和可扩展性：

• 单链表 — 适用于同构输入数据和线性处理。示例：JPEG 解码 → 过滤 → 保存为 BMP。
• 树 — 用于基于输入类型的路由。例如，将 H.264 或 HEVC 视频流定向到不同的解码器。
• 图 — 对于有状态操作或反馈循环是必需的。TCP 流量处理需要会话上下文存储和节点重新遍历。

C 语言实现

系统围绕动态加载模块（Linux 上的 .so）构建。每个插件导出一个 node_t 结构，定义其行为和连接。核心基础设施在 infra.h 头文件中：

typedef enum NODE_TYPE_C {
    INPUT_NODE,
    OUTPUT_NODE,
    PROCESSING_NODE
} NODE_TYPE_T;

typedef struct matrix_s {
  unsigned int width_;
  unsigned int height_;
  unsigned char *data_;
} matrix_t;

typedef struct data_s {
  matrix_t matrix_;
  char *metadata_;
} data_t;

typedef void (*init_function_t)(void);
typedef unsigned short (*processing_function_t)(data_t**, unsigned short);

typedef struct node_s {
  char *name_;
  NODE_TYPE_T type_;
  char *prev_;
  char *next_;
  init_function_t init;
  processing_function_t processing;
} node_t;

typedef node_t* (*get_node_structure)(void);

#define REGISTER_NODE(plugin_name, type, processing_function, init_function, prev, next) \
    static node_t node = {.name_ = plugin_name, .type_ = type, \
                          .init = init_function, .processing = processing_function, \
                          .prev_ = prev, .next_ = next}; \
    node_t* getnode_structure() { return &node; }

每个插件实现两个函数：init() 用于资源设置，processing() 用于核心逻辑。REGISTER_NODE 宏自动生成导出。

示例：视频分析管道

考虑一个合成场景：通过网络接收 JPEG 帧，交换 R 和 G 通道，然后保存为 BMP 和 CSV。

输入插件（input_plugin.c） 每次迭代抓取最多五个帧，解码它们并传递下去：

static unsigned short processing(data_t **data, unsigned short count) {
  if(data == NULL) return 0;
  unsigned short i = 0;
  for (; i < 5; ++i) {
    unsigned char *jpeg_image = get_image();
    if (jpeg_image == NULL) break;
    decode_image(data[i], jpeg_image);
    free(jpeg_image);
  }
  return i;
}
REGISTER_NODE("input-node", INPUT_NODE, processing, init, NULL, "processing-node");

处理插件（processing_plugin.c） 交换像素通道：

static unsigned short processing(data_t **data, unsigned short count) {
  for (unsigned short i = 0; i < count; ++i) {
    unsigned int sz = data[i].width_ * data[i].height_;
    for (unsigned int j = 0; j < sz; j += 3) {
      unsigned char pix = data[i].data_[j];
      data[i].data_[j] = data[i].data_[j + 1];
      data[i].data_[j + 1] = pix;
    }
  }
  return count;
}
REGISTER_NODE("processing-node", PROCESSING_NODE, processing, init, "input-node", "output-node");

输出插件（output_plugin.c） 保存结果并释放内存：

static unsigned short processing(data_t **data, unsigned short count) {
  for (unsigned short i = 0; i < count; ++i) {
    save_data(data[i]);
    free(data[i].data_);
    free(data[i].metadata_);
  }
  free(data);
  return count;
}
REGISTER_NODE("output-node", OUTPUT_NODE, processing, init, "processing-node", NULL);

main.c 中的主循环加载插件，设置执行顺序，并在无限循环中运行处理以最小化空闲时间。

关键要点

• 基于 C 的流系统避免了 GC 语言中常见的不可预测暂停。
• 在设计时选择图拓扑以适应特定用例。
• 动态插件加载无需核心重新编译即可实现扩展性。
• 每个节点仅处理一种角色：输入、处理或输出。
• 内存管理完全由开发者控制——要求纪律性，但提供终极精确性。

— Editorial Team