Linux 线程池从零实现：pthread 与 C++11 std::thread 双版本实战
1. 进程与线程：30 秒快速理解
1.1 两个「基本单位」
-
进程：操作系统资源分配的基本单位，拥有独立的虚拟地址空间，进程之间相互隔离。
-
线程：CPU 调度的基本单位，寄生在进程内部，共享进程的地址空间和全局数据。
| 对比项 | 进程 Process | 线程 Thread |
|---|---|---|
| 本质 | 资源分配的基本单位 | CPU 调度的基本单位 |
| 地址空间 | 独立，互相隔离 | 共享所属进程的地址空间 |
| 创建方式 | fork() / clone() | pthread_create / std::thread |
| 通信方式 | 管道、信号、共享内存、Socket 等 IPC | 共享变量 + 互斥锁 / 条件变量 |
| 创建与切换开销 | 大 | 小 |
| 崩溃影响 | 互不影响 | 一个线程异常可能影响整个进程 |
| 适用场景 | 隔离要求高、多进程架构 | 大量并发、共享数据多 |
图 1 进程是资源分配的单位，线程是 CPU 调度的单位。
1.2 为什么要「线程池」
线程的创建和销毁是有开销的（内核栈、调度器数据结构、上下文切换）。如果每个任务都现场 pthread_create 再 join，高频短任务场景下开销占比会很高。
线程池的思路是：提前创建好 N 个线程常驻，任务来了丢进队列，由空闲线程取走执行。 好处有三个：
-
复用线程：减少创建 / 销毁开销；
-
控制并发度：防止线程无限制增长拖垮系统；
-
任务与执行解耦：提交方只关心「提交」，不关心「谁执行」。
2. 高频面试题速答（进程 / 线程基础）
Q1：进程和线程有什么区别？进程是资源分配的基本单位（独立地址空间、相互隔离）；线程是 CPU 调度的基本单位（共享地址空间）。进程更重、隔离更好；线程更轻、同步成本更高。
Q2：fork() 为什么「调用一次，返回两次」？fork() 复制当前进程生成子进程：父进程返回子进程 PID（>0），子进程返回 0，失败返回 -1。靠返回值区分父子两条执行流。
Q3：线程同步有哪些方式？互斥锁（mutex）、条件变量（cond）、读写锁（rwlock）、自旋锁（spinlock）、信号量（semaphore）、原子操作（atomic）。嵌入式里最常用互斥锁 + 条件变量。
Q4：死锁产生的四个必要条件？互斥、持有并等待、不可剥夺、循环等待。避免方法：所有线程按同一顺序加锁、一次性申请所有锁、用 trylock 超时放弃、尽量缩小锁粒度。
Q5：条件变量为什么要配合互斥锁使用？「检查条件 → 睡眠」必须是原子的：先加锁检查，不满足就 wait（wait 内部会释放锁并睡眠，被唤醒后重新拿锁）。否则会出现「刚检查完条件、正要睡，任务恰好被提交」的丢失唤醒。
Q6：等待条件为什么用 while 而不是 if？防止虚假唤醒（spurious wakeup）：即使条件不满足，线程也可能被唤醒。所以醒来后必须用 while 重新检查条件，不满足就继续等。
3. 线程池原理与设计要点
图 2 线程池三件套：任务队列 + N 个工作线程 + 互斥锁 / 条件变量。
一个线程池最少需要 5 个成员：
-
任务队列：存放待执行任务（链表 / 队列）。
-
工作线程数组：预先创建 N 个线程，循环取任务。
-
互斥锁：保护任务队列的入队 / 出队。
-
条件变量：队列为空时让工作线程睡眠，有任务时唤醒。
-
停止标志：用于优雅关闭线程池。
三个关键设计点：
-
锁外执行任务：取出任务后立刻解锁，再执行任务函数。否则一个慢任务会阻塞所有线程取任务。
-
while 等待条件：防止虚假唤醒，
pthread_cond_wait必须用 while 循环包裹。 -
优雅退出：置
shutdown标志 →broadcast唤醒所有线程 → 线程处理完剩余任务后自行退出 →join回收。
4. Linux pthread 线程池实现（C 语言）
4.1 设计思路
-
用链表实现任务队列，节点保存函数指针 + 参数；
-
每个工作线程是一个
while(1)循环：没任务就睡在条件变量上，有任务就出队执行； -
提交任务 = 加锁 + 入队 +
signal唤醒一个线程 + 解锁； -
关闭线程池 = 置标志 +
broadcast+join。
4.2 完整代码
/* Linux 线程池实现：pthread 版本 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
/* 任务节点（链表） */
typedef struct task {
void (*func)(void *arg); /* 任务函数 */
void *arg; /* 任务参数 */
struct task *next;
} task_t;
/* 线程池结构 */
typedef struct threadpool {
pthread_mutex_t lock; /* 保护任务队列的互斥锁 */
pthread_cond_t not_empty; /* 队列非空条件变量 */
pthread_t *threads; /* 工作线程数组 */
task_t *head; /* 任务队列头 */
task_t *tail; /* 任务队列尾 */
int thread_count;
int shutdown; /* 1 = 停止 */
} threadpool_t;
/* 工作线程主函数 */
static void *worker(void *arg) {
threadpool_t *pool = (threadpool_t *)arg;
for (;;) {
pthread_mutex_lock(&pool->lock);
/* 队列空且未关闭 -> 挂起等待条件变量 */
while (pool->head == NULL && !pool->shutdown) {
pthread_cond_wait(&pool->not_empty, &pool->lock);
}
/* 关闭且队列已清空 -> 退出线程 */
if (pool->shutdown && pool->head == NULL) {
pthread_mutex_unlock(&pool->lock);
pthread_exit(NULL);
}
/* 取出队头任务（在锁内出队，锁外执行） */
task_t *t = pool->head;
pool->head = t->next;
if (pool->head == NULL) pool->tail = NULL;
pthread_mutex_unlock(&pool->lock);
t->func(t->arg); /* 执行任务 */
free(t); /* 释放任务节点 */
}
return NULL;
}
/* 创建线程池 */
threadpool_t *threadpool_create(int thread_count) {
threadpool_t *pool = calloc(1, sizeof(*pool));
pool->thread_count = thread_count;
pthread_mutex_init(&pool->lock, NULL);
pthread_cond_init(&pool->not_empty, 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;
}
/* 提交任务 */
int threadpool_add(threadpool_t *pool, void (*func)(void *), void *arg) {
task_t *t = malloc(sizeof(*t));
t->func = func;
t->arg = arg;
t->next = NULL;
pthread_mutex_lock(&pool->lock);
if (pool->tail) {
pool->tail->next = t; /* 追加到队尾 */
} else {
pool->head = t; /* 队列为空，作为第一个任务 */
}
pool->tail = t;
pthread_cond_signal(&pool->not_empty); /* 唤醒一个空闲工作线程 */
pthread_mutex_unlock(&pool->lock);
return 0;
}
/* 销毁线程池：先置关闭标志，再唤醒所有线程，等它们处理完剩余任务后退出 */
void threadpool_destroy(threadpool_t *pool) {
pthread_mutex_lock(&pool->lock);
pool->shutdown = 1;
pthread_mutex_unlock(&pool->lock);
pthread_cond_broadcast(&pool->not_empty); /* 唤醒所有等待中的线程 */
for (int i = 0; i < pool->thread_count; i++) {
pthread_join(pool->threads[i], NULL); /* 回收工作线程 */
}
/* 清理可能残留的任务节点（正常情况下应为空） */
task_t *t;
while (pool->head) {
t = pool->head;
pool->head = t->next;
free(t);
}
free(pool->threads);
pthread_mutex_destroy(&pool->lock);
pthread_cond_destroy(&pool->not_empty);
free(pool);
}
/* ---------------- 测试 ---------------- */
typedef struct {
int id;
} job_t;
void my_task(void *arg) {
job_t *j = (job_t *)arg;
printf("task %2d 由线程 %lu 执行\n", j->id, (unsigned long)pthread_self());
usleep(100000); /* 模拟耗时 */
}
int main(void) {
threadpool_t *pool = threadpool_create(4); /* 4 个工作线程 */
job_t jobs[20];
for (int i = 0; i < 20; i++) {
jobs[i].id = i;
threadpool_add(pool, my_task, &jobs[i]);
}
sleep(2); /* 等待任务执行完 */
threadpool_destroy(pool); /* 优雅关闭 */
printf("线程池已关闭\n");
return 0;
}
4.3 代码讲解
-
worker()：工作线程的主体。加锁后先检查队列：空且未关闭就
pthread_cond_wait睡眠；关闭且队列清空就退出；否则出队。出队后立刻解锁，在锁外执行任务，这是避免慢任务阻塞整个线程池的关键。 -
threadpool_create()：初始化锁和条件变量，一次性创建 N 个线程。线程创建后立即进入 worker 循环等待任务。
-
threadpool_add()：加锁 → 尾插链表 →
signal唤醒一个线程 → 解锁。signal只唤醒一个，正好匹配「提交一个任务只需要一个线程来处理」。 -
threadpool_destroy()：置
shutdown = 1后broadcast唤醒所有线程。worker 醒来发现「关闭且队列已空」就退出，join逐个回收。这样能保证已提交的任务全部执行完，不会丢任务。
4.4 编译运行
gcc -o threadpool threadpool.c -pthread -std=c11 ./threadpool
可能的输出（线程号因系统而异，20 个任务由 4 个线程并发消化）：
task 0 由线程 140251943358528 执行 task 1 由线程 140251934965824 执行 task 2 由线程 140251926573120 执行 task 3 由线程 140251918180416 执行 task 4 由线程 140251934965824 执行 ... task 19 由线程 140251918180416 执行 线程池已关闭
5. C++11 std::thread 线程池实现
5.1 设计思路
-
用
std::function<void()>抹平任务类型，任意可调用对象都能入队； -
用
std::packaged_task+std::future，让任务可以取回返回值； -
用 RAII：析构函数里自动关闭并
join，杜绝忘记回收线程； -
enqueue模板化，支持任意参数个数和类型的任务。
5.2 完整代码
/* C++11 线程池实现：std::thread + std::packaged_task + std::future */
#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <stdexcept>
class ThreadPool {
public:
explicit ThreadPool(size_t threads)
: stop_(false) {
for (size_t i = 0; i < threads; ++i) {
workers_.emplace_back([this] {
for (;;) {
std::function<void()> task;
{
/* 条件变量必须在持锁状态下等待 */
std::unique_lock<std::mutex> lock(queue_mutex_);
/* 用 while 防止虚假唤醒 */
cv_.wait(lock, [this] {
return stop_ || !tasks_.empty();
});
if (stop_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task(); /* 锁外执行任务 */
}
});
}
}
/* 提交任务，返回 std::future 用于获取结果 */
template <class F, class... Args>
auto enqueue(F&& f, Args&&... args)
-> std::future<typename std::result_of<F(Args...)>::type> {
using return_type = typename std::result_of<F(Args...)>::type;
/* packaged_task 包装可调用对象，通过 future 取返回值 */
auto task = std::make_shared<std::packaged_task<return_type()>>(
std::bind(std::forward<F>(f), std::forward<Args>(args)...));
std::future<return_type> res = task->get_future();
{
std::unique_lock<std::mutex> lock(queue_mutex_);
if (stop_) throw std::runtime_error("enqueue on stopped ThreadPool");
/* 再包一层 lambda，统一抹平为无参无返回 */
tasks_.emplace([task]() { (*task)(); });
}
cv_.notify_one(); /* 唤醒一个空闲工作线程 */
return res;
}
/* 析构：置停止标志，唤醒全部线程并回收 */
~ThreadPool() {
{
std::unique_lock<std::mutex> lock(queue_mutex_);
stop_ = true;
}
cv_.notify_all();
for (std::thread &worker : workers_) {
worker.join();
}
}
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(const ThreadPool&) = delete;
private:
std::vector<std::thread> workers_; /* 工作线程 */
std::queue<std::function<void()>> tasks_; /* 任务队列 */
std::mutex queue_mutex_; /* 队列互斥锁 */
std::condition_variable cv_; /* 条件变量 */
bool stop_; /* 停止标志 */
};
int main() {
ThreadPool pool(4); /* 4 个工作线程 */
/* 有返回值的任务：用 future 拿结果 */
auto f1 = pool.enqueue([](int a, int b) { return a + b; }, 3, 4);
auto f2 = pool.enqueue([](int a) { return a * a; }, 9);
/* 无返回值的任务 */
for (int i = 0; i < 10; ++i) {
pool.enqueue([i] {
std::cout << "task " << i << " in thread "
<< std::this_thread::get_id() << std::endl;
std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
}
std::cout << "3 + 4 = " << f1.get() << std::endl;
std::cout << "9 * 9 = " << f2.get() << std::endl;
/* 离开作用域时自动析构，优雅关闭 */
return 0;
}
5.3 代码讲解
-
worker 循环：和 C 版本质相同——持锁等待
stop_ || !tasks_.empty()，条件满足后取任务、锁外执行。 -
enqueue模板：F是任意可调用对象，Args...是参数包。先用packaged_task<return_type()>绑定参数，再用 lambda 包成std::function<void()>入队，同时返回std::future给调用方取结果。 -
RAII 析构：置
stop_ = true→notify_all()→ 逐个join()。哪怕调用方忘了手动关闭，对象销毁时也会自动优雅关闭。 -
注意：
std::result_of是 C++11 的写法（C++17 起可用std::invoke_result替代，语义相同）。C++ 标准库线程是跨平台的，同样的代码在 Linux / Windows / RTOS 上都能编译。
5.4 编译运行
g++ -o threadpool threadpool.cpp -std=c++11 -pthread ./threadpool
输出示例（多线程并发打印，顺序交错是正常现象）：
task 0 in thread 2 task 1 in thread 3 3 + 4 = 7 9 * 9 = 81 task 2 in thread 4 ... task 9 in thread 2
生产环境中多线程写日志/控制台要加日志锁，否则输出会互相穿插。
6. 任务流转：条件变量如何驱动线程池
图 3 生产者和工作线程通过「互斥锁 + 条件变量」协作。
-
生产者：加锁 → 任务入队 →
signal/notify_one唤醒一个线程 → 解锁； -
工作线程：加锁 → 条件不满足则
wait（释放锁并睡眠）→ 被唤醒后重新拿锁 → 出队 → 解锁 → 锁外执行； -
notify_one只唤醒一个线程，适合「每提交一个任务就够一个线程处理」；notify_all/broadcast唤醒所有线程，适合关闭线程池的场景。
7. 线程池面试题深入解析
Q1：什么是线程池？为什么用线程池？预先创建一组线程并复用，避免频繁创建 / 销毁线程的开销，同时限制最大并发数，防止系统资源被打满。
Q2：线程池有哪些核心参数？
-
核心线程数：常驻线程数量；
-
最大线程数：任务暴增时最多扩到多少；
-
任务队列容量：有界还是无界；
-
拒绝策略：队列满时怎么办（丢弃、阻塞、抛异常）。
Q3：线程数怎么定？
-
CPU 密集型：≈ CPU 核数，再多只会增加上下文切换开销；
-
IO 密集型：≈ 核数 × 2 或更多，线程大部分时间在等 IO、不占 CPU；
-
最终结合压测调优，没有银弹。
Q4：怎么优雅关闭线程池？置停止标志 → 唤醒所有线程 → 等待队列中已提交的任务执行完 → 线程退出 → join 回收。不要直接 pthread_cancel 强杀线程，可能造成资源泄漏或数据不一致。
Q5：有界队列和无界队列的区别？无界队列实现简单，但任务堆积会吃光内存（触发 OOM）；有界队列配合拒绝策略更可控，生产环境推荐。
Q6：线程池和进程池的区别？线程池共享地址空间、切换快，但一个线程崩溃可能影响整个进程；进程池隔离好，但创建 / 切换开销大、通信要走 IPC。嵌入式里常用「多进程 + 进程内多线程」的混合架构。
8. 两种实现对比
| 对比项 | pthread 线程池（C） | std::thread 线程池（C++11） |
|---|---|---|
| 语言 | C | C++11 |
| 任务表示 | 函数指针 + void * | std::function + 模板 |
| 获取返回值 | 不支持，需自行包装 | std::future 直接支持 |
| 资源管理 | 手动 create / destroy / join | RAII，析构自动回收 |
| 可移植性 | POSIX 平台（Linux / Unix） | 标准库，跨平台 |
| 适用场景 | 嵌入式 C 工程、系统编程 | C++ 工程、通用服务 |
9. 总结
-
线程池 = 任务队列 + 工作线程 + 互斥锁 + 条件变量 + 停止标志；
-
三个关键细节：锁外执行任务、while 防虚假唤醒、置标志优雅退出；
-
C 版适合嵌入式 C 工程，C++ 版利用标准库更省心（future 拿结果、RAII 自动清理）；
-
面试重点：核心参数、线程数估算、关闭流程、有界 / 无界队列。
下一篇可以写：线程池进阶（动态扩容、任务优先级、超时取消）或进程池实现。欢迎评论区点菜 😄
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2301_80672443/article/details/163513694
