線程池都需要什么功能？

個(gè)人認(rèn)為線程池需要支持以下幾個(gè)基本功能：

• 核心線程數(shù)(core_threads)：線程池中擁有的最少線程個(gè)數(shù)，初始化時(shí)就會(huì)創(chuàng)建好的線程，常駐于線程池。

• 最大線程個(gè)數(shù)(max_threads)：線程池中擁有的最大線程個(gè)數(shù)，max_threads>=core_threads，當(dāng)任務(wù)的個(gè)數(shù)太多線程池執(zhí)行不過來時(shí)，內(nèi)部就會(huì)創(chuàng)建更多的線程用于執(zhí)行更多的任務(wù)，內(nèi)部線程數(shù)不會(huì)超過max_threads，多創(chuàng)建出來的線程在一段時(shí)間內(nèi)沒有執(zhí)行任務(wù)則會(huì)自動(dòng)被回收掉，最終線程個(gè)數(shù)保持在核心線程數(shù)。

• 超時(shí)時(shí)間(time_out)：如上所述，多創(chuàng)建出來的線程在time_out時(shí)間內(nèi)沒有執(zhí)行任務(wù)就會(huì)被回收。

• 可獲取當(dāng)前線程池中線程的總個(gè)數(shù)。

• 可獲取當(dāng)前線程池中空閑線程的個(gè)數(shù)。

• 開啟線程池功能的開關(guān)。

• 關(guān)閉線程池功能的開關(guān)，可以選擇是否立即關(guān)閉，立即關(guān)閉線程池時(shí)，當(dāng)前線程池里緩存的任務(wù)不會(huì)被執(zhí)行。

如何實(shí)現(xiàn)線程池？下面是自己實(shí)現(xiàn)的線程池邏輯。

線程池中主要的數(shù)據(jù)結(jié)構(gòu)

1. 鏈表或者數(shù)組：用于存儲(chǔ)線程池中的線程。

2. 隊(duì)列：用于存儲(chǔ)需要放入線程池中執(zhí)行的任務(wù)。

3. 條件變量：當(dāng)有任務(wù)需要執(zhí)行時(shí)，用于通知正在等待的線程從任務(wù)隊(duì)列中取出任務(wù)執(zhí)行。

代碼如下：

class ThreadPool { public: using PoolSeconds = std::chrono::seconds;  /** 線程池的配置 * core_threads: 核心線程個(gè)數(shù)，線程池中最少擁有的線程個(gè)數(shù)，初始化就會(huì)創(chuàng)建好的線程，常駐于線程池 * * max_threads: >=core_threads，當(dāng)任務(wù)的個(gè)數(shù)太多線程池執(zhí)行不過來時(shí)， * 內(nèi)部就會(huì)創(chuàng)建更多的線程用于執(zhí)行更多的任務(wù)，內(nèi)部線程數(shù)不會(huì)超過max_threads * * max_task_size: 內(nèi)部允許存儲(chǔ)的最大任務(wù)個(gè)數(shù)，暫時(shí)沒有使用 * * time_out: Cache線程的超時(shí)時(shí)間，Cache線程指的是max_threads-core_threads的線程, * 當(dāng)time_out時(shí)間內(nèi)沒有執(zhí)行任務(wù)，此線程就會(huì)被自動(dòng)回收 */ struct ThreadPoolConfig { int core_threads; int max_threads; int max_task_size; PoolSeconds time_out; };  /** * 線程的狀態(tài)：有等待、運(yùn)行、停止 */ enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 };  /** * 線程的種類標(biāo)識(shí)：標(biāo)志該線程是核心線程還是Cache線程，Cache是內(nèi)部為了執(zhí)行更多任務(wù)臨時(shí)創(chuàng)建出來的 */ enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 };  using ThreadPtr = std::shared_ptr<std::thread>; using ThreadId = std::atomic<int>; using ThreadStateAtomic = std::atomic; using ThreadFlagAtomic = std::atomic;  /** * 線程池中線程存在的基本單位，每個(gè)線程都有個(gè)自定義的ID，有線程種類標(biāo)識(shí)和狀態(tài) */ 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<std::mutex>;  private: ThreadPoolConfig config_;  std::listworker_threads_;  std::queue<std::function<void()>> 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_; // 用于為新創(chuàng)建的線程分配ID  std::atomic<bool> is_shutdown_now_; std::atomic<bool> is_shutdown_; std::atomic<bool> is_available_;};

線程池的初始化

在構(gòu)造函數(shù)中將各個(gè)成員變量都附初值，同時(shí)判斷線程池的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;}

如何開啟線程池功能？

創(chuàng)建核心線程數(shù)個(gè)線程，常駐于線程池，等待任務(wù)的執(zhí)行，線程ID由GetNextThreadId()統(tǒng)一分配。

// 開啟線程池功能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;}

如何關(guān)閉線程？

這里有兩個(gè)標(biāo)志位，is_shutdown_now置為true表示立即關(guān)閉線程，is_shutdown置為true則表示先執(zhí)行完隊(duì)列里的任務(wù)再關(guān)閉線程池。

// 關(guān)掉線程池，內(nèi)部還沒有執(zhí)行的任務(wù)會(huì)繼續(xù)執(zhí)行void ShutDown() { ShutDown(false); cout << "shutdown" << endl;} // 執(zhí)行關(guān)掉線程池，內(nèi)部還沒有執(zhí)行的任務(wù)直接取消，不會(huì)再執(zhí)行void ShutDownNow() { ShutDown(true); cout << "shutdown now" << endl;} // privatevoid 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()函數(shù)，默認(rèn)會(huì)創(chuàng)建Core線程，也可以選擇創(chuàng)建Cache線程，線程內(nèi)部會(huì)有一個(gè)死循環(huán)，不停的等待任務(wù)，有任務(wù)到來時(shí)就會(huì)執(zhí)行，同時(shí)內(nèi)部會(huì)判斷是否是Cache線程，如果是Cache線程，timeout時(shí)間內(nèi)沒有任務(wù)執(zhí)行就會(huì)自動(dòng)退出循環(huán)，線程結(jié)束。

這里還會(huì)檢查is_shutdown和is_shutdown_now標(biāo)志，根據(jù)兩個(gè)標(biāo)志位是否為true來判斷是否結(jié)束線程。

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<void()> 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::thread>(std::move(func)); if (thread_ptr->ptr->joinable()) { thread_ptr->ptr->detach(); } this->worker_threads_.emplace_back(std::move(thread_ptr));}

如何將任務(wù)放入線程池中執(zhí)行？

見如下代碼，將任務(wù)使用std::bind封裝成std::function放入任務(wù)隊(duì)列中，任務(wù)較多時(shí)內(nèi)部還會(huì)判斷是否有空閑線程，如果沒有空閑線程，會(huì)自動(dòng)創(chuàng)建出最多(max_threads-core_threads)個(gè)Cache線程用于執(zhí)行任務(wù)。

// 放在線程池中執(zhí)行函數(shù)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::futureres = 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));}

如何獲取當(dāng)前線程池中線程的總個(gè)數(shù)？

		

			
			

				
			

		

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

如何獲取當(dāng)前線程池中空閑線程的個(gè)數(shù)？

waiting_thread_num值表示空閑線程的個(gè)數(shù)，該變量在線程循環(huán)內(nèi)部會(huì)更新。

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

簡(jiǎn)單的測(cè)試代碼

		

			
			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

			

				
			

		

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;}