本文转载自微信公众号「程序喵大人」,作者程序喵大人 。转载本文请联系程序喵大人公众号。
之前分享过一次手写线程池 - C语言版,然后有朋友问是否有C++线程池实现的文章:
其实关于C++线程池的文章我好久以前写过,但估计很多新朋友都没有看到过,这里也重新发一下!
本人在开发过程中经常会遇到需要使用线程池的需求,但查了一圈发现在C++中完备的线程池第三方库还是比较少的,于是打算自己搞一个,链接地址文章最后附上,目前还只是初版,可能还有很多问题,望各位指正。
线程池都需要什么功能?
个人认为线程池需要支持以下几个基本功能:
- 核心线程数(core_threads):线程池中拥有的最少线程个数,初始化时就会创建好的线程,常驻于线程池。
- 最大线程个数(max_threads):线程池中拥有的最大线程个数,max_threads>=core_threads,当任务的个数太多线程池执行不过来时,内部就会创建更多的线程用于执行更多的任务,内部线程数不会超过max_threads,多创建出来的线程在一段时间内没有执行任务则会自动被回收掉,最终线程个数保持在核心线程数。
- 超时时间(time_out):如上所述,多创建出来的线程在time_out时间内没有执行任务就会被回收。
- 可获取当前线程池中线程的总个数。
- 可获取当前线程池中空闲线程的个数。
- 开启线程池功能的开关。
- 关闭线程池功能的开关,可以选择是否立即关闭,立即关闭线程池时,当前线程池里缓存的任务不会被执行。
如何实现线程池?下面是自己实现的线程池逻辑。
线程池中主要的数据结构
链表或者数组:用于存储线程池中的线程。
队列:用于存储需要放入线程池中执行的任务。
条件变量:当有任务需要执行时,用于通知正在等待的线程从任务队列中取出任务执行。
代码如下:
- class ThreadPool {
- public:
- using PoolSeconds = std::chrono::seconds;
-
-
- struct ThreadPoolConfig {
- int core_threads;
- int max_threads;
- int max_task_size;
- PoolSeconds time_out;
- };
-
-
- enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 };
-
-
- enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 };
-
- using ThreadPtr = std::shared_ptr
; - using ThreadId = std::atomic<int>;
- using ThreadStateAtomic = std::atomic
; - using ThreadFlagAtomic = std::atomic
; -
-
- struct ThreadWrapper {
- ThreadPtr ptr;
- ThreadId id;
- ThreadFlagAtomic flag;
- ThreadStateAtomic state;
-
- ThreadWrapper() {
- ptr = nullptr;
- id = 0;
- state.store(ThreadState::kInit);
- }
- };
- using ThreadWrapperPtr = std::shared_ptr
; - using ThreadPoolLock = std::unique_lock
; -
- private:
- ThreadPoolConfig config_;
-
- std::list
worker_threads_; -
- std::queue
function > tasks_; - std::mutex task_mutex_;
- std::condition_variable task_cv_;
-
- std::atomic<int> total_function_num_;
- std::atomic<int> waiting_thread_num_;
- std::atomic<int> thread_id_; // 用于为新创建的线程分配ID
-
- std::atomic
is_shutdown_now_; - std::atomic
is_shutdown_; - std::atomic
is_available_; - };
线程池的初始化
在构造函数中将各个成员变量都附初值,同时判断线程池的config是否合法。
- ThreadPool(ThreadPoolConfig config) : config_(config) {
- this->total_function_num_.store(0);
- this->waiting_thread_num_.store(0);
-
- this->thread_id_.store(0);
- this->is_shutdown_.store(false);
- this->is_shutdown_now_.store(false);
-
- if (IsValidConfig(config_)) {
- is_available_.store(true);
- } else {
- 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;
- }
如何开启线程池功能?
创建核心线程数个线程,常驻于线程池,等待任务的执行,线程ID由GetNextThreadId()统一分配。
- // 开启线程池功能
- bool Start() {
- if (!IsAvailable()) {
- return false;
- }
- int core_thread_num = config_.core_threads;
- cout << "Init thread num " << core_thread_num << endl;
- while (core_thread_num-- > 0) {
- AddThread(GetNextThreadId());
- }
- cout << "Init thread end" << endl;
- return true;
- }
如何关闭线程?
这里有两个标志位,is_shutdown_now置为true表示立即关闭线程,is_shutdown置为true则表示先执行完队列里的任务再关闭线程池。
- // 关掉线程池,内部还没有执行的任务会继续执行
- void ShutDown() {
- ShutDown(false);
- cout << "shutdown" << endl;
- }
-
- // 执行关掉线程池,内部还没有执行的任务直接取消,不会再执行
- void ShutDownNow() {
- ShutDown(true);
- cout << "shutdown now" << endl;
- }
-
- // private
- void ShutDown(bool is_now) {
- if (is_available_.load()) {
- if (is_now) {
- this->is_shutdown_now_.store(true);
- } else {
- this->is_shutdown_.store(true);
- }
- this->task_cv_.notify_all();
- is_available_.store(false);
- }
- }
如何为线程池添加线程?
见AddThread()函数,默认会创建Core线程,也可以选择创建Cache线程,线程内部会有一个死循环,不停的等待任务,有任务到来时就会执行,同时内部会判断是否是Cache线程,如果是Cache线程,timeout时间内没有任务执行就会自动退出循环,线程结束。
这里还会检查is_shutdown和is_shutdown_now标志,根据两个标志位是否为true来判断是否结束线程。
- void AddThread(int id) { AddThread(id, ThreadFlag::kCore); }
-
- void AddThread(int id, ThreadFlag thread_flag) {
- cout << "AddThread " << id << " flag " << static_cast<int>(thread_flag) << endl;
- ThreadWrapperPtr thread_ptr = std::make_shared
(); - thread_ptr->id.store(id);
- thread_ptr->flag.store(thread_flag);
- auto func = [this, thread_ptr]() {
- for (;;) {
- std::function
task; - {
- ThreadPoolLock lock(this->task_mutex_);
- if (thread_ptr->state.load() == ThreadState::kStop) {
- break;
- }
- cout << "thread id " << thread_ptr->id.load() << " running start" << endl;
- thread_ptr->state.store(ThreadState::kWaiting);
- ++this->waiting_thread_num_;
- bool is_timeout = false;
- if (thread_ptr->flag.load() == ThreadFlag::kCore) {
- this->task_cv_.wait(lock, [this, thread_ptr] {
- return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
- thread_ptr->state.load() == ThreadState::kStop);
- });
- } else {
- this->task_cv_.wait_for(lock, this->config_.time_out, [this, thread_ptr] {
- return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
- thread_ptr->state.load() == ThreadState::kStop);
- });
- is_timeout = !(this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() ||
- thread_ptr->state.load() == ThreadState::kStop);
- }
- --this->waiting_thread_num_;
- cout << "thread id " << thread_ptr->id.load() << " running wait end" << endl;
-
- if (is_timeout) {
- thread_ptr->state.store(ThreadState::kStop);
- }
-
- if (thread_ptr->state.load() == ThreadState::kStop) {
- cout << "thread id " << thread_ptr->id.load() << " state stop" << endl;
- break;
- }
- if (this->is_shutdown_ && this->tasks_.empty()) {
- cout << "thread id " << thread_ptr->id.load() << " shutdown" << endl;
- break;
- }
- if (this->is_shutdown_now_) {
- cout << "thread id " << thread_ptr->id.load() << " shutdown now" << endl;
- break;
- }
- thread_ptr->state.store(ThreadState::kRunning);
- task = std::move(this->tasks_.front());
- this->tasks_.pop();
- }
- task();
- }
- cout << "thread id " << thread_ptr->id.load() << " running end" << endl;
- };
- thread_ptr->ptr = std::make_shared
(std::move(func)); - if (thread_ptr->ptr->joinable()) {
- thread_ptr->ptr->detach();
- }
- this->worker_threads_.emplace_back(std::move(thread_ptr));
- }
如何将任务放入线程池中执行?
见如下代码,将任务使用std::bind封装成std::function放入任务队列中,任务较多时内部还会判断是否有空闲线程,如果没有空闲线程,会自动创建出最多(max_threads-core_threads)个Cache线程用于执行任务。
- // 放在线程池中执行函数
- template
- auto Run(F &&f, Args &&... args) -> std::shared_ptr
>> { - if (this->is_shutdown_.load() || this->is_shutdown_now_.load() || !IsAvailable()) {
- return nullptr;
- }
- if (GetWaitingThreadSize() == 0 && GetTotalThreadSize() < config_.max_threads) {
- AddThread(GetNextThreadId(), ThreadFlag::kCache);
- }
-
- using return_type = std::result_of_t
; - auto task = std::make_shared
>( - std::bind(std::forward
(f), std:: forward(args)...)); - total_function_num_++;
-
- std::future
res = task->get_future(); - {
- ThreadPoolLock lock(this->task_mutex_);
- this->tasks_.emplace([task]() { (*task)(); });
- }
- this->task_cv_.notify_one();
- return std::make_shared
>>(std:: move(res)); - }
如何获取当前线程池中线程的总个数?
- int GetTotalThreadSize() { return this->worker_threads_.size(); }
如何获取当前线程池中空闲线程的个数?
waiting_thread_num值表示空闲线程的个数,该变量在线程循环内部会更新。
- int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }
简单的测试代码
- int main() {
- cout << "hello" << endl;
- ThreadPool pool(ThreadPool::ThreadPoolConfig{4, 5, 6, std::chrono::seconds(4)});
- pool.Start();
- std::this_thread::sleep_for(std::chrono::seconds(4));
- cout << "thread size " << pool.GetTotalThreadSize() << endl;
- std::atomic<int> index;
- index.store(0);
- std::thread t([&]() {
- for (int i = 0; i < 10; ++i) {
- pool.Run([&]() {
- cout << "function " << index.load() << endl;
- std::this_thread::sleep_for(std::chrono::seconds(4));
- index++;
- });
- // std::this_thread::sleep_for(std::chrono::seconds(2));
- }
- });
- t.detach();
- cout << "=================" << endl;
-
- std::this_thread::sleep_for(std::chrono::seconds(4));
- pool.Reset(ThreadPool::ThreadPoolConfig{4, 4, 6, std::chrono::seconds(4)});
- std::this_thread::sleep_for(std::chrono::seconds(4));
- cout << "thread size " << pool.GetTotalThreadSize() << endl;
- cout << "waiting size " << pool.GetWaitingThreadSize() << endl;
- cout << "---------------" << endl;
- pool.ShutDownNow();
- getchar();
- cout << "world" << endl;
- return 0;
- }