C 语言工业级通用组件手写 04：阻塞队列

前言：

• 承接前面环形缓冲区、内存池、双向链表三大工业级组件，本篇章继续完善 C 语言底层通用组件库。
• 在嵌入式多线程开发、后台服务、异步中间件、日志系统、消息调度场景中，纯数据容器已经无法满足业务需求。环形缓冲区、链表只能解决数据存储问题，无法解决线程同步、流量阻塞、任务等待问题。
• 阻塞队列是工业级多线程架构的核心通信基石，用于解耦生产者线程与消费者线程：生产过快则阻塞、消费过快则等待，完美适配异步解耦、削峰填谷、线程协同场景。
• 本篇基于「双向链表 + 互斥锁 + 条件变量」实现标准工业级阻塞队列，支持空队列阻塞消费、满队列阻塞写入、全自动线程同步、无数据竞争、无虚假唤醒，源码可直接投产使用。

一、阻塞队列的核心本质与应用场景

1. 什么是阻塞队列

普通队列：数据读写立刻返回，不阻塞、不等待。阻塞队列：带线程同步能力的线程安全队列。

具备两大核心阻塞特性：

• 队列为空：消费者线程阻塞等待，不空转、不耗 CPU
• 队列已满：生产者线程阻塞等待，拒绝爆队列、防数据丢失

是真正意义上适配多线程工程架构的消息容器。

2. 解决的核心痛点

• 解决线程空转 CPU 飙高：普通轮询队列 while (1) 空跑，占用大量 CPU 资源。
• 解决生产消费速率不匹配：生产者速度快、消费者慢，自动限流阻塞，防止数据溢出。
• 解决多线程数据竞争：原生链表 / 缓冲区非线程安全，多线程并发必乱，阻塞队列自带锁保护。
• 解决手动同步复杂问题：无需业务层手动判断空满、休眠、唤醒，组件内部全自动管理。
• 解决异步架构解耦问题：线程之间不直接耦合，通过队列完成数据流转，架构更稳定。

3. 典型工业级落地场景

• 异步日志系统：日志线程满阻塞、落盘线程空阻塞，零 CPU 空转。
• 网络消息调度：收包线程生产、解析线程消费，削峰防抖动。
• 嵌入式多线程框架：任务队列、工作线程池、事件调度队列。
• 后台服务中间件：异步消息队列、请求排队、流量限流。
• 音视频流媒体：帧数据生产消费、匀速播放、卡顿优化。

二、核心实现原理

1. 底层存储结构

工业级阻塞队列底层优先使用链表实现（本篇方案）：

• 动态长度、无容量浪费、支持无限扩容（可限最大容量）
• 入队出队 O (1)
• 适配高频动态增删

区别于数组队列：固定容量、容易溢出、灵活性差，不适合工业动态业务。

2. 线程同步模型

标准工业同步三要素：

• 互斥锁 mutex：保护队列读写，保证同一时刻只有一个线程操作队列
• 非空条件变量 not_empty：队列空时阻塞消费者，有数据唤醒消费者
• 非满条件变量 not_full：队列满时阻塞生产者，有空位唤醒生产者

3. 阻塞唤醒逻辑

消费者取数据： 队列空 → 阻塞等待 not_empty 队列有数据 → 被唤醒 → 取走数据 → 唤醒生产者

生产者写数据： 队列满 → 阻塞等待 not_full 队列有空位 → 被唤醒 → 写入数据 → 唤醒消费者

4. 防虚假唤醒设计

POSIX 条件变量存在虚假唤醒机制（无信号也可能醒）。 工业级必须用while 循环判断，不能用 if，杜绝虚假唤醒导致的逻辑错乱。

三、工业级设计规范

1. 封装性设计

不透明结构体封装，内部锁、条件变量、队列节点、容量全部隐藏。 外部只能通过统一接口操作，杜绝外部篡改同步状态导致死锁、崩溃。

2. 接口设计原则

3. 鲁棒性要求

• 所有接口做空指针校验、参数合法性校验
• 自动处理虚假唤醒，工业级标准 while 等待
• 锁生命周期闭环，无死锁、无漏解锁
• 队列销毁安全，防止线程野唤醒
• 支持阻塞 / 非阻塞双模式，适配不同业务

4. 线程安全约束

天然线程安全支持：多生产者、多消费者并发场景 所有同步、阻塞、唤醒逻辑内部封装完成，业务层零加锁。

四、完整可复用源码

1. 头文件 block_queue.h

#ifndef BLOCK_QUEUE_H #define BLOCK_QUEUE_H #include <stdint.h> #include <stddef.h> #include <pthread.h> #ifdef __cplusplus extern "C" { #endif /* 不透明结构体 */ typedef struct block_queue block_queue_t; /** * @brief 创建阻塞队列 * @param max_capacity 队列最大容量 * @return 成功返回句柄，失败NULL */ block_queue_t *block_queue_create(size_t max_capacity); /** * @brief 销毁阻塞队列 * @param queue 队列句柄 */ void block_queue_destroy(block_queue_t *queue); /** * @brief 阻塞式入队 * @param queue 队列句柄 * @param data 数据指针 * @return 0成功 -1失败 */ int block_queue_push(block_queue_t *queue, void *data); /** * @brief 阻塞式出队 * @param queue 队列句柄 * @return 出队数据指针 */ void *block_queue_pop(block_queue_t *queue); /** * @brief 非阻塞入队 */ int block_queue_try_push(block_queue_t *queue, void *data); /** * @brief 非阻塞出队 */ void *block_queue_try_pop(block_queue_t *queue); /** * @brief 获取当前队列元素数量 */ size_t block_queue_size(block_queue_t *queue); #ifdef __cplusplus } #endif #endif

2. 实现文件 block_queue.c

#include "block_queue.h" #include <stdlib.h> #include <string.h> /* 队列节点 */ typedef struct queue_node { void *data; struct queue_node *next; } queue_node_t; /* 阻塞队列结构体（完全隐藏） */ struct block_queue { queue_node_t *head; queue_node_t *tail; size_t size; size_t max_capacity; pthread_mutex_t mutex; pthread_cond_t not_empty; pthread_cond_t not_full; }; block_queue_t *block_queue_create(size_t max_capacity) { if (max_capacity == 0) { return NULL; } block_queue_t *queue = (block_queue_t *)malloc(sizeof(block_queue_t)); if (queue == NULL) { return NULL; } memset(queue, 0, sizeof(block_queue_t)); queue->max_capacity = max_capacity; queue->head = NULL; queue->tail = NULL; queue->size = 0; pthread_mutex_init(&queue->mutex, NULL); pthread_cond_init(&queue->not_empty, NULL); pthread_cond_init(&queue->not_full, NULL); return queue; } void block_queue_destroy(block_queue_t *queue) { if (queue == NULL) { return; } pthread_mutex_lock(&queue->mutex); /* 清空所有节点 */ queue_node_t *cur = queue->head; while (cur != NULL) { queue_node_t *tmp = cur; cur = cur->next; free(tmp); } queue->head = NULL; queue->tail = NULL; queue->size = 0; pthread_mutex_unlock(&queue->mutex); /* 销毁同步资源 */ pthread_mutex_destroy(&queue->mutex); pthread_cond_destroy(&queue->not_empty); pthread_cond_destroy(&queue->not_full); free(queue); } int block_queue_push(block_queue_t *queue, void *data) { if (queue == NULL || data == NULL) { return -1; } pthread_mutex_lock(&queue->mutex); /* 满队列阻塞，while防虚假唤醒 */ while (queue->size >= queue->max_capacity) { pthread_cond_wait(&queue->not_full, &queue->mutex); } /* 创建新节点 */ queue_node_t *node = (queue_node_t *)malloc(sizeof(queue_node_t)); if (node == NULL) { pthread_mutex_unlock(&queue->mutex); return -1; } node->data = data; node->next = NULL; if (queue->tail == NULL) { queue->head = node; queue->tail = node; } else { queue->tail->next = node; queue->tail = node; } queue->size++; /* 唤醒阻塞消费者 */ pthread_cond_signal(&queue->not_empty); pthread_mutex_unlock(&queue->mutex); return 0; } void *block_queue_pop(block_queue_t *queue) { if (queue == NULL) { return NULL; } pthread_mutex_lock(&queue->mutex); /* 空队列阻塞，while防虚假唤醒 */ while (queue->size == 0) { pthread_cond_wait(&queue->not_empty, &queue->mutex); } queue_node_t *node = queue->head; void *data = node->data; queue->head = queue->head->next; queue->size--; if (queue->head == NULL) { queue->tail = NULL; } free(node); /* 唤醒阻塞生产者 */ pthread_cond_signal(&queue->not_full); pthread_mutex_unlock(&queue->mutex); return data; } int block_queue_try_push(block_queue_t *queue, void *data) { if (queue == NULL || data == NULL) { return -1; } pthread_mutex_lock(&queue->mutex); if (queue->size >= queue->max_capacity) { pthread_mutex_unlock(&queue->mutex); return -1; } queue_node_t *node = (queue_node_t *)malloc(sizeof(queue_node_t)); if (node == NULL) { pthread_mutex_unlock(&queue->mutex); return -1; } node->data = data; node->next = NULL; if (queue->tail == NULL) { queue->head = node; queue->tail = node; } else { queue->tail->next = node; queue->tail = node; } queue->size++; pthread_cond_signal(&queue->not_empty); pthread_mutex_unlock(&queue->mutex); return 0; } void *block_queue_try_pop(block_queue_t *queue) { if (queue == NULL) { return NULL; } pthread_mutex_lock(&queue->mutex); if (queue->size == 0) { pthread_mutex_unlock(&queue->mutex); return NULL; } queue_node_t *node = queue->head; void *data = node->data; queue->head = queue->head->next; queue->size--; if (queue->head == NULL) { queue->tail = NULL; } free(node); pthread_cond_signal(&queue->not_full); pthread_mutex_unlock(&queue->mutex); return data; } size_t block_queue_size(block_queue_t *queue) { if (queue == NULL) { return 0; } pthread_mutex_lock(&queue->mutex); size_t size = queue->size; pthread_mutex_unlock(&queue->mutex); return size; }

五、实战演示：生产者消费者线程示例

#include <stdio.h> #include <unistd.h> #include <pthread.h> #include "block_queue.h" #define QUEUE_MAX_CAP 8 // 模拟业务数据 typedef struct { int seq; char msg[32]; } task_data_t; void *producer_thread(void *arg) { block_queue_t *queue = (block_queue_t *)arg; for (int i = 1; i <= 20; i++) { task_data_t *task = (task_data_t *)malloc(sizeof(task_data_t)); task->seq = i; snprintf(task->msg, 32, "task_%d", i); // 阻塞入队，队列满自动等待 block_queue_push(queue, task); printf("【生产者】投递任务：%d\n", i); usleep(100000); } return NULL; } void *consumer_thread(void *arg) { block_queue_t *queue = (block_queue_t *)arg; while (1) { // 队列为空自动阻塞等待 task_data_t *task = (task_data_t *)block_queue_pop(queue); if (task == NULL) continue; printf("【消费者】处理任务：%d -> %s\n", task->seq, task->msg); free(task); usleep(200000); } return NULL; } int main(void) { block_queue_t *queue = block_queue_create(QUEUE_MAX_CAP); pthread_t producer, consumer; pthread_create(&producer, NULL, producer_thread, queue); pthread_create(&consumer, NULL, consumer_thread, queue); pthread_join(producer, NULL); pthread_join(consumer, NULL); block_queue_destroy(queue); return 0; }

运行效果：

• 队列满时生产者自动阻塞，不再疯狂生产
• 队列空时消费者休眠等待，CPU 0 占用
• 生产消费速率不匹配自动平衡，无数据丢失、无错乱

六、工业级进阶优化方向

1. 超时阻塞等待

增加timed_push / timed_pop接口，支持限时阻塞，防止永久死等。

2. 批量入队出队

批量操作减少锁竞争，大幅提升高并发吞吐。

3. 队列优雅退出标记

增加退出标志，支持线程安全退出，避免销毁时线程卡死。

4. 内存池搭配使用

任务节点不再频繁 malloc，搭配 02 内存池，彻底消除内存碎片。

七、高频面试考点与易错坑点

1. 经典面试问答

Q1：阻塞队列和普通队列的核心区别？

答：

1. 普通队列无同步机制、无阻塞能力，多线程不安全，需要业务层轮询判断，CPU 占用高。
2. 阻塞队列自带锁与条件变量，空阻塞、满阻塞、自动唤醒，零空转、线程安全、解耦能力极强，是多线程工业架构标准组件。

Q2：条件变量为什么必须 while 循环等待，不能 if？

答：POSIX 条件变量存在虚假唤醒，线程可能在无信号情况下被唤醒。 if 判断会导致逻辑错乱、取空数据、越界报错。while 循环会二次校验状态，是工业级强制规范。

Q3：mutex 为什么必须和 cond 绑定使用？

答：条件变量本身不具备数据保护能力，必须配合互斥锁保证队列状态原子性，防止并发篡改导致死锁、丢失唤醒。

Q4：阻塞队列适用于什么架构？

答：生产者消费者模型、线程池任务队列、异步解耦架构、流量削峰系统，是所有多线程异步架构的底层基石。

2. 常见易错坑点

虚假唤醒不处理，if 等待导致偶发崩溃 不加锁直接操作队列，多线程数据错乱 唤醒丢失、漏解锁导致死锁 队列销毁时机错误，线程野指针访问 消费者无限空转，CPU 100% 占用

总结

阻塞队列是C 语言多线程工业开发的终极同步容器。 如果说前三个组件解决了数据存储、内存管理、对象管理，本篇阻塞队列解决的是线程协同、异步解耦、流量控制的工程难题。

---

本篇组件零空转、无竞争、线程安全、可直接投产，是实现线程池、异步日志、消息中间件的必备底层组件。

创作不易，如果对你有帮助，欢迎点赞、收藏、转发。