C++11实现简单线程池：原理与代码解析

线程池核心架构

线程池的本质是预先创建一组固定数量的工作线程，这些线程在一个循环中持续运行。它们的工作模式遵循典型的“生产者-消费者”逻辑：

1. 主线程（生产者）：负责接收外部任务请求，将任务封装并插入到共享的任务队列尾部。
2. 工作线程（消费者）：从任务队列头部获取任务并执行。如果队列为空，工作线程将进入睡眠状态，释放 CPU 资源；当有新任务加入时，被通知唤醒。

在实现过程中，需要重点处理以下两个关键组件的协同：

• 互斥锁（Mutex）：保护任务队列这一共享资源的原子性操作，确保读写队列时不会发生竞态条件。
• 条件变量（Condition Variable）：用于解决线程的阻塞与唤醒机制。避免工作线程在无任务时进行高频轮询导致的 CPU 浪费。

第1步：定义数据结构与接口

首先定义用于管理工作的链表节点和线程池本体结构体。结构体中包含互斥锁、条件变量、任务队列头指针以及当前队列大小，这些字段都是线程间共享的临界区数据。

typedef struct worker {
    void *(*process)(void *arg);  // 具体的工作函数
    void *arg;                    // 函数的参数
    struct worker *next;          // 链表后驱指针
}Worker;

typedef struct {
    pthread_mutex_t queue_lock;   // 互斥锁
    pthread_cond_t queue_ready;   // 条件变量

    Worker *queue_head;           // 任务队列头结点
    int cur_queue_size;           // 当前队列中的任务数

    int max_thread_num;           // 最大线程数
    pthread_t *threadid;          // 存储线程ID

    int shutdown;                 // 关闭标志
}ThreadPool;

第2步：初始化线程池

调用 pool_init 函数完成初始配置。主要包含三个动作：申请内存、初始化同步对象（锁和条件变量），以及启动指定数量的工作线程。

void pool_init(int max_thread_num) {
    pool = (ThreadPool *) malloc(sizeof(ThreadPool));

    // 初始化锁和条件变量，传入NULL表示使用默认属性
    pthread_mutex_init(&(pool->queue_lock), NULL);
    pthread_cond_init(&(pool->queue_ready), NULL);

    pool->queue_head = NULL;
    pool->cur_queue_size = 0;
    pool->max_thread_num = max_thread_num;
    pool->shutdown = 0;

    // 分配线程ID数组内存
    pool->threadid = (pthread_t *) malloc(max_thread_num * sizeof(pthread_t));

    for (int i = 0; i < max_thread_num; i++) {
        // 创建工作线程
        pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);
    }
}

此时，系统内已有 max_thread_num 个线程在后台运行，等待着后续任务的到来。

第3步：添加任务到队列

主线程通过 pool_add_worker 函数向队列插入新任务。为了确保线程安全，必须先获取互斥锁，修改完队列指针和计数后再释放锁，最后发送信号唤醒可能正在等待的工作线程。

int pool_add_worker(void *(*process)(void *arg), void *arg) {
    Worker *newworker = (Worker *) malloc(sizeof(Worker));
    newworker->process = process;
    newworker->arg = arg;
    newworker->next = NULL;

    pthread_mutex_lock(&(pool->queue_lock));

    // 尾插法增加任务
    Worker *member = pool->queue_head;
    if (member != NULL) {
        while (member->next != NULL)
            member = member->next;
        member->next = newworker;
    } else {
        pool->queue_head = newworker;
    }
    ++pool->cur_queue_size;

    pthread_mutex_unlock(&(pool->queue_lock));

    // 唤醒至少一个处于等待状态的线程
    pthread_cond_signal(&(pool->queue_ready));
    return 0;
}

第4步：设计工作线程例程

这是线程池的核心逻辑。每个工作线程进入无限循环，其执行流程如下：

1. 加锁：访问队列前必须锁定 queue_lock。
2. 判断条件：检查队列是否为空且未收到终止信号。
   若为空：调用 pthread_cond_wait。该函数会自动释放锁并使当前线程挂起，直到有其他线程调用了 signal 或 broadcast。重新被唤醒后，该函数会自动重新获取锁。
若非空：跳出等待逻辑，准备处理任务。
3. 取出任务：从队列头部摘除第一个任务节点，更新队列指针和计数，随后立即解锁，允许其他工作线程同时去取下一个任务。
4. 执行任务：调用任务的回调函数，并释放任务节点内存。

void *thread_routine(void *arg) {
    printf("starting thread 0x%x\n", (int)pthread_self());
    while (1) {
        pthread_mutex_lock(&(pool->queue_lock));

        // 当队列为空且未收到shutdown指令时，等待
        while (pool->cur_queue_size == 0 && !pool->shutdown) {
            printf("thread 0x%x is waiting\n", (int)pthread_self());
            pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
        }

        // 处理关闭信号
        if (pool->shutdown) {
            pthread_mutex_unlock(&(pool->queue_lock));
            printf("thread 0x%x will exit\n", (int)pthread_self());
            pthread_exit(NULL);
        }

        // 取出一个任务
        --pool->cur_queue_size;
        Worker *worker = pool->queue_head;
        pool->queue_head = worker->next;

        pthread_mutex_unlock(&(pool->queue_lock));

        // 执行具体业务逻辑
        (*(worker->process))(worker->arg);
        free(worker);
    }
    pthread_exit(NULL);
}

第5步：安全销毁线程池

清理现场时，需要将 shutdown 标志位设为1，然后通过 pthread_cond_broadcast 一次性唤醒所有沉睡的线程。这样所有线程都能检测到关闭信号退出循环。最后，主线程通过 pthread_join 回收子线程资源，并释放剩余的任务节点和内存块。

int pool_destroy() {
    if (pool->shutdown)
        return -1;

    pool->shutdown = 1;
    // 唤醒所有阻塞的线程，使其感知到shutdown信号
    pthread_cond_broadcast(&(pool->queue_ready));

    // 等待所有线程结束
    for (int i = 0; i < pool->max_thread_num; ++i)
        pthread_join(pool->threadid[i], NULL);

    free(pool->threadid);

    // 清理残留任务
    Worker *head = NULL;
    while (pool->queue_head != NULL) {
        head = pool->queue_head;
        pool->queue_head = pool->queue_head->next;
        free(head);
    }

    pthread_mutex_destroy(&(pool->queue_lock));
    pthread_cond_destroy(&(pool->queue_ready));
    free(pool);
    pool = NULL;
    return 0;
}

编译与测试注意事项

由于使用了 POSIX 线程库，在 Linux/GCC环境下编译时需要链接动态库：
gcc main.cpp -o pool -lpthread

在主函数中，可以按以下方式测试：先初始化 3 个线程，然后循环投递 10 个简单任务（如延时打印），最后暂停 5 秒等待执行完毕再销毁池。

以上就是C++11多线程线程池的详细内容，更多关于C++并发编程的资料请关注本站其它相关文章！