基于C11线程池模型 - 缓冲式线程池-拓冰建站

摘要：本文介绍缓冲式线程池（CachedThreadPool）的核心概念、实现原理和关键组件。重点阐述其动态线程管理机制、线程组的数据结构设计，并提供一个完整、可编译的线程安全任务队列（SyncQueue1）实现代码。

1. 缓冲式线程池概述

缓冲式线程池（CachedThreadPool）是一种动态调整线程数量的线程池实现。与固定大小的线程池（FixedThreadPool）不同，CachedThreadPool 的线程数量会根据任务负载动态变化：当有新任务提交时，如果线程池中有空闲线程，则会立即使用空闲线程执行任务；如果没有空闲线程，则会创建一个新的线程来执行任务。当线程空闲一段时间后（超过设定的空闲时间），该线程会被回收销毁，从而避免资源浪费。

2. 核心设计思路

2.1 动态线程管理

缓冲式线程池的核心优势在于其弹性伸缩能力。它适用于任务数量波动大、执行时间较短的场景。通过动态创建和回收线程，既能保证高并发时的处理能力，又能在低负载时释放系统资源。

2.2 线程组数据结构

采用unordered_map管理线程组，使用shared_ptr管理线程对象。

unordered_map 的优势：底层为无序哈希表，查找和删除操作的平均时间复杂度为 O(1)。对于动态可变的工作线程集合，需要根据线程 ID 迅速定位到指定线程进行销毁或管理，哈希表的高效查找特性非常适合此场景。
shared_ptr 管理线程：C++ 标准库中的std::thread不支持拷贝，只支持移动语义，且线程对象析构时会自动调用join()。使用shared_ptr可以实现多地方共享线程对象，统一生命周期管理，支持延迟销毁和安全异步回收，同时规避了移动语义带来的代码复杂性。

3. 任务队列实现

线程池需要一个线程安全的队列来存储待执行的任务。下面是一个完整的、可编译的线程安全同步队列SyncQueue1的实现，支持阻塞式插入和取出操作。

#ifndef SYNCQUEUE1_HPP #define SYNCQUEUE1_HPP #include <list> #include <mutex> #include <atomic> #include <condition_variable> template <typename T> class SyncQueue1 { private: std::list<T> queue_; mutable std::mutex mutex_; std::condition_variable m_not_empty_; std::condition_variable m_not_full_; size_t max_size_; size_t size_; size_t m_waittime_; std::atomic<bool> needstop_; bool isFull() const; bool isEmpty() const; template <typename F> int add(F&& x) { std::unique_lock<std::mutex> locker(mutex_); if (!m_not_full_.wait_for(locker, std::chrono::seconds(m_waittime_), [this] { return needstop_ || !isFull(); })) { return 1; // 超时 } if (needstop_) { return 2; // 已停止 } queue_.push_back(std::forward<F>(x)); size_++; m_not_empty_.notify_one(); return 0; // 成功 } public: SyncQueue1(size_t max_size) : max_size_(max_size), size_(0), needstop_(false) {} ~SyncQueue1() { if (!needstop_) { stop(); } } void put(const T& x) { add(x); } void put(T&& x) { add(std::move(x)); } void take(T& x) { std::unique_lock<std::mutex> lock(mutex_); m_not_empty_.wait(lock, [this] { return !isEmpty() || needstop_; }); if (needstop_) { return; } x = std::move(queue_.front()); queue_.pop_front(); --size_; m_not_full_.notify_one(); } void take(std::list<T>& q) { std::unique_lock<std::mutex> lock(mutex_); m_not_empty_.wait(lock, [this] { return !isEmpty() || needstop_; }); if (needstop_) { return; } q = std::move(queue_); size_ = 0; m_not_full_.notify_one(); } bool empty() const { return isEmpty(); } bool full() const { return isFull(); } void stop() { { std::unique_lock<std::mutex> lock(mutex_); needstop_ = true; } m_not_empty_.notify_all(); m_not_full_.notify_all(); } size_t size() const { std::unique_lock<std::mutex> lock(mutex_); return size_; } }; template <typename T> inline bool SyncQueue1<T>::isFull() const { std::unique_lock<std::mutex> lock(mutex_); return size_ >= max_size_; } template <typename T> inline bool SyncQueue1<T>::isEmpty() const { std::unique_lock<std::mutex> lock(mutex_); return size_ == 0; } #endif // SYNCQUEUE1_HPP

3.1 代码说明

SyncQueue1是一个模板类，提供以下核心功能：

线程安全操作：所有对内部队列的访问都通过互斥锁（std::mutex）保护。
条件变量同步：使用两个条件变量实现生产者-消费者模型：
- m_not_empty_：队列非空时通知消费者线程可以取数据。
- m_not_full_：队列未满时通知生产者线程可以放数据。
超时机制：add方法中的wait_for支持超时返回，避免无限等待。
优雅停止：通过needstop_原子标志位和stop()方法，可以安全地停止所有等待线程。
批量取出：take(std::list<T>&)方法可以一次性取出队列中所有元素，提高批量处理效率。

4. CacheThreadPool 的实现

在理解了缓冲式线程池的核心设计思路和任务队列实现后，本节将详细解析一个完整的CacheThreadPool实现。这个实现结合了前面介绍的动态线程管理机制和线程安全的任务队列，提供了一个可实际使用的线程池类。

4.1 类定义与成员变量

CacheThreadPool类的定义包含了以下核心成员变量：

#ifndef CACHETHREADPOOL_HPP #define CACHETHREADPOOL_HPP #include <mutex> #include <unordered_map> #include <functional> #include <future> #include <atomic> #include <chrono> #include "SyncQueue1.hpp" int MaxTaskCount = 2; const int KeepAliveTime = 10; using task = std::function<void(void)>; class CacheThreadPool { private: SyncQueue1<task> m_queue_; // 线程安全的任务队列 std::unordered_map<std::thread::id, std::shared_ptr<std::thread>> m_threadgroup; // 线程组管理 int m_coreThreadSize; // 核心线程数 int m_maxThreadSize; // 最大线程数 std::atomic_int m_idThreadSize; // 空闲线程数 std::atomic_int m_curThreadSize; // 当前线程总数 mutable std::mutex mutex_; // 保护线程组操作的互斥锁 std::atomic_bool m_running; // 线程池运行状态标志 std::once_flag m_flag; // 确保 Stop() 只执行一次

关键设计要点：

任务队列：使用前面实现的SyncQueue1<task>作为线程安全的任务存储容器。
线程组管理：采用unordered_map存储线程 ID 到线程对象的映射，便于快速查找和删除。
线程数量控制：通过m_coreThreadSize和m_maxThreadSize分别控制核心线程数和最大线程数。
原子计数器：使用原子类型确保线程计数的线程安全性。
once_flag：确保Stop()方法只被调用一次，避免重复释放资源。

4.2 构造函数与初始化

public: CacheThreadPool(int initNumThreads = 8, int taskPoolsize = MaxTaskCount) : m_coreThreadSize(initNumThreads), m_maxThreadSize(2 * std::thread::hardware_concurrency() + 1), m_idThreadSize(0), m_curThreadSize(0), m_queue_(taskPoolsize), m_running(false) { Start(m_coreThreadSize); }

构造函数特点：

默认参数：核心线程数默认为 8，任务队列大小默认为MaxTaskCount（2）。
最大线程数计算：m_maxThreadSize = 2 * std::thread::hardware_concurrency() + 1，基于 CPU 核心数动态计算。
初始化顺序：先初始化所有成员变量，再调用Start()方法启动核心线程。

4.3 线程启动与工作循环

private: void Start(int size) { m_running = true; m_curThreadSize = size; for (int i = 0; i < size; ++i) { auto tha = std::make_shared<std::thread>(std::thread(&CacheThreadPool::RunInThread, this)); std::thread::id tid = tha->get_id(); m_threadgroup.emplace(tid, std::move(tha)); m_idThreadSize++; } } void RunInThread() { auto tid = std::this_thread::get_id(); auto starttime = std::chrono::high_resolution_clock().now(); while (m_running) { task task; if (m_queue_.empty() && m_queue_.size() == 0) { auto now = std::chrono::high_resolution_clock().now(); auto intervalTime = std::chrono::duration_cast<std::chrono::seconds>(now - starttime); std::unique_lock<std::mutex> locker(mutex_); if (intervalTime.count() >= KeepAliveTime && m_curThreadSize > m_coreThreadSize) { m_threadgroup.find(tid)->second->detach(); m_threadgroup.erase(tid); m_curThreadSize--; m_idThreadSize++; return; } } if (!m_queue_.take(task) && m_running) { m_idThreadSize--; task(); m_idThreadSize++; starttime = std::chrono::high_resolution_clock().now(); } } }

工作线程的核心逻辑：

空闲检测：当任务队列为空时，线程进入空闲状态并开始计时。
超时回收：如果线程空闲时间超过KeepAliveTime（10秒）且当前线程数大于核心线程数，则回收该线程。
任务执行：从队列中取出任务并执行，执行完成后重置空闲计时器。
线程安全：线程回收操作需要加锁保护，避免并发修改线程组。

4.4 任务提交接口

public: template <class Func, class... Args> void execute(Func &&func, Args&& ...args) { auto task = std::make_shared<std::function<void()>>(std::bind(func, std::forward<Args>(args)...)); if (m_queue_.put([task]() { (*task)(); }) != 0) { (*task)(); } } template <class Func, class... Args> auto submit(Func&& func, Args&&... args) -> std::future<decltype(func(args...))> { using RecType = decltype(func(args...)); auto task = std::make_shared<std::packaged_task<RecType()>>( std::bind(std::forward<Func>(func), std::forward<Args>(args)...)); std::future<RecType> result = task->get_future(); if (m_queue_.put([task]() { (*task)(); }) != 0) { (*task)(); } if (m_idThreadSize <= 0 && m_curThreadSize < m_maxThreadSize) { std::lock_guard<std::mutex> loker(mutex_); auto tha = std::make_shared<std::thread>(std::thread(&CacheThreadPool::RunInThread, this)); std::thread::id tid = tha->get_id(); m_threadgroup.emplace(tid, std::move(tha)); m_idThreadSize++; m_curThreadSize++; } return result; }

两种任务提交方式：

execute()：提交无返回值的任务，使用std::function<void()>包装。
submit()：提交有返回值的任务，返回std::future对象，支持异步获取结果。

动态线程创建机制：在submit()方法中，当没有空闲线程且当前线程数未达到最大值时，会自动创建新线程。

4.5 停止机制

private: void StopThreadGroup() { m_queue_.stop(); m_running = false; for (auto& thread : m_threadgroup) { thread.second->join(); } m_threadgroup.clear(); } public: void Stop() { std::call_once(m_flag, [this] { StopThreadGroup(); }); }

优雅停止的实现：

停止任务队列：调用m_queue_.stop()通知所有等待线程。
设置运行标志：将m_running设为false，让工作线程退出循环。
等待线程结束：对所有线程调用join()等待其自然结束。
清理资源：清空线程组容器。
线程安全：使用std::call_once确保Stop()只被执行一次。

4.6 代码优化建议

当前实现有几个可以改进的地方：

任务队列超时处理：m_queue_.take(task)返回bool类型，但代码中使用了!m_queue_.take(task)的逻辑需要确认是否正确。
空闲线程计数：m_idThreadSize的增减操作需要确保线程安全，特别是在多线程环境下。
异常处理：任务执行过程中可能抛出异常，需要添加适当的异常处理机制。
资源泄漏防护：在析构函数中应确保调用Stop()方法。

这个CacheThreadPool实现展示了缓冲式线程池的核心思想：动态创建和回收线程，结合高效的任务队列，为短期异步任务提供高性能的处理能力。在实际使用中，可以根据具体需求调整核心线程数、最大线程数和空闲超时时间等参数。

使用示例：

#include "CacheThreadPool.hpp" #include <iostream> #include <chrono> int main() { CacheThreadPool pool(4); // 创建核心线程数为4的线程池 // 提交无返回值的任务 for (int i = 0; i < 20; ++i) { pool.execute([i]() { std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << std::endl; }); } // 提交有返回值的任务 auto future = pool.submit([]() -> int { std::this_thread::sleep_for(std::chrono::milliseconds(200)); return 42; }); // 获取任务结果 int result = future.get(); std::cout << "Task result: " << result << std::endl; // 等待所有任务完成 std::this_thread::sleep_for(std::chrono::seconds(2)); // 优雅停止线程池 pool.Stop(); return 0; }