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：阻塞队列和普通队列的核心区别？

答：

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

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

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

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

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

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

2. 常见易错坑点

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

总结

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

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

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