多线程任务队列通过共享队列和线程池实现并发，核心包括任务结构体、队列、线程池及同步机制；使用pthread库创建线程，互斥锁保护队列，条件变量唤醒线程；任务提交后由空闲线程执行，支持并行处理如网络请求等场景；需注意线程安全、内存管理和虚假唤醒等问题。
在Linux环境下，多线程任务队列是一种常见的并发编程模型，适用于需要并行处理大量短时或耗时任务的场景，比如网络服务器、日志处理系统或图像批量处理程序。它通过一个共享的任务队列和多个工作线程协作完成任务分发与执行，提升程序吞吐量和响应速度。
基本组件设计
一个多线程任务队列通常包含以下几个核心部分：
- 任务结构体（Task）：封装要执行的函数及其参数。
- 任务队列（Task Queue）：使用队列数据结构（如链表或循环缓冲区）存储待处理任务。
- 线程池（Thread Pool）：一组预先创建的工作线程，从队列中取出任务并执行。
- 同步机制：使用互斥锁（mutex）保护共享队列，条件变量（condition variable）实现线程等待/唤醒。
实现步骤与代码框架
以下是一个简化的C语言示例，展示如何在Linux下使用pthread库构建一个基础的多线程任务队列。
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
<p>// 任务函数类型定义
typedef void (<em>task_func_t)(void </em>);</p><p>// 任务结构
typedef struct task {
task_func_t func;
void <em>arg;
struct task </em>next;
} task_t;</p><p>// 任务队列结构
typedef struct {
task_t <em>head;
task_t </em>tail;
pthread_mutex_t lock;
pthread_cond_t cond;
int shutdown;
} task_queue_t;</p><p>// 线程池结构
typedef struct {
pthread_t <em>threads;
int thread_count;
task_queue_t </em>queue;
} thread_pool_t;</p><p>// 初始化任务队列
void queue_init(task_queue_t *q) {
q->head = NULL;
q->tail = NULL;
pthread_mutex_init(&q->lock, NULL);
pthread_cond_init(&q->cond, NULL);
q->shutdown = 0;
}</p><p>// 向队列添加任务
void queue_push(task_queue_t <em>q, task_func_t func, void </em>arg) {
task_t <em>t = (task_t </em>)malloc(sizeof(task_t));
t->func = func;
t->arg = arg;
t->next = NULL;</p><pre class='brush:php;toolbar:false;'>pthread_mutex_lock(&q->lock);
if (q->tail) {
q->tail->next = t;
} else {
q->head = t;
}
q->tail = t;
pthread_cond_signal(&q->cond); // 唤醒一个等待线程
pthread_mutex_unlock(&q->lock);}
// 从队列获取任务（阻塞） task_t queue_pop(task_queue_t q) { pthread_mutex_lock(&q->lock); while (q->head == NULL && !q->shutdown) { pthread_cond_wait(&q->cond, &q->lock); }
if (q->shutdown) {
pthread_mutex_unlock(&q->lock);
return NULL;
}
task_t *t = q->head;
q->head = t->next;
if (!q->head) q->tail = NULL;
pthread_mutex_unlock(&q->lock);
return t;}
// 工作线程主函数 void worker(void arg) { task_queue_t q = (task_queue_t)arg;
while (1) {
task_t *t = queue_pop(q);
if (!t) break; // 收到关闭信号
t->func(t->arg);
free(t);
}
return NULL;}
// 创建线程池 thread_pool_t pool_create(int num_threads) { thread_pool_t pool = (thread_pool_t)malloc(sizeof(thread_pool_t)); pool->thread_count = num_threads; pool->threads = (pthread_t)malloc(num_threads sizeof(pthread_t)); pool->queue = (task_queue_t)malloc(sizeof(task_queue_t));
queue_init(pool->queue);
for (int i = 0; i < num_threads; i++) {
pthread_create(&pool->threads[i], NULL, worker, pool->queue);
}
return pool;}
// 提交任务到线程池 void pool_submit(thread_pool_t pool, task_func_t func, void arg) { queue_push(pool->queue, func, arg); }
// 销毁线程池 void pool_destroy(thread_pool_t *pool) { pthread_mutex_lock(&pool->queue->lock); pool->queue->shutdown = 1; pthread_cond_broadcast(&pool->queue->cond); pthread_mutex_unlock(&pool->queue->lock);
for (int i = 0; i < pool->thread_count; i++) {
pthread_join(pool->threads[i], NULL);
}
free(pool->threads);
free(pool->queue);
free(pool);}
使用示例
下面是一个简单的测试函数，演示如何提交任务：
void print_task(void *arg) {
int id = *(int*)arg;
printf("正在执行任务 %d\n", id);
sleep(1); // 模拟耗时操作
}
<p>int main() {
thread_pool_t *pool = pool_create(4); // 创建4个线程</p><pre class='brush:php;toolbar:false;'>for (int i = 0; i < 8; i++) {
int *id = (int*)malloc(sizeof(int));
*id = i;
pool_submit(pool, print_task, id);
}
sleep(2); // 等待任务执行
pool_destroy(pool);
return 0;}
编译时需链接pthread库：
gcc -o thread_pool thread_pool.c -lpthread
关键注意事项
- 任务函数内部不应访问共享资源，除非自行加锁，否则容易引发竞态条件。
- 动态分配的参数（如上面的id）需确保在任务执行完后才释放，避免悬空指针。
- 条件变量配合while循环检查条件，防止虚假唤醒。
- 线程安全的内存管理需谨慎，尤其是任务取消或异常退出时的资源清理。
基本上就这些。这个模型虽简单，但足以支撑大多数后台任务调度需求。根据实际需要可扩展为支持优先级队列、动态扩容线程数或定时任务等功能。不复杂但容易忽略细节。
