Linux如何构建多线程任务队列_Linux线程架构示例-linux运维-PHP中文网

Linux如何构建多线程任务队列_Linux线程架构示例

尊渡假赌尊渡假赌尊渡假赌

发布： 2025-11-29 21:15:06

原创

255人浏览过

多线程任务队列通过共享队列和线程池实现并发，核心包括任务结构体、队列、线程池及同步机制；使用pthread库创建线程，互斥锁保护队列，条件变量唤醒线程；任务提交后由空闲线程执行，支持并行处理如网络请求等场景；需注意线程安全、内存管理和虚假唤醒等问题。

linux如何构建多线程任务队列_linux线程架构示例

在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);

登录后复制

}

AIBox 一站式AI创作平台

AIBox365一站式AI创作平台，支持ChatGPT、GPT4、Claue3、Gemini、Midjourney等国内外大模型

217

查看详情

// 从队列获取任务（阻塞） 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