13. 多线程事件驱动：线程池事件处理、工作窃取模式、线程安全事件分发

多线程事件驱动，说白了就是让一堆工人（线程）同时干活，谁闲了就抢活干。我最早接触这个概念是在做一个网络网关项目，单线程处理几百个连接，CPU 占用率才 30%，但响应延迟已经高得离谱了。后来改成线程池 + 事件驱动，吞吐量直接翻了三倍。

嗯，这一章我们重点聊三个东西：线程池怎么管、工作窃取怎么玩、事件分发怎么保证线程安全。

13.1 线程池事件处理模型

先看最基础的线程池模型。事件来了，丢进一个共享队列，线程们从队列里取事件处理。

我习惯用固定大小的线程池。为什么？因为动态调整线程数其实很坑。我在一个项目中试过根据负载自动增减线程，结果线程数忽高忽低，系统反而更不稳定。后来就老实了，固定线程数 + 任务队列背压控制，稳得很。

来看一个简单的线程池事件处理框架：

// 事件结构体
typedef struct {
    int event_type;
    void *data;
    void (*handler)(void *data);
} event_t;

// 线程池结构
typedef struct {
    pthread_t *threads;
    int thread_count;
    event_queue_t *queue;  // 线程安全队列
    volatile int running;
} thread_pool_t;

// 工作线程函数
void *worker_thread(void *arg) {
    thread_pool_t *pool = (thread_pool_t *)arg;
    
    while (pool->running) {
        event_t evt;
        // 从队列取事件，阻塞等待
        if (event_queue_pop(pool->queue, &evt, 1000) == 0) {
            // 执行事件处理
            evt.handler(evt.data);
        }
    }
    return NULL;
}

这里有个关键点：event_queue_pop 必须是线程安全的阻塞操作。我一般用条件变量 + 互斥锁实现，或者用信号量。

13.2 工作窃取模式

固定线程池有个问题：每个线程从同一个队列取任务，竞争锁的开销很大。工作窃取模式就是来解决这个问题的。

说白了，每个线程有自己的任务队列。线程优先处理自己队列里的任务。如果自己的队列空了，就去偷别的线程的任务。

我画个图帮你理解：

看到没？线程 3 的队列空了，它就去偷线程 1 队尾的任务。为什么要偷队尾？因为线程 1 是从队头取任务的，这样冲突概率最小。

工作窃取模式的实现要点：

• 每个线程一个双端队列（deque）：线程从头部取任务，窃取者从尾部取任务
• 窃取操作使用 try_lock：如果锁被占用就跳过，不要阻塞
• 窃取失败就重试：轮询其他线程的队列

13.3 线程安全事件分发

事件分发说白了就是：事件来了，该交给哪个线程处理？怎么保证不出乱子？

我见过三种常见的分发策略：

策略	原理	适用场景	我的评价
轮询分发	按顺序分配给各个线程	任务无关联，负载均衡	简单但可能不均衡
哈希分发	按事件 ID 哈希到固定线程	需要保证同一事件源顺序	我项目里最常用
优先级分发	高优先级事件优先分配	实时性要求高的场景	小心优先级反转

哈希分发有个好处：同一个客户端的事件永远落在同一个线程上，这样就不需要加锁了。我在做网络服务器时就是这么干的——每个连接绑定一个线程，所有事件都在那个线程里处理，省去了大量的锁竞争。

来看一个哈希分发的实现：

// 事件分发器
typedef struct {
    thread_pool_t *pool;
    int (*hash_func)(event_t *evt);
} event_dispatcher_t;

// 分发事件到指定线程
void event_dispatcher_dispatch(event_dispatcher_t *disp, event_t *evt) {
    int thread_id = disp->hash_func(evt) % disp->pool->thread_count;
    // 直接 push 到对应线程的私有队列
    thread_queue_push(disp->pool->threads[thread_id].queue, evt);
}

// 哈希函数示例：按事件源 ID 哈希
int hash_by_source_id(event_t *evt) {
    // 假设事件数据中包含 source_id
    int source_id = *(int *)evt->data;
    return source_id;
}

这里要注意：哈希函数要尽量均匀分布。我遇到过一个问题，某个哈希函数把所有事件都映射到了同一个线程上，其他线程闲得发慌。后来改用 source_id % thread_count 配合质数取模，才解决了。

13.4 线程安全队列的实现

不管哪种分发方式，最终都要落到线程安全的队列操作上。我直接给一个无锁队列的简化版：

// 无锁队列节点
typedef struct lf_node {
    event_t event;
    struct lf_node *next;
} lf_node_t;

// 无锁队列
typedef struct {
    lf_node_t *head;
    lf_node_t *tail;
    // 使用 CAS 操作保证原子性
} lockfree_queue_t;

// 入队（简化版）
void lockfree_push(lockfree_queue_t *q, event_t evt) {
    lf_node_t *node = malloc(sizeof(lf_node_t));
    node->event = evt;
    node->next = NULL;
    
    lf_node_t *old_tail;
    do {
        old_tail = q->tail;
    } while (!__sync_bool_compare_and_swap(&q->tail->next, NULL, node));
    
    __sync_bool_compare_and_swap(&q->tail, old_tail, node);
}

13.5 实战中的避坑指南

最后总结几个我实际项目中遇到的坑：

• 事件处理不能阻塞：线程池里的线程如果被阻塞了，整个系统就卡住了。我一般要求事件处理函数必须在 1ms 内返回，超过的用异步方式处理。
• 小心死锁：事件 A 等待事件 B 的结果，事件 B 又在等待事件 A 的锁，这就死锁了。我习惯用超时机制，超过一定时间就报错。
• 线程局部存储：每个线程最好有自己的内存池，避免频繁 malloc/free。我在一个高并发项目里，光是把事件分配改成线程局部内存池，性能就提升了 40%。
• 监控线程池状态：队列长度、线程利用率、事件处理延迟，这些指标一定要能实时看到。不然出了问题你都不知道是哪里卡住了。

嗯，多线程事件驱动其实就这些核心思想。你想想看，无非就是怎么分配任务、怎么减少竞争、怎么保证安全。把这三个问题想清楚了，剩下的就是工程细节了。

公众号:蓝海资料掘金营，微信deep3321