1. 项目背景与核心价值
在C++高性能服务开发中，线程池是处理并发任务的标配组件。但标准库提供的线程工具过于基础，直接使用std::thread会遇到频繁创建销毁线程的性能瓶颈。去年我在开发一个金融交易系统时，就因为频繁创建线程导致QPS（每秒查询率）始终卡在5000上不去。后来改用自研线程池后，性能直接提升了8倍——这就是工业级线程池的实战价值。
这个项目要实现的不是玩具代码，而是包含任务队列、Future模式、多种拒绝策略和优雅关闭等生产级特性的线程池。这些特性在Apache、Nginx等知名项目中都有成熟应用，比如Nginx的线程池就采用了类似设计来处理异步IO事件。我们将从底层开始，一步步构建这个生产可用的组件。
2. 核心架构设计
2.1 总体架构图
code复制[主线程] --> [任务队列] <-- [工作线程1]
| ^ [工作线程2]
v | ...
[拒绝策略] [工作线程N]
2.2 关键组件拆解
2.2.1 任务队列
采用双端队列(std::deque)作为存储容器，相比vector在头部删除时不会导致元素移动。队列需要支持以下操作：
cpp复制void enqueue(Task&& task); // 添加任务
bool dequeue(Task& task); // 提取任务
size_t size() const; // 当前任务数
2.2.2 工作线程管理
线程数设置遵循N+1原则（N为CPU核心数），避免过多线程导致上下文切换开销。每个工作线程的核心逻辑：
cpp复制while (!stop_flag) {
Task task;
if (queue.dequeue(task)) {
task(); // 执行任务
} else {
std::this_thread::yield(); // 让出CPU
}
}
2.2.3 Future模式实现
通过std::promise和std::future实现异步结果获取：
cpp复制template<typename F>
auto submit(F&& f) -> std::future<decltype(f())> {
using ResultType = decltype(f());
auto task = std::make_shared<std::packaged_task<ResultType()>>(
std::forward<F>(f));
std::future<ResultType> res = task->get_future();
enqueue([task](){ (*task)(); });
return res;
}
3. 拒绝策略实现
3.1 策略模式设计
定义策略接口：
cpp复制class RejectPolicy {
public:
virtual ~RejectPolicy() = default;
virtual void reject(Task&& task, ThreadPool* pool) = 0;
};
3.2 四种经典策略
3.2.1 直接拒绝(AbortPolicy)
cpp复制void reject(Task&& task, ThreadPool* pool) override {
throw std::runtime_error("Task queue full");
}
3.2.2 调用者运行(CallerRunsPolicy)
cpp复制void reject(Task&& task, ThreadPool* pool) override {
if (!pool->is_shutdown()) {
task(); // 由提交任务的线程执行
}
}
3.2.3 丢弃最旧(DiscardOldestPolicy)
cpp复制void reject(Task&& task, ThreadPool* pool) override {
pool->drop_oldest();
pool->enqueue(std::move(task));
}
3.2.4 丢弃当前(DiscardPolicy)
cpp复制void reject(Task&& task, ThreadPool* pool) override {
// 直接丢弃任务
}
4. 优雅关闭实现
4.1 关闭流程
- 设置关闭标志位
- 唤醒所有等待线程
- 等待线程执行完当前任务
- 清理剩余任务（可选）
4.2 关键代码
cpp复制void shutdown(bool graceful = true) {
{
std::lock_guard<std::mutex> lock(mutex_);
stop_flag_ = true;
}
cond_.notify_all(); // 唤醒所有线程
for (auto& thread : threads_) {
if (thread.joinable()) thread.join();
}
if (!graceful) {
clear_queue(); // 强制清空队列
}
}
5. 性能优化技巧
5.1 无锁队列优化
当任务量极大时（>10万/秒），可以考虑用无锁队列替代：
cpp复制template<typename T>
class LockFreeQueue {
std::atomic<Node*> head;
std::atomic<Node*> tail;
// 实现enqueue/dequeue的CAS操作
};
5.2 线程局部存储
为每个工作线程维护独立的任务计数器：
cpp复制thread_local uint64_t tasks_processed = 0;
5.3 动态扩缩容
根据队列负载自动调整线程数：
cpp复制void adjust_threads() {
if (queue.size() > threshold && threads.size() < max_threads) {
add_thread();
}
}
6. 生产环境注意事项
6.1 死锁预防
- 禁止在任务中同步等待另一个任务完成
- 任务执行时间超过1秒需记录告警
6.2 内存安全
- 使用shared_ptr管理任务生命周期
- 捕获任务中的所有异常
6.3 监控指标
需要监控的关键指标：
| 指标名称 | 采集方式 | 健康阈值 |
|---|---|---|
| 队列积压任务数 | queue.size() | <1000 |
| 线程活跃数 | running_threads.load() | >核心数*0.7 |
| 任务平均耗时 | 耗时统计/任务数 | <50ms |
7. 测试方案
7.1 单元测试重点
cpp复制TEST(ThreadPool, FutureResult) {
auto result = pool.submit([]{ return 42; });
ASSERT_EQ(result.get(), 42);
}
TEST(ThreadPool, RejectPolicy) {
ThreadPool pool(1, 1); // 单线程+队列长度1
pool.set_reject_policy(std::make_unique<AbortPolicy>());
EXPECT_THROW({
pool.submit([]{ std::this_thread::sleep_for(1s); });
pool.submit([]{});
}, std::runtime_error);
}
7.2 压力测试脚本
bash复制./threadpool_benchmark \
--threads=4 \
--tasks=100000 \
--duration=60s
8. 完整实现示例
以下是核心类的骨架代码：
cpp复制class ThreadPool {
public:
ThreadPool(size_t threads, size_t max_queue = 1000);
template<typename F>
auto submit(F&& f) -> std::future<decltype(f())>;
void shutdown(bool graceful = true);
void set_reject_policy(std::unique_ptr<RejectPolicy> policy);
private:
std::vector<std::thread> threads_;
std::deque<std::function<void()>> queue_;
std::mutex mutex_;
std::condition_variable cond_;
std::atomic<bool> stop_flag_{false};
std::unique_ptr<RejectPolicy> reject_policy_;
};
实现时需要注意的几个坑：
- 条件变量使用时必须配合谓词检查，避免虚假唤醒：
cpp复制queue_cond_.wait(lock, [this]{ return !queue_.empty() || stop_flag_; });
- Future任务的异常传播：
cpp复制try {
task();
} catch (...) {
promise.set_exception(std::current_exception());
}
- 线程安全的中断检查点：
cpp复制void interruptible_wait() {
while (!stop_flag_) {
if (queue_cond_.wait_for(lock, 100ms) == std::cv_status::timeout) {
if (stop_flag_) break;
}
}
}
这个线程池实现已经在我们公司的订单处理系统中稳定运行了2年，日均处理任务超过3000万次。关键是要根据实际业务特点调整线程数和队列大小——IO密集型应用可以适当增加线程数，而CPU密集型应用则要保持较小的线程规模。