63、消息队列设计：生产者-消费者模型、线程安全队列、无锁队列

消息队列这东西，说白了就是让不同线程之间能「递纸条」。

我刚开始做网络编程那会儿，总觉得线程间通信不就是读写个全局变量嘛，加个锁不就完了？结果呢，锁多了死锁，锁少了数据乱套。后来才明白，消息队列才是正经的解决方案。

生产者-消费者模型：最经典的协作模式

这个模型其实不复杂。一个线程负责生产数据，另一个线程负责消费数据。中间用个队列缓冲一下。

为什么要缓冲？你想想看，生产者可能一下子来很多数据，消费者处理不过来。没有队列的话，要么丢数据，要么生产者等着——这俩都不好。

我在项目中遇到过一种情况：生产者是网卡收包线程，消费者是业务处理线程。收包速度忽高忽低，业务处理又比较耗时。没有队列的话，收包线程就得等业务线程处理完才能收下一个包——那网络延迟就上去了。

线程安全队列：加锁是基本功

最简单的线程安全队列，就是给普通队列加个互斥锁。伪代码大概这样：

// 线程安全队列 - 互斥锁版本
typedef struct {
    void** buffer;
    int capacity;
    int head;
    int tail;
    int count;
    pthread_mutex_t mutex;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;
} safe_queue_t;

// 入队
int safe_queue_push(safe_queue_t* q, void* data) {
    pthread_mutex_lock(&q->mutex);
    
    while (q->count == q->capacity) {
        // 队列满了，等待消费者取走数据
        pthread_cond_wait(&q->not_full, &q->mutex);
    }
    
    q->buffer[q->tail] = data;
    q->tail = (q->tail + 1) % q->capacity;
    q->count++;
    
    // 通知消费者有数据了
    pthread_cond_signal(&q->not_empty);
    
    pthread_mutex_unlock(&q->mutex);
    return 0;
}

// 出队
void* safe_queue_pop(safe_queue_t* q) {
    pthread_mutex_lock(&q->mutex);
    
    while (q->count == 0) {
        // 队列空了，等待生产者放入数据
        pthread_cond_wait(&q->not_empty, &q->mutex);
    }
    
    void* data = q->buffer[q->head];
    q->head = (q->head + 1) % q->capacity;
    q->count--;
    
    // 通知生产者队列有空位了
    pthread_cond_signal(&q->not_full);
    
    pthread_mutex_unlock(&q->mutex);
    return data;
}

嗯，这里要注意：pthread_cond_wait 一定要在循环里调用，不能只用 if。为什么？因为可能有「虚假唤醒」——线程被唤醒了，但条件其实还没满足。我刚开始写的时候就被这个坑过，查了半天才发现是 while 写成了 if。

无锁队列：高性能场景的利器

加锁的队列虽然简单可靠，但性能上有个问题：锁竞争。当生产者和消费者都很多的时候，锁就成了瓶颈。

无锁队列，说白了就是不用锁，靠原子操作来保证线程安全。最经典的是「多生产者-多消费者无锁队列」，基于 CAS（Compare-And-Swap）实现。

// 无锁队列 - 单生产者单消费者版本
// 这个实现基于环形缓冲区，不需要锁
typedef struct {
    void** buffer;
    int capacity;
    // 注意：head 和 tail 用原子类型
    atomic_int head;   // 消费者读取的位置
    atomic_int tail;   // 生产者写入的位置
} lockfree_queue_t;

// 入队（单生产者）
int lockfree_push(lockfree_queue_t* q, void* data) {
    int tail = atomic_load(&q->tail);
    int next_tail = (tail + 1) % q->capacity;
    
    // 检查队列是否满了
    if (next_tail == atomic_load(&q->head)) {
        return -1;  // 队列满
    }
    
    q->buffer[tail] = data;
    // 写屏障，确保数据写入完成后再更新 tail
    atomic_store(&q->tail, next_tail);
    return 0;
}

// 出队（单消费者）
void* lockfree_pop(lockfree_queue_t* q) {
    int head = atomic_load(&q->head);
    
    // 检查队列是否空了
    if (head == atomic_load(&q->tail)) {
        return NULL;  // 队列空
    }
    
    void* data = q->buffer[head];
    int next_head = (head + 1) % q->capacity;
    atomic_store(&q->head, next_head);
    return data;
}

三种方案的对比

方案	适用场景	性能	复杂度
互斥锁队列	低并发、开发周期短	中等	低
条件变量队列	需要阻塞等待的场景	中等（有上下文切换）	中
无锁队列	高并发、低延迟	高	高

我个人建议：能用锁解决的，先用锁。别一上来就搞无锁队列。无锁队列调试起来太痛苦了，我有一回为了修一个内存序的问题，折腾了整整两天。

消息队列的核心流程

下面这张图展示了生产者-消费者模型的核心逻辑：

选型建议

• 开发初期、并发不高：用互斥锁 + 条件变量。简单可靠，出问题好排查。
• 高并发、低延迟：考虑无锁队列。但要做好心理准备，调试起来很费劲。
• 单生产者单消费者：无锁队列的实现会简单很多，推荐优先考虑。
• 多生产者多消费者：无锁队列的复杂度会翻倍，建议先用锁方案，等性能瓶颈了再优化。

消息队列的设计，说白了就是在「简单」和「高性能」之间找平衡。我见过不少项目，一开始就上了无锁队列，结果开发周期拉长了好几倍，最后性能也没比加锁版本好多少。嗯，选型这事儿，还是得看实际场景。

---