当前位置：首页 > news >正文

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

news 2026/7/16 10:36:11

前言：

承接前面环形缓冲区、内存池、双向链表三大工业级组件，本篇章继续完善 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. 接口设计原则

接口函数	功能说明
block_queue_create	创建阻塞队列，设置最大容量
block_queue_destroy	销毁队列，释放所有资源
block_queue_push	阻塞式入队，满则等待
block_queue_pop	阻塞式出队，空则等待
block_queue_try_push	非阻塞入队，不等待
block_queue_try_pop	非阻塞出队，不等待
block_queue_size	获取当前队列元素个数

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; }