实战项目四：实时数据流处理引擎
各位，终于到了这个系列里最硬核的项目了。
实时数据流处理引擎，说白了就是一条数据从源头进来，经过解析、过滤、转换、聚合，最后输出到目标端。整个过程不能卡顿，不能丢数据，还要能扛住突发流量。我当年在物联网平台做设备数据采集时，就吃过这个亏——消息队列设计得不够健壮，高峰期直接丢包，被运维同事追着骂了一周。
今天咱们就把这个坑填上。
项目需求与架构概览
我们要实现的是一个轻量级的数据流引擎，核心能力如下：
- 支持多路数据源接入（比如串口、网络 socket、文件管道）
- 每个数据源独立的消息队列缓冲
- 可配置的过滤与转换流水线
- 基于事件驱动的调度，不阻塞主循环
- 支持动态启停数据流
先看整体架构图，我习惯在动手写代码前把结构画清楚。
你看，数据从左边进来，先进入各自的消息队列。事件调度器轮询所有队列，一旦有数据就触发对应的处理流水线。流水线里的每个环节都是独立的回调函数，可以动态插拔。这就是事件驱动架构的精髓——不轮询、不阻塞、按需响应。
消息队列的核心设计
消息队列是整个引擎的基石。我见过很多项目直接用链表做队列，结果在频繁入队出队时内存碎片化严重。我的做法是：用环形缓冲区 + 固定大小节点池。
/* 消息队列结构 */
typedef struct msg_queue {
msg_node_t *buffer; /* 环形缓冲区 */
uint32_t capacity; /* 容量 */
uint32_t head; /* 读指针 */
uint32_t tail; /* 写指针 */
uint32_t count; /* 当前消息数 */
/* 事件通知 */
int event_fd; /* eventfd，用于唤醒调度器 */
/* 统计 */
uint32_t peak_count; /* 峰值深度 */
uint32_t overflow; /* 溢出次数 */
} msg_queue_t;这里有个关键点：event_fd。Linux 的 eventfd 是事件驱动的好帮手。当队列从空变为非空时，往 eventfd 写一个字节，调度器的 epoll 立刻就能感知到。反过来，队列空了也不用管，调度器自然就休眠了。
核心原则：消息队列的入队操作必须是非阻塞的，出队操作可以是阻塞的（但只在调度器线程中阻塞）。
事件调度器的实现
调度器我选 epoll，原因很简单——它能处理海量文件描述符，而且边缘触发模式效率极高。你想想看，如果每个数据源都开一个线程去轮询，系统开销得多大？
/* 事件调度器主循环 */
void event_loop_run(event_loop_t *loop) {
struct epoll_event events[MAX_EVENTS];
while (loop->running) {
int nfds = epoll_wait(loop->epfd, events, MAX_EVENTS, -1);
for (int i = 0; i < nfds; i++) {
event_handler_t *handler = (event_handler_t *)events[i].data.ptr;
if (events[i].events & EPOLLIN) {
handler->on_readable(handler->arg);
}
if (events[i].events & EPOLLOUT) {
handler->on_writable(handler->arg);
}
if (events[i].events & (EPOLLERR | EPOLLHUP)) {
handler->on_error(handler->arg);
}
}
}
}这段代码看起来简单，但实际项目中坑不少。我曾经在 epoll_wait 返回后直接遍历所有事件，结果在回调里删除了某个 handler，导致后续访问野指针。后来我加了个延迟删除机制——先把要删除的 handler 标记一下，等下一轮 epoll_wait 之前再统一清理。
注意：在事件回调中不要直接释放 handler 或关闭 fd。应该用 defer 机制，否则 epoll 内部状态会乱掉。
处理流水线的设计
流水线我采用责任链模式。每个处理节点都是一个函数指针，数据依次经过每个节点。这样做的好处是：你可以随时插入新的处理环节，而不需要修改现有代码。
/* 流水线节点 */
typedef struct pipeline_node {
const char *name;
int (*process)(void *data, uint32_t len, void *ctx);
void *ctx;
struct pipeline_node *next;
} pipeline_node_t;
/* 执行流水线 */
int pipeline_execute(pipeline_t *pl, void *data, uint32_t len) {
pipeline_node_t *node = pl->head;
int ret = 0;
while (node) {
ret = node->process(data, len, node->ctx);
if (ret != 0) {
/* 节点返回非0表示处理失败，终止流水线 */
log_warn("pipeline node %s failed: %d", node->name, ret);
return ret;
}
node = node->next;
}
return 0;
}举个例子，一个典型的流水线可能是这样的：
- 解析器：把原始字节流解析成结构体
- 过滤器：丢弃无效或重复的数据
- 转换器：把单位从毫伏转为摄氏度，或者做协议转换
- 聚合器：每10条数据做一次滑动平均
每个节点只做一件事，而且可以单独测试。我习惯给每个节点加一个统计计数器，记录处理了多少条数据、失败了多少条。线上排查问题时，这些统计信息就是救命稻草。
动态启停与资源管理
实时数据流引擎必须支持动态启停。比如某个数据源掉线了，你不能停掉整个引擎，只能停掉那一路流。
我的做法是：每个数据流对应一个 stream_t 结构体，里面包含队列、流水线、状态机。启停操作通过原子变量控制，调度器在每次轮询时检查状态。
typedef enum {
STREAM_STOPPED,
STREAM_RUNNING,
STREAM_PAUSED,
STREAM_ERROR
} stream_state_t;
typedef struct data_stream {
stream_state_t state;
msg_queue_t queue;
pipeline_t pipeline;
event_handler_t handler;
/* ... */
} data_stream_t;
/* 启动一个数据流 */
int stream_start(data_stream_t *stream) {
if (atomic_cas(&stream->state, STREAM_STOPPED, STREAM_RUNNING)) {
/* 注册事件处理器到调度器 */
event_loop_add_handler(loop, &stream->handler);
return 0;
}
return -1; /* 状态不对 */
}这里用了原子比较交换（CAS），保证状态切换的线程安全性。你想想看，如果主线程在启动流，同时另一个线程在停止流，没有原子操作就会出大乱子。
性能优化与避坑
最后聊几个实战中容易踩的坑。
技巧：消息队列的缓冲区大小不要设成 2 的幂次减 1，而要设成 2 的幂次。这样 head 和 tail 的取模运算可以用位运算代替，速度能快一个数量级。
我曾经在一个项目中，消息队列的容量是 1023。结果每次入队出队都要做一次除法取模，CPU 占用率直接飙到 30%。改成 1024 后，用 head & (capacity - 1) 代替取模，占用率降到 3%。
还有一个坑：eventfd 的读写。eventfd 的 read 操作会清空计数器，如果你在多个线程中同时读同一个 eventfd，可能会丢失事件。我的做法是：每个队列独享一个 eventfd，调度器线程是唯一的读者。
嗯，说到这，核心的设计思路基本都覆盖了。实时数据流引擎的关键就三点：非阻塞队列、事件驱动调度、可插拔流水线。把这三点吃透了，你就能应对大部分实时数据处理场景。
代码实现上，建议先从单线程版本开始，跑通了再加多线程优化。别一上来就搞多线程，调试起来你会怀疑人生的。