用 C 写一个线程安全的生产者-消费者队列：pthread + 互斥锁 + 条件变量

写多线程程序时，最先遇到的往往不是算法，而是数据怎么在线程之间安全地传递。

一个线程负责采集或计算，另一个线程负责落盘、发送或处理。如果两个线程直接读写同一个数组，很快就会遇到三类问题：数据还没写完就被读走、数组写满后覆盖了没处理的数据、队列为空时消费者只能空转轮询，白白烧 CPU。

有界阻塞队列就是解决这个问题的经典模型：队列满时生产者阻塞等待，队列空时消费者阻塞等待。本文用 pthread 实现一个这样的队列，并顺带讲清两个排查起来最头疼的问题——数据竞争和死锁——分别怎么定位。

队列的设计：一把锁、两个条件变量、一个环形缓冲区

队列内部用环形缓冲区存数据。相比链表，环形缓冲区不需要频繁分配节点，访问也更集中；只需要记录三个位置信息：

• head：下一个出队的位置
• tail：下一个入队的位置
• count：当前元素个数，用它判断满和空，避免只靠 head 和 tail 时分不清空与满

同步上只用一把互斥锁，但要用两个条件变量：

为什么不用一个条件变量？因为生产者和消费者等待的是两个不同的条件。如果共用一个变量，入队后发出的信号可能唤醒另一个生产者，它醒来发现队列还是满的，只能再次睡下，而真正等待数据的消费者反而没被叫醒。用两个条件变量，唤醒对象从一开始就是明确的。

还有一个容易被忽略的状态：closed。队列不能只解决“怎么传”，还要解决“怎么结束”。生产者全部退出后，需要一种方式告诉消费者：不会再有新数据了，把剩下的取完就可以退出。只靠超时轮询判断退出既慢又不可靠，所以这里用一个关闭标志，配合广播唤醒所有等待者。

完整代码：可直接编译运行

下面是完整示例。队列容量为 8，2 个生产者各生产 20 条数据，2 个消费者共同消费，最后校验消费总数和数据总和是否与预期一致。校验通过会输出 OK。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#define QUEUE_CAP 8
#define PRODUCER_NUM 2
#define CONSUMER_NUM 2
#define ITEMS_PER_PRODUCER 20

typedef struct {
    int *buffer;              /* 环形缓冲区，动态分配 */
    int capacity;             /* 容量上限 */
    int head;                 /* 出队位置 */
    int tail;                 /* 入队位置 */
    int count;                /* 当前元素个数 */
    int closed;               /* 1 表示不再接收新数据 */
    pthread_mutex_t mutex;    /* 保护以上全部字段 */
    pthread_cond_t not_full;  /* 队列不满时可写，生产者等它 */
    pthread_cond_t not_empty; /* 队列非空时可读，消费者等它 */
} bounded_queue_t;

int bq_init(bounded_queue_t *q, int capacity) {
    if (capacity <= 0) return -1;
    q->buffer = malloc(sizeof(int) * (size_t)capacity);
    if (!q->buffer) return -1;
    q->capacity = capacity;
    q->head = 0;
    q->tail = 0;
    q->count = 0;
    q->closed = 0;
    pthread_mutex_init(&q->mutex, NULL);
    pthread_cond_init(&q->not_full, NULL);
    pthread_cond_init(&q->not_empty, NULL);
    return 0;
}

void bq_destroy(bounded_queue_t *q) {
    pthread_mutex_destroy(&q->mutex);
    pthread_cond_destroy(&q->not_full);
    pthread_cond_destroy(&q->not_empty);
    free(q->buffer);
    q->buffer = NULL;
}

入队和出队是对称的：先加锁，再检查条件，不满足就等待，满足才操作数据，最后发信号通知对方。

/* 入队：队列满时阻塞等待；队列已关闭时返回 -1 */
int bq_push(bounded_queue_t *q, int value) {
    pthread_mutex_lock(&q->mutex);
    while (q->count == q->capacity && !q->closed) {
        pthread_cond_wait(&q->not_full, &q->mutex);
    }
    if (q->closed) {
        pthread_mutex_unlock(&q->mutex);
        return -1;
    }
    q->buffer[q->tail] = value;
    q->tail = (q->tail + 1) % q->capacity;
    q->count++;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->mutex);
    return 0;
}

/* 出队：队列空时阻塞等待；队列已关闭且已取空时返回 -1 */
int bq_pop(bounded_queue_t *q, int *out) {
    pthread_mutex_lock(&q->mutex);
    while (q->count == 0 && !q->closed) {
        pthread_cond_wait(&q->not_empty, &q->mutex);
    }
    if (q->count == 0) { /* 已关闭且取空 */
        pthread_mutex_unlock(&q->mutex);
        return -1;
    }
    *out = 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 0;
}

/* 关闭队列：唤醒所有等待者，让消费者取空后退出 */
void bq_close(bounded_queue_t *q) {
    pthread_mutex_lock(&q->mutex);
    q->closed = 1;
    pthread_cond_broadcast(&q->not_empty);
    pthread_cond_broadcast(&q->not_full);
    pthread_mutex_unlock(&q->mutex);
}

主函数负责把流程串起来：先启动消费者，再启动生产者；等生产者全部结束后再关闭队列，消费者取完剩余数据自然退出。

typedef struct {
    bounded_queue_t *q;
    int id;
} worker_arg_t;

static long g_consumed_count = 0;
static long g_consumed_sum = 0;
static pthread_mutex_t g_stat_mutex = PTHREAD_MUTEX_INITIALIZER;

void *producer(void *arg) {
    worker_arg_t *w = (worker_arg_t *)arg;
    for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
        int value = w->id * 1000 + i;
        if (bq_push(w->q, value) != 0) break;
    }
    return NULL;
}

void *consumer(void *arg) {
    worker_arg_t *w = (worker_arg_t *)arg;
    int value;
    while (bq_pop(w->q, &value) == 0) {
        pthread_mutex_lock(&g_stat_mutex);
        g_consumed_count++;
        g_consumed_sum += value;
        pthread_mutex_unlock(&g_stat_mutex);
    }
    return NULL;
}

int main(void) {
    bounded_queue_t q;
    if (bq_init(&q, QUEUE_CAP) != 0) {
        fprintf(stderr, "bq_init failed\n");
        return 1;
    }

    pthread_t prod[PRODUCER_NUM], cons[CONSUMER_NUM];
    worker_arg_t parg[PRODUCER_NUM], carg[CONSUMER_NUM];

    for (int i = 0; i < CONSUMER_NUM; i++) {
        carg[i].q = &q; carg[i].id = i;
        pthread_create(&cons[i], NULL, consumer, &carg[i]);
    }
    for (int i = 0; i < PRODUCER_NUM; i++) {
        parg[i].q = &q; parg[i].id = i;
        pthread_create(&prod[i], NULL, producer, &parg[i]);
    }

    for (int i = 0; i < PRODUCER_NUM; i++) pthread_join(prod[i], NULL);
    bq_close(&q); /* 生产结束，通知消费者取空退出 */
    for (int i = 0; i < CONSUMER_NUM; i++) pthread_join(cons[i], NULL);

    long expected_count = (long)PRODUCER_NUM * ITEMS_PER_PRODUCER;
    long expected_sum = 0;
    for (int p = 0; p < PRODUCER_NUM; p++)
        for (int i = 0; i < ITEMS_PER_PRODUCER; i++)
            expected_sum += p * 1000 + i;

    printf("consumed_count=%ld expected=%ld\n", g_consumed_count, expected_count);
    printf("consumed_sum=%ld expected=%ld\n", g_consumed_sum, expected_sum);
    printf("%s\n", (g_consumed_count == expected_count &&
                    g_consumed_sum == expected_sum) ? "OK" : "MISMATCH");

    bq_destroy(&q);
    return 0;
}

编译和运行：

gcc -std=c11 -Wall -Wextra -pthread bounded_queue.c -o bounded_queue
./bounded_queue

预期输出：

consumed_count=40 expected=40
consumed_sum=20380 expected=20380
OK

这段校验不是摆设。并发代码最怕“看起来跑通了”：多消费一条、少消费一条，单次运行都未必看得出来。用总数和总和双重校验连跑多次，才能初步确认没有丢数据、没有重复消费。后面讲数据竞争时会再回到这一点。

最容易写错的一行：等待条件变量必须用 while

上面的代码里，等待写的是 while，不是 if：

while (q->count == 0 && !q->closed) {
    pthread_cond_wait(&q->not_empty, &q->mutex);
}

这是新手最常见的坑，而且用 if 写的版本在简单测试里往往也能跑，问题要到多消费者、高并发时才暴露。原因有两个。

第一是虚假唤醒。pthread_cond_wait 允许在没有收到任何信号的情况下返回，标准并不保证“被唤醒就一定是有人发了信号”。用 if 时，线程醒来后不复查条件就直接取数据，队列可能仍然是空的，读到的就是无效位置。

第二是唤醒后条件已经变了。假设队列里只有一条数据，广播或连续信号唤醒了两个消费者：第一个消费者取走数据，第二个消费者随后拿到锁往下执行时，队列已经重新变空。用 while，它会重新判断条件并再次等待；用 if，它会直接去取一个不存在的元素，计数还会变成负数，后面所有判断就都乱了。

记住一个固定写法：条件变量没有独立的语义，它只是“等待某个条件成立”的手段。把条件写成谓词，等待前后都只认谓词：

• 等待前：谓词不成立才等
• 等待后：回到循环开头重新看谓词
• 改状态的人：改完、还在持锁时发信号或广播，然后解锁

另外，pthread_cond_wait 做的事情也要理解清楚：它会原子地释放互斥锁并让线程睡眠，被唤醒后先重新拿回锁再返回。所谓原子，是为了避免“刚释放锁还没睡下，信号就发过来了”这种丢失唤醒。理解这一点，就不会再写出先解锁、再自己拼等待逻辑的代码。

signal 和 broadcast 的选择同样有规律：一次入队只新增一条数据，最多让一个消费者能成功取到，用 signal 就够了；关闭队列则要让所有等待者都醒来重新判断状态，必须用 broadcast。凡是“状态变化可能让多类等待者都需要重新判断”的场合，都优先考虑广播。

数据竞争怎么排查：先让工具说话

数据竞争指的是两个线程并发访问同一块内存，至少有一个是写操作，且中间没有同步。它的典型表现是：偶发、难复现、printf 一加就消失、开 -O2 就出现、换台机器就变样。遇到这种现象，先别靠读代码猜，第一步是让工具把竞争点抓出来。

第一步：用 ThreadSanitizer 定位

GCC 和 Clang 都支持 ThreadSanitizer，重新编译时加上选项即可，业务代码不用改：

gcc -std=c11 -g -O1 -fsanitize=thread -pthread bounded_queue.c -o bq_tsan
./bq_tsan

一旦检测到竞争，它会直接报出两个访问点的函数、行号和线程创建栈。比如把示例中统计计数 g_consumed_count++ 外面的锁去掉，两个消费者同时累加，ThreadSanitizer 就会指向这一行，而不是让你对着整个文件猜。

注意两点：带 sanitizer 的二进制只能用于测试，不要发布；而且它只能测到实际跑过的路径，所以测试用例要尽量跑满多生产者、多消费者和队列打满、取空这些边界。

第二步：小问题用 Helgrind 补充

如果环境不方便用 ThreadSanitizer，可以用 Valgrind 的 Helgrind：

gcc -std=c11 -g -pthread bounded_queue.c -o bounded_queue
valgrind --tool=helgrind ./bounded_queue

Helgrind 不用重新编译插桩，还能顺带报锁顺序不一致这类潜在死锁风险，代价是运行速度慢很多，适合小用例复现，不适合压测。

第三步：回到代码看访问清单

工具给出竞争地址后，按地址倒查：这个变量到底被哪些线程、哪些函数访问？逐个确认每处访问是否持同一把锁。最常见的遗漏有三种：

• 只保护了写，没保护读，以为“只是读一下没关系”——读到写一半的值同样是竞争
• 两处访问各用一把不同的锁，看起来都加锁了，实际互不相干
• 临界区只圈住了数据操作，漏掉了与之配套的计数或状态标志，本例中就是 count 必须和缓冲区在同一把锁里改

修复后把校验用例连跑几十次，并分别在普通构建和 ThreadSanitizer 构建下各跑一轮，两边都干净才算过。

死锁怎么排查：先抓现场，再谈修改

死锁和数据竞争的表现正好相反：程序不是偶发算错，而是直接停住不动，CPU 占用接近零。四个必要条件——互斥、持有并等待、不可抢占、循环等待——同时成立才会死锁，其中最值得下手的是循环等待，因为它完全由加锁顺序决定。

现场排查：attach 上去看每个线程卡在哪

程序卡住时不要先 kill，先用 gdb 附上去看全部线程的栈：

gdb -p <进程PID>
(gdb) thread apply all bt
(gdb) info threads

重点看两件事：每个线程各卡在哪个 pthread_mutex_lock 或 pthread_cond_wait 上，以及它们等待的锁地址是不是同几把。如果线程 A 持有锁 1 等锁 2、线程 B 持有锁 2 等锁 1，栈上一眼就能对上，这就是典型的加锁顺序反转。

等待条件变量时还要分清一种假死锁：线程都停在 pthread_cond_wait 上，看起来像锁死，其实可能只是没人发信号，或发信号时谓词还没改对。判断方法是看生产者线程在干什么——如果生产者早已退出却没调用关闭、没广播，消费者就会永远等下去。本例把关闭动作单独做成 bq_close 并广播两个条件变量，就是为了堵住这个口子。

预防：三条能落地的规矩

• 全局统一加锁顺序。项目里只要有两把以上锁，就把获取顺序写成明文规矩，所有路径只按这个顺序拿锁，释放顺序则不作要求。需要反向获取时，先释放再按正序重新拿。

• 临界区只放共享数据操作。不要持锁调用日志、回调、网络收发这类耗时或行为不可控的代码，它们会把临界区拉长，还可能在回调里反过来拿另一把锁，当场构成循环等待。

• 能一把锁解决就先用一把锁。本例全程只有队列这一把锁（统计计数另有一把，但两者从不嵌套获取），就不存在顺序问题。等性能剖析真正指出锁竞争是瓶颈时，再考虑拆细粒度锁；拆锁的同时必须重新审计全部加锁路径。

开发阶段还可以给互斥锁设置错误检查属性，重复加锁、未持锁解锁这类误用会直接返回错误码，而不是默默卡死：

pthread_mutexattr_t attr;
pthread_mutexattr_init(&attr);
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
pthread_mutex_init(&mutex, &attr);
pthread_mutexattr_destroy(&attr);

它只能查出同线程重复加锁等低级错误，查不出跨线程的顺序反转，但作为调试期的护栏，成本几乎为零。

小结：先写对，再谈无锁

这个队列真正值得记住的只有四点：共享状态全部收进一把锁里改；条件变量等待一律配 while 循环复查谓词；唤醒分清 signal 与 broadcast 的场合；用关闭标志加广播给出明确的结束协议。做到这四点，多生产者多消费者也不会丢数据、不会假死。

排查方法同样可以收成两句：结果偶发算错，先跑 ThreadSanitizer 抓数据竞争；程序直接卡住，先 attach 看全部线程栈对锁顺序。

很多人写完这一版就会想换成无锁队列。无锁结构在高竞争下确实可能更快，但它把正确性验证的难度抬高了一个量级：内存序、ABA 问题、生命周期管理，任何一处出错都比锁版本更难复现和定位。建议的练习顺序是：先给本文的队列加上超时等待版本（pthread_cond_timedwait），再试着把元素类型从 int 改成结构体指针、支持任意任务对象，最后再去读无锁队列的实现并对比测试数据。能把锁版本写对、测透的人，换到无锁时才知道自己到底在换什么。