C++執行緒池

阿新 • • 發佈：2020-09-08

1.基礎概念

執行緒池（thread pool）: 一種執行緒使用模式，執行緒過多會帶來排程開銷，進而影響快取區域性性和整體效能。而執行緒池維護著多個執行緒，等待著監督管理者分配可併發執行的任務。這避免了在短時間任務建立與銷燬執行緒的代價。執行緒池不僅能夠保證核心的充分利用，還能防止過分排程。可用執行緒資料取決於可用的併發處理器、處理器核心、記憶體、網路sockets等數量。

2. 執行緒池的組成

2.1 執行緒池管理器

建立一定數量的執行緒，啟動執行緒，調配任務，管理著執行緒池。
執行緒池目前只需要啟動(start())，停止方法(stop())，及任務新增方法(addTask).
start()建立一定數量的執行緒池，進入執行緒迴圈.
stop()停止所有的執行緒迴圈，回收所有資源.
addTask()新增任務.

2.2 工作執行緒

執行緒池中執行緒，線上程池中等待並執行分配任務.
該文選用條件變數實現等待和通知機制

2.3 任務介面

新增任務介面，以供工作執行緒排程任務的執行.

2.4 任務佇列

用於存放沒有處理的任務，提供一種緩衝機制，同時任務佇列具有排程功能，高階優先的任務放在任務佇列前面；本文選用priority_queue與pair的結合用作任務優先佇列結構.

3. 執行緒池工作的四種情況

假設我們的執行緒池大小為3，任務佇列目前不做大小限制。

3.1 主程式當前沒有任務要執行，執行緒池中任務佇列為空閒狀態

下面情況下所有工作執行緒處於空閒的等待狀態，任務緩衝佇列為空.

3.2 主程式新增小於等於執行緒池中執行緒數量得任務

基於3.1情況，所有的工作執行緒已處於等待狀態，主執行緒開始新增三個任務，新增後通知(notif())喚醒執行緒池中的執行緒開始取(take())任務執行。此時的任務緩衝佇列還是空。

3.3 主程式新增任務數量大於當前執行緒池中執行緒數量的任務

基於3.2情況，所有工作執行緒都在工作中，主執行緒開始新增第四個任務，新增後發現現線上程池中執行緒用完了，於是存入任務緩衝佇列。工作執行緒空閒後主動從任務佇列取任務執行。

3.4 主程式新增任務數量大於當前執行緒池中執行緒數量的任務，且任務緩衝佇列已滿

此情況發生情形3且設定了任務緩衝佇列大小後面，主程式新增第N個任務，新增後發現執行緒池中執行緒已經用完了，任務緩衝佇列已滿，於是進入等待狀態，等待任務緩衝佇列中任務騰空通知。但是這種情形會阻塞主執行緒，本文不限制任務佇列的大小，必要時再優化。

4. 執行緒池的C++實現

參考連線： Thread pool；ThreadPool

執行緒池的主要組成由三個部分構成：

任務佇列（Task Quene）
執行緒池（Thread Pool）
完成佇列（Completed Tasks）

等待通知機制通過條件變數實現，Logger和CurrentThread，用於除錯，可以無視。

#ifndef _THREADPOOL_HH
#define _THREADPOOL_HH

#include <vector>
#include <utility>
#include <queue>
#include <thread>
#include <functional>
#include <mutex>

#include "Condition.hh"

class ThreadPool{
public:
  static const int kInitThreadsSize = 3;
  enum taskPriorityE { level0, level1, level2, };
  typedef std::function<void()> Task;
  typedef std::pair<taskPriorityE, Task> TaskPair;

  ThreadPool();
  ~ThreadPool();

  void start();
  void stop();
  void addTask(const Task&);
  void addTask(const TaskPair&);

private:
  ThreadPool(const ThreadPool&);//禁止複製拷貝.
  const ThreadPool& operator=(const ThreadPool&);

  struct TaskPriorityCmp
  {
    bool operator()(const ThreadPool::TaskPair p1, const ThreadPool::TaskPair p2)
    {
        return p1.first > p2.first; //first的小值優先
    }
  };

  void threadLoop();
  Task take();

  typedef std::vector<std::thread*> Threads;
  typedef std::priority_queue<TaskPair, std::vector<TaskPair>, TaskPriorityCmp> Tasks;

  Threads m_threads;
  Tasks m_tasks;

  std::mutex m_mutex;
  Condition m_cond;
  bool m_isStarted;
};

#endif

//Cpp

#include <assert.h>

#include "Logger.hh" // debug
#include "CurrentThread.hh" // debug
#include "ThreadPool.hh"

ThreadPool::ThreadPool()
  :m_mutex(),
  m_cond(m_mutex),
  m_isStarted(false)
{

}

ThreadPool::~ThreadPool()
{
  if(m_isStarted)
  {
    stop();
  }
}

void ThreadPool::start()
{
  assert(m_threads.empty());
  m_isStarted = true;
  m_threads.reserve(kInitThreadsSize);
  for (int i = 0; i < kInitThreadsSize; ++i)
  {
    m_threads.push_back(new std::thread(std::bind(&ThreadPool::threadLoop, this)));
  }

}

void ThreadPool::stop()
{
  LOG_TRACE << "ThreadPool::stop() stop.";
  {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_isStarted = false;
    m_cond.notifyAll();
    LOG_TRACE << "ThreadPool::stop() notifyAll().";
  }

  for (Threads::iterator it = m_threads.begin(); it != m_threads.end() ; ++it)
  {
    (*it)->join();
    delete *it;
  }
  m_threads.clear();
}


void ThreadPool::threadLoop()
{
  LOG_TRACE << "ThreadPool::threadLoop() tid : " << CurrentThread::tid() << " start.";
  while(m_isStarted)
  {
    Task task = take();
    if(task)
    {
      task();
    }
  }
  LOG_TRACE << "ThreadPool::threadLoop() tid : " << CurrentThread::tid() << " exit.";
}

void ThreadPool::addTask(const Task& task)
{
  std::unique_lock<std::mutex> lock(m_mutex);
  /*while(m_tasks.isFull())
    {//when m_tasks have maxsize
      cond2.wait();
    }
  */
  TaskPair taskPair(level2, task);
  m_tasks.push(taskPair);
  m_cond.notify();
}

void ThreadPool::addTask(const TaskPair& taskPair)
{
  std::unique_lock<std::mutex> lock(m_mutex);
  /*while(m_tasks.isFull())
    {//when m_tasks have maxsize
      cond2.wait();
    }
  */
  m_tasks.push(taskPair);
  m_cond.notify();
}

ThreadPool::Task ThreadPool::take()
{
  std::unique_lock<std::mutex> lock(m_mutex);
  //always use a while-loop, due to spurious wakeup
  while(m_tasks.empty() && m_isStarted)
  {
    LOG_TRACE << "ThreadPool::take() tid : " << CurrentThread::tid() << " wait.";
    m_cond.wait(lock);
  }

  LOG_TRACE << "ThreadPool::take() tid : " << CurrentThread::tid() << " wakeup.";

  Task task;
  Tasks::size_type size = m_tasks.size();
  if(!m_tasks.empty() && m_isStarted)
  {
    task = m_tasks.top().second;
    m_tasks.pop();
    assert(size - 1 == m_tasks.size());
    /*if (TaskQueueSize_ > 0)
    {
      cond2.notify();
    }*/
  }

  return task;

}

4.1 佇列

佇列作為先進先出的資料結構，當有可用的工作時，執行緒從佇列中獲取工作並執行。如果兩個執行緒同時執行相同的工作會出現程式崩潰。為了避免這種問題，需要再標準C++ Queue上實現一個包裝器，使用mutex來限制併發訪問。

void enqueue(T& t) {
    std::unique_lock<std::mutex> lock(m_mutex);
    m_queue.push(t);
}

要排隊做的第一件事情就是鎖定互斥鎖來確保沒有其他人正在訪問該資源。然後，將元素推送到隊列當中。當鎖超出範圍時，它會自動釋放，這樣使Queue執行緒安全，因此不用擔心許多執行緒在相同時間訪問或者修改它。

4.2 提交函式

執行緒池最重要的方法是負責向佇列新增任務。

5. 測試程式

5.1 start()、stop()

測試執行緒池基本的建立退出工作，以及檢測資源是否回收正常。

int main(){
  {
  ThreadPool threadPool;
  threadPool.start();
  getchar();}

  getchar();
  return 0;
}

./test.out 
2018-11-25 16:50:36.054805 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3680 start.
2018-11-25 16:50:36.054855 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3680 wait.
2018-11-25 16:50:36.055633 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3679 start.
2018-11-25 16:50:36.055676 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3679 wait.
2018-11-25 16:50:36.055641 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3681 start.
2018-11-25 16:50:36.055701 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3681 wait.
2018-11-25 16:50:36.055736 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3682 start.
2018-11-25 16:50:36.055746 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3682 wait.

2018-11-25 16:51:01.411792 [TRACE] [ThreadPool.cpp:36] [stop] ThreadPool::stop() stop.
2018-11-25 16:51:01.411863 [TRACE] [ThreadPool.cpp:39] [stop] ThreadPool::stop() notifyAll().
2018-11-25 16:51:01.411877 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3680 wakeup.
2018-11-25 16:51:01.411883 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3680 exit.
2018-11-25 16:51:01.412062 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3682 wakeup.
2018-11-25 16:51:01.412110 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3682 exit.
2018-11-25 16:51:01.413052 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3679 wakeup.
2018-11-25 16:51:01.413098 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3679 exit.
2018-11-25 16:51:01.413112 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3681 wakeup.
2018-11-25 16:51:01.413141 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3681 exit.

5.2 addTask()、 PriorityTaskQueue

測試新增任務介面，以及優先順序任務佇列。主執行緒首先新增5個普通任務，1s後新增一個高優先順序任務，當前3個執行緒中的最先一個空閒後，會最先執行後面新增的priorityFunc().

std::mutex g_mutex;

void priorityFunc()
{
  for (int i = 1; i < 4; ++i)
  {
      std::this_thread::sleep_for(std::chrono::seconds(1));
      std::lock_guard<std::mutex> lock(g_mutex);
      LOG_DEBUG << "priorityFunc() [" << i << "at thread [ " << CurrentThread::tid() << "] output";// << std::endl;
  }

}

void testFunc()
{
  // loop to print character after a random period of time
  for (int i = 1; i < 4; ++i)
  {
      std::this_thread::sleep_for(std::chrono::seconds(1));
      std::lock_guard<std::mutex> lock(g_mutex);
      LOG_DEBUG << "testFunc() [" << i << "] at thread [ " << CurrentThread::tid() << "] output";// << std::endl;
  }

}


int main()
{
  ThreadPool threadPool;
  threadPool.start();

  for(int i = 0; i < 5 ; i++)
    threadPool.addTask(testFunc);

  std::this_thread::sleep_for(std::chrono::seconds(1));

  threadPool.addTask(ThreadPool::TaskPair(ThreadPool::level0, priorityFunc));

  getchar();
  return 0;
}

./test.out 
2018-11-25 18:24:20.886837 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4121 start.
2018-11-25 18:24:20.886893 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
2018-11-25 18:24:20.887580 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4120 start.
2018-11-25 18:24:20.887606 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
2018-11-25 18:24:20.887610 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4122 start.
2018-11-25 18:24:20.887620 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
2018-11-25 18:24:21.887779 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4120] output
2018-11-25 18:24:21.887813 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4122] output
2018-11-25 18:24:21.888909 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4121] output
2018-11-25 18:24:22.888049 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4120] output
2018-11-25 18:24:22.888288 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4122] output
2018-11-25 18:24:22.889978 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4121] output
2018-11-25 18:24:23.888467 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4120] output
2018-11-25 18:24:23.888724 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
2018-11-25 18:24:23.888778 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4122] output
2018-11-25 18:24:23.888806 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
2018-11-25 18:24:23.890413 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4121] output
2018-11-25 18:24:23.890437 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
2018-11-25 18:24:24.889247 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [1at thread [ 4120] output
2018-11-25 18:24:24.891187 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4121] output
2018-11-25 18:24:24.893163 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4122] output
2018-11-25 18:24:25.889567 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [2at thread [ 4120] output
2018-11-25 18:24:25.891477 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4121] output
2018-11-25 18:24:25.893450 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4122] output
2018-11-25 18:24:26.890295 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [3at thread [ 4120] output
2018-11-25 18:24:26.890335 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4120 wait.
2018-11-25 18:24:26.892265 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4121] output
2018-11-25 18:24:26.892294 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4121 wait.
2018-11-25 18:24:26.894274 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4122] output
2018-11-25 18:24:26.894299 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4122 wait.

2018-11-25 18:24:35.359003 [TRACE] [ThreadPool.cpp:37] [stop] ThreadPool::stop() stop.
2018-11-25 18:24:35.359043 [TRACE] [ThreadPool.cpp:42] [stop] ThreadPool::stop() notifyAll().
2018-11-25 18:24:35.359061 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
2018-11-25 18:24:35.359067 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4120 exit.
2018-11-25 18:24:35.359080 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
2018-11-25 18:24:35.359090 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4122 exit.
2018-11-25 18:24:35.359123 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
2018-11-25 18:24:35.359130 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4121 exit.

上述原始碼下載
https://github.com/BethlyRoseDaisley/ThreadPool

參考文獻
[1] https://www.cnblogs.com/ailumiyana/p/10016965.html#原始碼下載

C++執行緒池

1.基礎概念

2. 執行緒池的組成

2.1 執行緒池管理器

2.2 工作執行緒

2.3 任務介面

2.4 任務佇列

3. 執行緒池工作的四種情況

3.1 主程式當前沒有任務要執行，執行緒池中任務佇列為空閒狀態

3.2 主程式新增小於等於執行緒池中執行緒數量得任務

3.3 主程式新增任務數量大於當前執行緒池中執行緒數量的任務

3.4 主程式新增任務數量大於當前執行緒池中執行緒數量的任務，且任務緩衝佇列已滿

4. 執行緒池的C++實現

4.1 佇列

4.2 提交函式

5. 測試程式

5.1 start()、stop()

5.2 addTask()、 PriorityTaskQueue

C# 執行緒池

C++執行緒池

C# 執行緒池ThreadPool的用法簡析

C 執行緒池

C++執行緒池的實現

.NET C# 執行緒池

C++實現執行緒池的簡單方法示例

執行緒池原理及C語言實現執行緒池

C# 執行緒執行緒池 Task的使用介紹

C++實現執行緒池

c++11執行緒池

C++11 簡單實現執行緒池的方法

C#多執行緒中等待執行緒池中的所有執行緒執行完畢後再執行下一個執行緒

C++11多執行緒程式設計(六)——執行緒池的實現

c++可以攜帶任意引數的執行緒池

實現執行緒池-Linux C版本

C++ 動態伸縮執行緒池

C++11 執行緒池

c++ 11 執行緒池---完全使用c++ 11新特性

[C++]C風格、C++風格和C++11特性的執行緒池

C++執行緒池

1.基礎概念

2. 執行緒池的組成

2.1 執行緒池管理器

2.2 工作執行緒

2.3 任務介面

2.4 任務佇列

3. 執行緒池工作的四種情況

3.1 主程式當前沒有任務要執行，執行緒池中任務佇列為空閒狀態

3.2 主程式新增小於等於執行緒池中執行緒數量得任務

3.3 主程式新增任務數量大於當前執行緒池中執行緒數量的任務

3.4 主程式新增任務數量大於當前執行緒池中執行緒數量的任務，且任務緩衝佇列已滿

4. 執行緒池的C++實現

4.1 佇列

4.2 提交函式

5. 測試程式

5.1 start()、stop()

5.2 addTask()、 PriorityTaskQueue

相關推薦