C++ 线程池
线程池都需要什么功能？
线程池需要支持以下几个基本功能：

  • 核心线程数(core_threads)：线程池中拥有的最少线程个数，初始化时就会创建好的线程，常驻于线程池。
  • 最大线程个数(max_threads)：线程池中拥有的最大线程个数，max_threads>=core_threads，当任务的个数太多线程池执行不过来时，内部就会创建更多的线程用于执行更多的任务，内部线程数不会超过max_threads，多创建出来的线程在一段时间内没有执行任务则会自动被回收掉，最终线程个数保持在核心线程数。
  • 超时时间(time_out)：如上所述，多创建出来的线程在time_out时间内没有执行任务就会被回收。
  • 可获取当前线程池中线程的总个数。
  • 可获取当前线程池中空闲线程的个数。
  • 开启线程池功能的开关。

  • 关闭线程池功能的开关，可以选择是否立即关闭，立即关闭线程池时，当前线程池里缓存的任务不会被执行。

    如何实现线程池？下面是自己实现的线程池逻辑。

    线程池中主要的数据结构
    1. 链表或者数组：用于存储线程池中的线程。
    2. 队列：用于存储需要放入线程池中执行的任务。
    3. 条件变量：当有任务需要执行时，用于通知正在等待的线程从任务队列中取出任务执行。
    代码如下：

    1. class ThreadPool {
    2.   public:
    3.   using PoolSeconds = std::chrono::seconds;
    5.   /** 线程池的配置
    6.  * core_threads: 核心线程个数，线程池中最少拥有的线程个数，初始化就会创建好的线程，常驻于线程池
    7.  *
    8.  * max_threads: >=core_threads，当任务的个数太多线程池执行不过来时，
    9.  * 内部就会创建更多的线程用于执行更多的任务，内部线程数不会超过max_threads
    10.  *
    11.  * max_task_size: 内部允许存储的最大任务个数，暂时没有使用
    12.  *
    13.  * time_out: Cache线程的超时时间，Cache线程指的是max_threads-core_threads的线程,
    14.  * 当time_out时间内没有执行任务，此线程就会被自动回收
    15.  */
    16.   struct ThreadPoolConfig {
    17.       int core_threads;
    18.       int max_threads;
    19.       int max_task_size;
    20.       PoolSeconds time_out;
    21.   };
    23.   /**
    24.  * 线程的状态：有等待、运行、停止
    25.  */
    26.   enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 };
    28.   /**
    29.  * 线程的种类标识：标志该线程是核心线程还是Cache线程，Cache是内部为了执行更多任务临时创建出来的
    30.  */
    31.   enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 };
    33.   using ThreadPtr = std::shared_ptr<std::thread>;
    34.   using ThreadId = std::atomic<int>;
    35.   using ThreadStateAtomic = std::atomic<ThreadState>;
    36.   using ThreadFlagAtomic = std::atomic<ThreadFlag>;
    38.   /**
    39.  * 线程池中线程存在的基本单位，每个线程都有个自定义的ID，有线程种类标识和状态
    40.  */
    41.   struct ThreadWrapper {
    42.       ThreadPtr ptr;
    43.       ThreadId id;
    44.       ThreadFlagAtomic flag;
    45.       ThreadStateAtomic state;
    47.       ThreadWrapper() {
    48.           ptr = nullptr;
    49.           id = 0;
    50.           state.store(ThreadState::kInit);
    51.       }
    52.   };
    53.   using ThreadWrapperPtr = std::shared_ptr<ThreadWrapper>;
    54.   using ThreadPoolLock = std::unique_lock<std::mutex>;
    56.   private:
    57.   ThreadPoolConfig config_;
    59.   std::list<ThreadWrapperPtr> worker_threads_;
    61.   std::queue<std::function<void()>> tasks_;
    62.   std::mutex task_mutex_;
    63.   std::condition_variable task_cv_;
    65.   std::atomic<int> total_function_num_;
    66.   std::atomic<int> waiting_thread_num_;
    67.   std::atomic<int> thread_id_; // 用于为新创建的线程分配ID
    69.   std::atomic<bool> is_shutdown_now_;
    70.   std::atomic<bool> is_shutdown_;
    71.   std::atomic<bool> is_available_;
    72. };

    线程池的初始化

    在构造函数中将各个成员变量都附初值，同时判断线程池的config是否合法。 ```cpp ThreadPool(ThreadPoolConfig config) : config(config) { this->total_function_num.store(0); this->waitingthread_num.store(0);

    this->threadid.store(0); this->isshutdown.store(false); this->isshutdown_now.store(false);

    if (IsValidConfig(config_)) {

    1.   is_available_.store(true);

    } else {

    1.   is_available_.store(false);

    } }

bool IsValidConfig(ThreadPoolConfig config) { if (config.core_threads < 1 || config.max_threads < config.core_threads || config.time_out.count() < 1) { return false; } return true; }

  1. <a name="PdCw5"></a>
  2. ### 开启线程池功能
  3. 创建核心线程数个线程，常驻于线程池，等待任务的执行，线程ID由GetNextThreadId()统一分配。
  4. ```cpp
  5. // 开启线程池功能
  6. bool Start() {
  7.     if (!IsAvailable()) {
  8.         return false;
  9.     }
  10.     int core_thread_num = config_.core_threads;
  11.     cout << "Init thread num " << core_thread_num << endl;
  12.     while (core_thread_num-- > 0) {
  13.         AddThread(GetNextThreadId());
  14.     }
  15.     cout << "Init thread end" << endl;
  16.     return true;
  17. }

关闭线程

这里有两个标志位，isshutdown_now置为true表示立即关闭线程，isshutdown置为true则表示先执行完队列里的任务再关闭线程池。

  1. // 关掉线程池，内部还没有执行的任务会继续执行
  2. void ShutDown() {
  3.     ShutDown(false);
  4.     cout << "shutdown" << endl;
  5. }
  7. // 执行关掉线程池，内部还没有执行的任务直接取消，不会再执行
  8. void ShutDownNow() {
  9.     ShutDown(true);
  10.     cout << "shutdown now" << endl;
  11. }
  13. // private
  14. void ShutDown(bool is_now) {
  15.     if (is_available_.load()) {
  16.         if (is_now) {
  17.             this->is_shutdown_now_.store(true);
  18.         } else {
  19.             this->is_shutdown_.store(true);
  20.         }
  21.         this->task_cv_.notify_all();
  22.         is_available_.store(false);
  23.     }
  24. }

为线程池添加线程

见AddThread()函数，默认会创建Core线程，也可以选择创建Cache线程，线程内部会有一个死循环，不停的等待任务，有任务到来时就会执行，同时内部会判断是否是Cache线程，如果是Cache线程，timeout时间内没有任务执行就会自动退出循环，线程结束。
这里还会检查is_shutdown和isshutdown_now标志，根据两个标志位是否为true来判断是否结束线程。

  1. void AddThread(int id) { AddThread(id, ThreadFlag::kCore); }
  3. void AddThread(int id, ThreadFlag thread_flag) {
  4.     cout << "AddThread " << id << " flag " << static_cast<int>(thread_flag) << endl;
  5.     ThreadWrapperPtr thread_ptr = std::make_shared<ThreadWrapper>();
  6.     thread_ptr->id.store(id);
  7.     thread_ptr->flag.store(thread_flag);
  8.     auto func = [this, thread_ptr]() {
  9.         for (;;) {
  10.             std::function<void()> task;
  11.             {
  12.                 ThreadPoolLock lock(this->task_mutex_);
  13.                 if (thread_ptr->state.load() == ThreadState::kStop) {
  14.                     break;
  15.                 }
  16.                 cout << "thread id " << thread_ptr->id.load() << " running start" << endl;
  17.                 thread_ptr->state.store(ThreadState::kWaiting);
  18.                 ++this->waiting_thread_num_;
  19.                 bool is_timeout = false;
  20.                 if (thread_ptr->flag.load() == ThreadFlag::kCore) {
  21.                     this->task_cv_.wait(lock, [this, thread_ptr] {
  22.                         return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
  23.                                 thread_ptr->state.load() == ThreadState::kStop);
  24.                     });
  25.                 } else {
  26.                     this->task_cv_.wait_for(lock, this->config_.time_out, [this, thread_ptr] {
  27.                         return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
  28.                                 thread_ptr->state.load() == ThreadState::kStop);
  29.                     });
  30.                     is_timeout = !(this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
  31.                                    thread_ptr->state.load() == ThreadState::kStop);
  32.                 }
  33.                 --this->waiting_thread_num_;
  34.                 cout << "thread id " << thread_ptr->id.load() << " running wait end" << endl;
  36.                 if (is_timeout) {
  37.                     thread_ptr->state.store(ThreadState::kStop);
  38.                 }
  40.                 if (thread_ptr->state.load() == ThreadState::kStop) {
  41.                     cout << "thread id " << thread_ptr->id.load() << " state stop" << endl;
  42.                     break;
  43.                 }
  44.                 if (this->is_shutdown_ && this->tasks_.empty()) {
  45.                     cout << "thread id " << thread_ptr->id.load() << " shutdown" << endl;
  46.                     break;
  47.                 }
  48.                 if (this->is_shutdown_now_) {
  49.                     cout << "thread id " << thread_ptr->id.load() << " shutdown now" << endl;
  50.                     break;
  51.                 }
  52.                 thread_ptr->state.store(ThreadState::kRunning);
  53.                 task = std::move(this->tasks_.front());
  54.                 this->tasks_.pop();
  55.             }
  56.             task();
  57.         }
  58.         cout << "thread id " << thread_ptr->id.load() << " running end" << endl;
  59.     };
  60.     thread_ptr->ptr = std::make_shared<std::thread>(std::move(func));
  61.     if (thread_ptr->ptr->joinable()) {
  62.         thread_ptr->ptr->detach();
  63.     }
  64.     this->worker_threads_.emplace_back(std::move(thread_ptr));
  65. }

将任务放入线程池中执行

见如下代码，将任务使用std::bind封装成std::function放入任务队列中，任务较多时内部还会判断是否有空闲线程，如果没有空闲线程，会自动创建出最多(max_threads-core_threads)个Cache线程用于执行任务。

  1. // 放在线程池中执行函数
  2. template <typename F, typename... Args>
  3. auto Run(F &&f, Args &&... args) -> std::shared_ptr<std::future<std::result_of_t<F(Args...)>>> {
  4.     if (this->is_shutdown_.load() || this->is_shutdown_now_.load() || !IsAvailable()) {
  5.         return nullptr;
  6.     }
  7.     if (GetWaitingThreadSize() == 0 && GetTotalThreadSize() < config_.max_threads) {
  8.         AddThread(GetNextThreadId(), ThreadFlag::kCache);
  9.     }
  11.     using return_type = std::result_of_t<F(Args...)>;
  12.     auto task = std::make_shared<std::packaged_task<return_type()>>(
  13.         std::bind(std::forward<F>(f), std::forward<Args>(args)...));
  14.     total_function_num_++;
  16.     std::future<return_type> res = task->get_future();
  17.     {
  18.         ThreadPoolLock lock(this->task_mutex_);
  19.         this->tasks_.emplace([task]() { (*task)(); });
  20.     }
  21.     this->task_cv_.notify_one();
  22.     return std::make_shared<std::future<std::result_of_t<F(Args...)>>>(std::move(res));
  23. }

获取当前线程池中线程的总个数

  1. int GetTotalThreadSize() { return this->worker_threads_.size(); }

获取当前线程池中空闲线程的个数

waitingthread_num值表示空闲线程的个数，该变量在线程循环内部会更新。

  1. int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }

简单的测试代码

  1. int main() {
  2.     cout << "hello" << endl;
  3.     ThreadPool pool(ThreadPool::ThreadPoolConfig{4, 5, 6, std::chrono::seconds(4)});
  4.     pool.Start();
  5.     std::this_thread::sleep_for(std::chrono::seconds(4));
  6.     cout << "thread size " << pool.GetTotalThreadSize() << endl;
  7.     std::atomic<int> index;
  8.     index.store(0);
  9.     std::thread t([&]() {
  10.         for (int i = 0; i < 10; ++i) {
  11.             pool.Run([&]() {
  12.                 cout << "function " << index.load() << endl;
  13.                 std::this_thread::sleep_for(std::chrono::seconds(4));
  14.                 index++;
  15.             });
  16.             // std::this_thread::sleep_for(std::chrono::seconds(2));
  17.         }
  18.     });
  19.     t.detach();
  20.     cout << "=================" << endl;
  22.     std::this_thread::sleep_for(std::chrono::seconds(4));
  23.     pool.Reset(ThreadPool::ThreadPoolConfig{4, 4, 6, std::chrono::seconds(4)});
  24.     std::this_thread::sleep_for(std::chrono::seconds(4));
  25.     cout << "thread size " << pool.GetTotalThreadSize() << endl;
  26.     cout << "waiting size " << pool.GetWaitingThreadSize() << endl;
  27.     cout << "---------------" << endl;
  28.     pool.ShutDownNow();
  29.     getchar();
  30.     cout << "world" << endl;
  31.     return 0;
  32. }