构建高效并发系统（C语言线程池从零实现详解）

为什么需要线程池？

直接使用 pthread_create 创建线程虽然简单，但在高并发场景下：

• 线程创建/销毁开销大
• 无法控制并发数量，可能导致系统崩溃
• 缺乏任务调度机制

而多线程编程结合线程池，能有效解决这些问题，提升系统稳定性与吞吐量。

线程池核心组件

一个基本的线程池包含以下部分：

1. 任务队列：存储待执行的任务（函数指针 + 参数）
2. 工作线程数组：固定数量的线程循环取任务执行
3. 互斥锁与条件变量：保证线程安全和任务通知
4. 管理接口：如初始化、添加任务、销毁等

完整代码实现

以下是基于 POSIX 线程（pthread）的线程池实现。我们将分步讲解关键部分。

#include <stdio.h>#include <stdlib.h>#include <pthread.h>#include <unistd.h>// 任务结构体typedef struct {    void (*function)(void *arg);    void *arg;} threadpool_task_t;// 线程池结构体typedef struct {    pthread_mutex_t lock;              // 互斥锁    pthread_cond_t  notify;            // 条件变量    pthread_t       *threads;          // 工作线程数组    threadpool_task_t *queue;          // 任务队列    int thread_count;                  // 线程数量    int queue_size;                    // 队列容量    int head;                          // 队列头    int tail;                          // 队列尾    int count;                         // 当前任务数    int shutdown;                      // 关闭标志    int started;                       // 已启动线程数} threadpool_t;// 工作线程函数static void *threadpool_worker(void *arg) {    threadpool_t *pool = (threadpool_t *)arg;    threadpool_task_t task;    for (;;) {        pthread_mutex_lock(&pool->lock);        // 等待任务或关闭信号        while ((pool->count == 0) && (!pool->shutdown)) {            pthread_cond_wait(&pool->notify, &pool->lock);        }        if (pool->shutdown) {            pthread_mutex_unlock(&pool->lock);            break;        }        // 取出任务        task.function = pool->queue[pool->head].function;        task.arg      = pool->queue[pool->head].arg;        pool->head = (pool->head + 1) % pool->queue_size;        pool->count--;        pthread_mutex_unlock(&pool->lock);        // 执行任务        (*(task.function))(task.arg);    }    pool->started--;    pthread_mutex_unlock(&pool->lock);    pthread_exit(NULL);    return NULL;}// 创建线程池threadpool_t *threadpool_create(int thread_count, int queue_size) {    threadpool_t *pool;    if (thread_count <= 0 || queue_size <= 0) return NULL;    if ((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL) return NULL;    // 初始化锁和条件变量    pthread_mutex_init(&pool->lock, NULL);    pthread_cond_init(&pool->notify, NULL);    // 分配内存    pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count);    pool->queue   = (threadpool_task_t *)malloc(sizeof(threadpool_task_t) * queue_size);    pool->thread_count = thread_count;    pool->queue_size   = queue_size;    pool->head         = 0;    pool->tail         = 0;    pool->count        = 0;    pool->shutdown     = 0;    pool->started      = 0;    // 创建工作线程    for (int i = 0; i < thread_count; ++i) {        if (pthread_create(&pool->threads[i], NULL, threadpool_worker, (void*)pool) != 0) {            threadpool_destroy(pool, 0);            return NULL;        }        pool->started++;    }    return pool;}// 添加任务int threadpool_add(threadpool_t *pool, void (*function)(void *), void *arg) {    int next, err = 0;    if (pool == NULL || function == NULL) return -1;    pthread_mutex_lock(&pool->lock);    if (pool->shutdown) {        err = 1;    } else {        next = (pool->tail + 1) % pool->queue_size;        if (next == pool->head) {            err = 2; // 队列满        } else {            pool->queue[pool->tail].function = function;            pool->queue[pool->tail].arg      = arg;            pool->tail = next;            pool->count++;            // 唤醒一个等待线程            pthread_cond_signal(&pool->notify);        }    }    pthread_mutex_unlock(&pool->lock);    return err;}// 销毁线程池int threadpool_destroy(threadpool_t *pool, int graceful) {    if (pool == NULL) return -1;    pthread_mutex_lock(&pool->lock);    if (pool->shutdown) {        pthread_mutex_unlock(&pool->lock);        return 0;    }    pool->shutdown = (graceful) ? 1 : 2;    pthread_cond_broadcast(&pool->notify);    pthread_mutex_unlock(&pool->lock);    // 等待所有线程结束    for (int i = 0; i < pool->thread_count; ++i) {        pthread_join(pool->threads[i], NULL);    }    // 释放资源    free(pool->threads);    free(pool->queue);    pthread_mutex_destroy(&pool->lock);    pthread_cond_destroy(&pool->notify);    free(pool);    return 0;}

如何使用这个线程池？

下面是一个简单的使用示例，展示如何提交任务并优雅关闭线程池：

void task_function(void *arg) {    int id = *(int *)arg;    printf("Task %d is running on thread %lu\n", id, pthread_self());    sleep(1); // 模拟工作}int main() {    threadpool_t *pool = threadpool_create(4, 16); // 4个线程，队列大小16    for (int i = 0; i < 10; ++i) {        int *id = malloc(sizeof(int));        *id = i;        threadpool_add(pool, task_function, id);    }    sleep(3); // 等待任务完成    threadpool_destroy(pool, 1); // 优雅关闭    return 0;}

应用场景与优化方向

该线程池适用于各类需要高性能服务器的场景，如Web服务器、数据库连接池、日志处理系统等。你可以在此基础上扩展：

• 动态调整线程数量（根据负载）
• 支持任务优先级
• 添加任务超时机制
• 集成内存池减少 malloc/free 开销

总结

通过本教程，你已经掌握了线程池实现的核心思想与完整代码。线程池是 C 语言并发编程的基石，理解其内部机制有助于你构建更稳定、高效的系统。建议你在实际项目中尝试使用，并根据需求进行定制化改进。

关键词回顾：C语言线程池、多线程编程、线程池实现、高性能服务器