C++实现线程池案例

admin

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++实现

线程池的主要组成由三个部分构成：

• 任务队列（Task Quene）
• 线程池（Thread Pool）
• 完成队列（Completed Tasks）
  

等待通知机制通过条件变量实现，Logger和CurrentThread，用于调试，可以无视。

1. #ifndef _THREADPOOL_HH
   
2. #define _THREADPOOL_HH
   
3. 
   
4. #include <vector>
   
5. #include <utility>
   
6. #include <queue>
   
7. #include <thread>
   
8. #include <functional>
   
9. #include <mutex>
   
10. 
    
11. #include "Condition.hh"
    
12. 
    
13. class ThreadPool{
    
14. public:
    
15.   static const int kInitThreadsSize = 3;
    
16.   enum taskPriorityE { level0, level1, level2, };
    
17.   typedef std::function<void()> Task;
    
18.   typedef std::pair<taskPriorityE, Task> TaskPair;
    
19. 
    
20.   ThreadPool();
    
21.   ~ThreadPool();
    
22. 
    
23.   void start();
    
24.   void stop();
    
25.   void addTask(const Task&);
    
26.   void addTask(const TaskPair&);
    
27. 
    
28. private:
    
29.   ThreadPool(const ThreadPool&);//禁止复制拷贝.
    
30.   const ThreadPool& operator=(const ThreadPool&);
    
31. 
    
32.   struct TaskPriorityCmp
    
33.   {
    
34.     bool operator()(const ThreadPool::TaskPair p1, const ThreadPool::TaskPair p2)
    
35.     {
    
36.         return p1.first > p2.first; //first的小值优先
    
37.     }
    
38.   };
    
39. 
    
40.   void threadLoop();
    
41.   Task take();
    
42. 
    
43.   typedef std::vector<std::thread*> Threads;
    
44.   typedef std::priority_queue<TaskPair, std::vector<TaskPair>, TaskPriorityCmp> Tasks;
    
45. 
    
46.   Threads m_threads;
    
47.   Tasks m_tasks;
    
48. 
    
49.   std::mutex m_mutex;
    
50.   Condition m_cond;
    
51.   bool m_isStarted;
    
52. };
    
53. 
    
54. #endif
    
55. 
    
56. //Cpp
    
57. 
    
58. #include <assert.h>
    
59. 
    
60. #include "Logger.hh" // debug
    
61. #include "CurrentThread.hh" // debug
    
62. #include "ThreadPool.hh"
    
63. 
    
64. ThreadPool::ThreadPool()
    
65.   :m_mutex(),
    
66.   m_cond(m_mutex),
    
67.   m_isStarted(false)
    
68. {
    
69. 
    
70. }
    
71. 
    
72. ThreadPool::~ThreadPool()
    
73. {
    
74.   if(m_isStarted)
    
75.   {
    
76.     stop();
    
77.   }
    
78. }
    
79. 
    
80. void ThreadPool::start()
    
81. {
    
82.   assert(m_threads.empty());
    
83.   m_isStarted = true;
    
84.   m_threads.reserve(kInitThreadsSize);
    
85.   for (int i = 0; i < kInitThreadsSize; ++i)
    
86.   {
    
87.     m_threads.push_back(new std::thread(std::bind(&ThreadPool::threadLoop, this)));
    
88.   }
    
89. 
    
90. }
    
91. 
    
92. void ThreadPool::stop()
    
93. {
    
94.   LOG_TRACE << "ThreadPool::stop() stop.";
    
95.   {
    
96.     std::unique_lock<std::mutex> lock(m_mutex);
    
97.     m_isStarted = false;
    
98.     m_cond.notifyAll();
    
99.     LOG_TRACE << "ThreadPool::stop() notifyAll().";
    
100.   }
     
101. 
     
102.   for (Threads::iterator it = m_threads.begin(); it != m_threads.end() ; ++it)
     
103.   {
     
104.     (*it)->join();
     
105.     delete *it;
     
106.   }
     
107.   m_threads.clear();
     
108. }
     
109. 
     
110. 
     
111. void ThreadPool::threadLoop()
     
112. {
     
113.   LOG_TRACE << "ThreadPool::threadLoop() tid : " << CurrentThread::tid() << " start.";
     
114.   while(m_isStarted)
     
115.   {
     
116.     Task task = take();
     
117.     if(task)
     
118.     {
     
119.       task();
     
120.     }
     
121.   }
     
122.   LOG_TRACE << "ThreadPool::threadLoop() tid : " << CurrentThread::tid() << " exit.";
     
123. }
     
124. 
     
125. void ThreadPool::addTask(const Task& task)
     
126. {
     
127.   std::unique_lock<std::mutex> lock(m_mutex);
     
128.   /*while(m_tasks.isFull())
     
129.     {//when m_tasks have maxsize
     
130.       cond2.wait();
     
131.     }
     
132.   */
     
133.   TaskPair taskPair(level2, task);
     
134.   m_tasks.push(taskPair);
     
135.   m_cond.notify();
     
136. }
     
137. 
     
138. void ThreadPool::addTask(const TaskPair& taskPair)
     
139. {
     
140.   std::unique_lock<std::mutex> lock(m_mutex);
     
141.   /*while(m_tasks.isFull())
     
142.     {//when m_tasks have maxsize
     
143.       cond2.wait();
     
144.     }
     
145.   */
     
146.   m_tasks.push(taskPair);
     
147.   m_cond.notify();
     
148. }
     
149. 
     
150. ThreadPool::Task ThreadPool::take()
     
151. {
     
152.   std::unique_lock<std::mutex> lock(m_mutex);
     
153.   //always use a while-loop, due to spurious wakeup
     
154.   while(m_tasks.empty() && m_isStarted)
     
155.   {
     
156.     LOG_TRACE << "ThreadPool::take() tid : " << CurrentThread::tid() << " wait.";
     
157.     m_cond.wait(lock);
     
158.   }
     
159. 
     
160.   LOG_TRACE << "ThreadPool::take() tid : " << CurrentThread::tid() << " wakeup.";
     
161. 
     
162.   Task task;
     
163.   Tasks::size_type size = m_tasks.size();
     
164.   if(!m_tasks.empty() && m_isStarted)
     
165.   {
     
166.     task = m_tasks.top().second;
     
167.     m_tasks.pop();
     
168.     assert(size - 1 == m_tasks.size());
     
169.     /*if (TaskQueueSize_ > 0)
     
170.     {
     
171.       cond2.notify();
     
172.     }*/
     
173.   }
     
174. 
     
175.   return task;
     
176. 
     
177. }
     

复制代码

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()

测试线程池基本的创建退出工作，以及检测资源是否回收正常。

1. int main(){
   
2.   {
   
3.   ThreadPool threadPool;
   
4.   threadPool.start();
   
5.   getchar();}
   
6. 
   
7.   getchar();
   
8.   return 0;
   
9. }

复制代码

1. ./test.out
   
2. 2018-11-25 16:50:36.054805 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3680 start.
   
3. 2018-11-25 16:50:36.054855 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3680 wait.
   
4. 2018-11-25 16:50:36.055633 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3679 start.
   
5. 2018-11-25 16:50:36.055676 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3679 wait.
   
6. 2018-11-25 16:50:36.055641 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3681 start.
   
7. 2018-11-25 16:50:36.055701 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3681 wait.
   
8. 2018-11-25 16:50:36.055736 [TRACE] [ThreadPool.cpp:53] [threadLoop] ThreadPool::threadLoop() tid : 3682 start.
   
9. 2018-11-25 16:50:36.055746 [TRACE] [ThreadPool.cpp:72] [take] ThreadPool::take() tid : 3682 wait.
   
10. 
    
11. 2018-11-25 16:51:01.411792 [TRACE] [ThreadPool.cpp:36] [stop] ThreadPool::stop() stop.
    
12. 2018-11-25 16:51:01.411863 [TRACE] [ThreadPool.cpp:39] [stop] ThreadPool::stop() notifyAll().
    
13. 2018-11-25 16:51:01.411877 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3680 wakeup.
    
14. 2018-11-25 16:51:01.411883 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3680 exit.
    
15. 2018-11-25 16:51:01.412062 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3682 wakeup.
    
16. 2018-11-25 16:51:01.412110 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3682 exit.
    
17. 2018-11-25 16:51:01.413052 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3679 wakeup.
    
18. 2018-11-25 16:51:01.413098 [TRACE] [ThreadPool.cpp:62] [threadLoop] ThreadPool::threadLoop() tid : 3679 exit.
    
19. 2018-11-25 16:51:01.413112 [TRACE] [ThreadPool.cpp:76] [take] ThreadPool::take() tid : 3681 wakeup.
    
20. 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().

1. std::mutex g_mutex;
   
2. 
   
3. void priorityFunc()
   
4. {
   
5.   for (int i = 1; i < 4; ++i)
   
6.   {
   
7.       std::this_thread::sleep_for(std::chrono::seconds(1));
   
8.       std::lock_guard<std::mutex> lock(g_mutex);
   
9.       LOG_DEBUG << "priorityFunc() [" << i << "at thread [ " << CurrentThread::tid() << "] output";// << std::endl;
   
10.   }
    
11. 
    
12. }
    
13. 
    
14. void testFunc()
    
15. {
    
16.   // loop to print character after a random period of time
    
17.   for (int i = 1; i < 4; ++i)
    
18.   {
    
19.       std::this_thread::sleep_for(std::chrono::seconds(1));
    
20.       std::lock_guard<std::mutex> lock(g_mutex);
    
21.       LOG_DEBUG << "testFunc() [" << i << "] at thread [ " << CurrentThread::tid() << "] output";// << std::endl;
    
22.   }
    
23. 
    
24. }
    
25. 
    
26. 
    
27. int main()
    
28. {
    
29.   ThreadPool threadPool;
    
30.   threadPool.start();
    
31. 
    
32.   for(int i = 0; i < 5 ; i++)
    
33.     threadPool.addTask(testFunc);
    
34. 
    
35.   std::this_thread::sleep_for(std::chrono::seconds(1));
    
36. 
    
37.   threadPool.addTask(ThreadPool::TaskPair(ThreadPool::level0, priorityFunc));
    
38. 
    
39.   getchar();
    
40.   return 0;
    
41. }
    

复制代码

1. ./test.out
   
2. 2018-11-25 18:24:20.886837 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4121 start.
   
3. 2018-11-25 18:24:20.886893 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
   
4. 2018-11-25 18:24:20.887580 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4120 start.
   
5. 2018-11-25 18:24:20.887606 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
   
6. 2018-11-25 18:24:20.887610 [TRACE] [ThreadPool.cpp:56] [threadLoop] ThreadPool::threadLoop() tid : 4122 start.
   
7. 2018-11-25 18:24:20.887620 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
   
8. 2018-11-25 18:24:21.887779 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4120] output
   
9. 2018-11-25 18:24:21.887813 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4122] output
   
10. 2018-11-25 18:24:21.888909 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4121] output
    
11. 2018-11-25 18:24:22.888049 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4120] output
    
12. 2018-11-25 18:24:22.888288 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4122] output
    
13. 2018-11-25 18:24:22.889978 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4121] output
    
14. 2018-11-25 18:24:23.888467 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4120] output
    
15. 2018-11-25 18:24:23.888724 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
    
16. 2018-11-25 18:24:23.888778 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4122] output
    
17. 2018-11-25 18:24:23.888806 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
    
18. 2018-11-25 18:24:23.890413 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4121] output
    
19. 2018-11-25 18:24:23.890437 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
    
20. 2018-11-25 18:24:24.889247 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [1at thread [ 4120] output
    
21. 2018-11-25 18:24:24.891187 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4121] output
    
22. 2018-11-25 18:24:24.893163 [DEBUG] [main.cpp:104] [testFunc] testFunc() [1] at thread [ 4122] output
    
23. 2018-11-25 18:24:25.889567 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [2at thread [ 4120] output
    
24. 2018-11-25 18:24:25.891477 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4121] output
    
25. 2018-11-25 18:24:25.893450 [DEBUG] [main.cpp:104] [testFunc] testFunc() [2] at thread [ 4122] output
    
26. 2018-11-25 18:24:26.890295 [DEBUG] [main.cpp:92] [priorityFunc] priorityFunc() [3at thread [ 4120] output
    
27. 2018-11-25 18:24:26.890335 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4120 wait.
    
28. 2018-11-25 18:24:26.892265 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4121] output
    
29. 2018-11-25 18:24:26.892294 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4121 wait.
    
30. 2018-11-25 18:24:26.894274 [DEBUG] [main.cpp:104] [testFunc] testFunc() [3] at thread [ 4122] output
    
31. 2018-11-25 18:24:26.894299 [TRACE] [ThreadPool.cpp:99] [take] ThreadPool::take() tid : 4122 wait.
    
32. 
    
33. 2018-11-25 18:24:35.359003 [TRACE] [ThreadPool.cpp:37] [stop] ThreadPool::stop() stop.
    
34. 2018-11-25 18:24:35.359043 [TRACE] [ThreadPool.cpp:42] [stop] ThreadPool::stop() notifyAll().
    
35. 2018-11-25 18:24:35.359061 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4120 wakeup.
    
36. 2018-11-25 18:24:35.359067 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4120 exit.
    
37. 2018-11-25 18:24:35.359080 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4122 wakeup.
    
38. 2018-11-25 18:24:35.359090 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4122 exit.
    
39. 2018-11-25 18:24:35.359123 [TRACE] [ThreadPool.cpp:103] [take] ThreadPool::take() tid : 4121 wakeup.
    
40. 2018-11-25 18:24:35.359130 [TRACE] [ThreadPool.cpp:65] [threadLoop] ThreadPool::threadLoop() tid : 4121 exit.

复制代码