线程池原理与实现

说实话，线程池这东西，我刚开始学多线程的时候觉得它挺玄乎的。不就是一堆线程嘛，用的时候创建，用完销毁不就完了？后来我在一个高并发服务项目里被现实狠狠教育了一顿——每次请求都创建线程，系统直接扛不住，CPU 上下文切换开销大得吓人。嗯，从那以后，我才真正理解了线程池的价值。

线程池设计思想

线程池的核心思想，说白了就四个字：复用 和 管控。

你想想看，线程的创建和销毁是有代价的。每次 pthread_create 都要分配栈空间、初始化资源，频繁操作就像你每次用工具都现买一个，用完就扔——太浪费了。线程池就是提前准备好一批线程，放在池子里，有任务就丢进去，线程自己抢着干。

我个人习惯把线程池比作一个「维修队」。队长（线程池管理器）手下有几个固定工人（工作线程），门口挂着一个任务黑板（任务队列）。客户来了，把需求写在黑板上，哪个工人闲了就去瞅一眼，领走任务干活。

线程池三大核心要素：

  • 线程池管理器：负责创建、销毁线程，调整线程数量
  • 任务队列：存放待执行的任务，通常是线程安全的队列
  • 工作线程：真正干活的线程，循环从队列取任务执行

我在项目中遇到过一种情况：任务来得忽快忽慢，如果线程数固定，高峰期排队太长，低峰期又浪费资源。所以后来我设计的线程池都支持动态调整——但这属于进阶玩法，咱们先把基础的搞明白。

任务队列管理

任务队列是线程池的「缓冲区」。生产者（提交任务的代码）往队列里放任务，消费者（工作线程）从队列里取任务。这里有个关键问题：多线程同时操作队列，必须加锁。

我常用的数据结构是 pthread_mutex_t 配合 pthread_cond_t。互斥锁保证队列操作的原子性，条件变量用来通知工作线程「有活干了」。

我的经验：任务队列最好用链表实现，因为任务数量不确定，数组扩容会有性能抖动。当然，如果你能预估最大任务数，用环形缓冲区更高效。

任务队列里存的是什么？通常是一个函数指针加上它的参数。在 C 语言里，我习惯这样定义：

typedef struct task_s {
    void (*func)(void *arg);  // 任务函数
    void *arg;                // 函数参数
    struct task_s *next;      // 链表指针
} task_t;

嗯，这里要注意：参数用 void * 是为了通用性。你传什么类型都行，只要在函数里自己强转回去。

工作线程模型

工作线程的模型其实很简单——一个无限循环。伪代码大概是这样的：

while (pool->is_running) {
    lock(mutex);
    while (queue_is_empty && pool->is_running) {
        wait(cond, mutex);  // 没任务就睡觉
    }
    if (!pool->is_running) break;
    task = pop_from_queue();
    unlock(mutex);
    task->func(task->arg);  // 执行任务
    free(task);
}

这里有个细节我吃过亏：为什么用 while 而不是 if 来检查队列是否为空？ 因为条件变量存在「虚假唤醒」——线程可能在没有收到信号的情况下被唤醒。用 while 再检查一次，能保证逻辑正确。

工作线程的数量怎么定？我一般遵循这个原则：

  • CPU 密集型任务：线程数 = CPU 核心数 + 1
  • IO 密集型任务：线程数 = CPU 核心数 × 2（甚至更多）
  • 混合型：根据实际压测调整

避坑指南：我曾经在一个项目里把线程数设成了 100，结果 CPU 全耗在上下文切换上了，任务吞吐量反而下降。后来压测发现 16 个线程是最优的。记住：线程不是越多越好。

简易线程池代码实现

下面我给出一个完整的简易线程池实现。代码不长，但麻雀虽小五脏俱全。

#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>

// 任务节点
typedef struct task_s {
    void (*func)(void *arg);
    void *arg;
    struct task_s *next;
} task_t;

// 线程池结构
typedef struct threadpool_s {
    pthread_t *threads;      // 工作线程数组
    int thread_count;        // 线程数量
    task_t *head;            // 任务队列头
    task_t *tail;            // 任务队列尾
    pthread_mutex_t mutex;   // 互斥锁
    pthread_cond_t cond;     // 条件变量
    int is_running;          // 运行标志
} threadpool_t;

// 工作线程函数
static void *worker(void *arg) {
    threadpool_t *pool = (threadpool_t *)arg;
    task_t *task;

    while (pool->is_running) {
        pthread_mutex_lock(&pool->mutex);
        // 队列为空且线程池在运行，则等待
        while (pool->head == NULL && pool->is_running) {
            pthread_cond_wait(&pool->cond, &pool->mutex);
        }
        if (!pool->is_running) {
            pthread_mutex_unlock(&pool->mutex);
            break;
        }
        // 取出任务
        task = pool->head;
        pool->head = task->next;
        if (pool->head == NULL) {
            pool->tail = NULL;
        }
        pthread_mutex_unlock(&pool->mutex);

        // 执行任务
        task->func(task->arg);
        free(task);
    }
    return NULL;
}

// 初始化线程池
threadpool_t *threadpool_create(int thread_count) {
    threadpool_t *pool = malloc(sizeof(threadpool_t));
    pool->thread_count = thread_count;
    pool->head = NULL;
    pool->tail = NULL;
    pool->is_running = 1;
    pthread_mutex_init(&pool->mutex, NULL);
    pthread_cond_init(&pool->cond, NULL);

    pool->threads = malloc(sizeof(pthread_t) * thread_count);
    for (int i = 0; i < thread_count; i++) {
        pthread_create(&pool->threads[i], NULL, worker, pool);
    }
    return pool;
}

// 提交任务
void threadpool_submit(threadpool_t *pool, void (*func)(void *), void *arg) {
    task_t *task = malloc(sizeof(task_t));
    task->func = func;
    task->arg = arg;
    task->next = NULL;

    pthread_mutex_lock(&pool->mutex);
    if (pool->tail == NULL) {
        pool->head = task;
        pool->tail = task;
    } else {
        pool->tail->next = task;
        pool->tail = task;
    }
    pthread_cond_signal(&pool->cond);  // 唤醒一个工作线程
    pthread_mutex_unlock(&pool->mutex);
}

// 销毁线程池
void threadpool_destroy(threadpool_t *pool) {
    pthread_mutex_lock(&pool->mutex);
    pool->is_running = 0;
    pthread_cond_broadcast(&pool->cond);  // 唤醒所有线程
    pthread_mutex_unlock(&pool->mutex);

    for (int i = 0; i < pool->thread_count; i++) {
        pthread_join(pool->threads[i], NULL);
    }

    // 清理剩余任务
    task_t *task = pool->head;
    while (task) {
        task_t *next = task->next;
        free(task);
        task = next;
    }

    pthread_mutex_destroy(&pool->mutex);
    pthread_cond_destroy(&pool->cond);
    free(pool->threads);
    free(pool);
}

使用示例：

void my_task(void *arg) {
    int *num = (int *)arg;
    printf("任务执行: %d\n", *num);
}

int main() {
    threadpool_t *pool = threadpool_create(4);
    int nums[] = {1, 2, 3, 4, 5};
    for (int i = 0; i < 5; i++) {
        threadpool_submit(pool, my_task, &nums[i]);
    }
    sleep(1);  // 等待任务完成
    threadpool_destroy(pool);
    return 0;
}

线程池核心流程

下面这张图展示了线程池的完整工作流程，我建议你多看几遍，理解清楚每个环节的交互关系：

线程池核心工作流程 主程序 / 调用方 threadpool_submit() 任务队列（链表） pthread_mutex_t pthread_cond_t 取出任务 工作线程池 线程 1 线程 2 线程 3 ... 执行任务：task->func(task->arg) 循环取任务 threadpool_destroy() broadcast 唤醒

这张图把整个流程串起来了：主程序提交任务到队列，工作线程从队列取任务执行，互斥锁和条件变量保证线程安全。销毁时广播唤醒所有线程，让它们优雅退出。

注意：上面的代码是教学版，生产环境还需要考虑：

  • 任务队列的容量限制（防止内存暴涨）
  • 线程池的动态扩容/缩容
  • 任务的超时处理
  • 更优雅的关闭策略（等待当前任务完成 vs 立即停止）

说实话，线程池的实现看起来简单，但真正用好它需要你对多线程的同步机制有深刻理解。我建议你先把上面的代码跑通，然后试着加一些功能——比如统计任务执行时间、记录线程活跃度。动手实践才是最好的学习方式。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →