1. 线程池原理
我们使用线程的时候就去创建一个线程，这样实现起来非常简便 但是就会有一个问题：如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。
那么有没有一种办法使得线程可以复用: 执行完一个任务，并不被销毁，而是可以继续执行其他的任务呢？
线程池是一种多线程处理形式，处理过程中将任务添加到队列
在创建线程后自动启动这些任务。线程池线程都是后台线程。
每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。
如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。
如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。
超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。
在各个编程语言的语种中都有线程池的概念，并且很多语言中直接提供了线程池，作为程序猿直接使用就可以了，下面给大家介绍一下线程池的实现原理：
-
线程池的组成主要分为3个部分，这三部分配合工作就可以得到一个完整的线程池：
任务队列，存储需要处理的任务，由工作的线程来处理这些任务- 通过线程池提供的API函数，将一个待处理的任务添加到任务队列，或者从任务队列中删除
- 已处理的任务会被从任务队列中删除
- 线程池的使用者，也就是调用线程池函数往任务队列中添加任务的线程就是生产者线程
工作的线程（任务队列任务的消费者） ，N个- 线程池中维护了一定数量的工作线程, 他们的作用是是不停的读任务队列, 从里边取出任务并处理
- 工作的线程相当于是任务队列的消费者角色，
- 如果任务队列为空, 工作的线程将会被阻塞 (使用条件变量/信号量阻塞)
- 如果阻塞之后有了新的任务, 由生产者将阻塞解除, 工作线程开始工作
管理者线程（不处理任务队列中的任务），1个- 它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测
- 当任务过多的时候, 可以适当的创建一些新的工作线程
- 当任务过少的时候, 可以适当的销毁一些工作的线程
- 它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测
2. C实现线程池
2.1 头文件
1#pragma once 2#include<stdio.h> 3#include<stdlib.h> 4#include<pthread.h> 5#include<string.h> 6#include<unistd.h> 7#define NUMBER 2 8 9typedef struct Task 10{ 11 void (*function)(void* args); 12 void* args; 13}Task; 14 15typedef struct ThreadPool 16{ 17 //任务队列 18 Task* task_queue; 19 int queue_capacity; //容量 20 int queue_size; //当前任务个数 21 int queue_front; //队头->消费用 22 int queue_tail; //队尾->生产用 23 24 25 pthread_t managerID; //管理者线程ID(1个) 26 pthread_t* threadID; //工作的线程ID(多个,用指针) 27 int min_num; //最小的线程数量 28 int max_num; //最大的线程数量 29 int busy_num; //在忙的线程数量 30 int live_num; //存活的线程数量 31 int exit_num; //将要销毁的线程数量 32 pthread_mutex_t mutex_pool; //线程池的锁 33 pthread_mutex_t mutex_busy_num; //busy_num的锁 34 pthread_cond_t producer_wait_consumer; //生产者等待消费者 35 pthread_cond_t consumer_wait_producer; //消费者等待生产者 36 37 int shutdown; //是否销毁线程池,销毁为1,否则为0 38}ThreadPool; 39 40 41//创建线程池并且初始化 42ThreadPool* Thread_Pool_Create(int min, int max, int queuesize); 43 44//销毁线程池 45int Thread_Pool_Destroy(ThreadPool* pool); 46 47//给线程池添加任务(相当于生产者) 48void Thread_Pool_Add(ThreadPool* pool, void(*function)(void*), void* args); 49 50//获取线程池中工作的线程个数 51int Get_Busy_Num(ThreadPool* pool); 52 53//获取线程池中存活的个数 54int Get_Live_Num(ThreadPool* pool); 55 56//以下为被封装的函数 57//工作的线程(消费者线程)任务函数 58void* Worker(void* args); 59 60//管理者线程任务函数 61void* Manager(void* args); 62 63//单个线程的退出 64void Thread_Exit(ThreadPool* pool);
2.2 源文件的定义
1#include"threadpool.h" 2 3ThreadPool* Thread_Pool_Create(int min, int max, int queuesize) 4{ 5 //开线程池 6 ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool)); 7 do 8 { 9 if (pool == NULL) 10 { 11 printf("malloc threadpool fail\n"); 12 break; 13 } 14 15 // 开线程ID 16 pool->threadID = (pthread_t*)malloc(sizeof(pthread_t) * max); 17 if (pool->threadID == NULL) 18 { 19 printf("malloc threadID fail\n"); 20 break; 21 } 22 //初始线程ID都初始化为0 23 memset(pool->threadID, 0, sizeof(pthread_t) * max); 24 pool->max_num = max; 25 pool->min_num = min; 26 pool->busy_num = 0; 27 pool->live_num = min; 28 pool->exit_num = 0; 29 30 //初始化锁和条件变量 31 if (pthread_mutex_init(&pool->mutex_pool, NULL) != 0 || 32 pthread_mutex_init(&pool->mutex_busy_num, NULL) != 0 || 33 pthread_cond_init(&pool->producer_wait_consumer, NULL) != 0 || 34 pthread_cond_init(&pool->consumer_wait_producer, NULL) != 0) 35 { 36 printf("mutex or cond init fail\n"); 37 break; 38 } 39 40 //初始化任务队列 41 pool->task_queue = (Task*)malloc(sizeof(Task) * queuesize); 42 if (pool->task_queue == NULL) 43 { 44 printf("malloc task_queue fail...\n"); 45 break; 46 } 47 48 pool->queue_capacity = queuesize; 49 pool->queue_size = 0; 50 pool->queue_front = 0; 51 pool->queue_tail = 0; 52 53 pool->shutdown = 0; 54 55 //创建线程 56 pthread_create(&pool->managerID, NULL, Manager, pool); 57 58 for (int i = 0; i < min; ++i) 59 { 60 pthread_create(&pool->threadID[i], NULL, Worker, pool); 61 } 62 return pool; //到这里说明成功,返回创建出来的线程池 63 } while (0); 64 65 //跳到这里说明失败了,释放资源 66 if (pool && pool->threadID) 67 free(pool->threadID); 68 if (pool) 69 free(pool); 70 71 return NULL; 72} 73 74 75 76int Thread_Pool_Destroy(ThreadPool* pool) 77{ 78 //如果线程池本就不存在 79 if (pool == NULL) 80 return -1; 81 82 //关闭线程池 83 pool->shutdown = 1; 84 85 //阻塞并回收管理者线程 86 pthread_join(pool->managerID, NULL); 87 88 //唤醒阻塞的消费者进程 89 for (int i = 0; i < pool->live_num; ++i) 90 pthread_cond_signal(&pool->consumer_wait_producer); 91 92 //释放堆内存 93 if (pool->task_queue) 94 free(pool->task_queue); 95 if (pool->threadID) 96 free(pool->threadID); 97 98 //销毁锁和条件变量 99 pthread_mutex_destroy(&pool->mutex_pool); 100 pthread_mutex_destroy(&pool->mutex_busy_num); 101 pthread_cond_destroy(&pool->consumer_wait_producer); 102 pthread_cond_destroy(&pool->producer_wait_consumer); 103 104 //销毁线程池 105 free(pool); 106 pool = NULL; 107 108 return 0; 109} 110 111 112 113void Thread_Pool_Add(ThreadPool* pool, void(*function)(void*), void* args) 114{ 115 pthread_mutex_lock(&pool->mutex_pool); 116 //如果任务队列满了 117 while (pool->queue_size == pool->queue_capacity && !pool->shutdown) 118 { 119 //生产者阻塞,等待消费者进程 120 pthread_cond_wait(&pool->producer_wait_consumer, &pool->mutex_pool); 121 } 122 123 //如果线程池要关闭 124 if (pool->shutdown) 125 { 126 pthread_mutex_unlock(&pool->mutex_pool); 127 return; 128 } 129 130 //添加任务 131 //队尾添加任务,队尾向后移动,队列任务数加一 132 pool->task_queue[pool->queue_tail].function = function; 133 pool->task_queue[pool->queue_tail].args = args; 134 pool->queue_tail = (pool->queue_tail + 1) % pool->queue_capacity; 135 pool->queue_size++; 136 137 //提醒消费者消费 138 pthread_cond_signal(&pool->consumer_wait_producer); 139 pthread_mutex_unlock(&pool->mutex_pool); 140} 141 142 143 144int Get_Busy_Num(ThreadPool* pool) 145{ 146 pthread_mutex_lock(&pool->mutex_busy_num); 147 int busynum = pool->busy_num; 148 pthread_mutex_unlock(&pool->mutex_busy_num); 149 return busynum; 150} 151 152 153 154int Get_Live_Num(ThreadPool* pool) 155{ 156 pthread_mutex_lock(&pool->mutex_pool); 157 int livenum = pool->live_num; 158 pthread_mutex_unlock(&pool->mutex_pool); 159 return livenum; 160} 161 162 163 164void* Worker(void* args) 165{ 166 ThreadPool* pool = (ThreadPool*)args; 167 168 while (1) 169 { 170 pthread_mutex_lock(&pool->mutex_pool); 171 172 //如果当前任务队列为空 173 while (pool->queue_size == 0 && !pool->shutdown) 174 { 175 //阻塞消费者线程 176 pthread_cond_wait(&pool->consumer_wait_producer,&pool->mutex_pool); 177 178 //如果要销毁线程 179 if (pool->exit_num > 0) 180 { 181 pool->exit_num--; 182 //同时如果存活线程大于最小值 183 if (pool->live_num > pool->min_num) 184 { 185 pool->live_num--; 186 pthread_mutex_unlock(&pool->mutex_pool); 187 Thread_Exit(pool); 188 } 189 } 190 } 191 192 193 //如果线程池要关闭 194 if (pool->shutdown) 195 { 196 pthread_mutex_unlock(&pool->mutex_pool); 197 Thread_Exit(pool); 198 } 199 200 201 //从任务队列取出任务 202 Task task; 203 task.function = pool->task_queue[pool->queue_front].function; 204 task.args = pool->task_queue[pool->queue_front].args; 205 206 //队头向后移动以下次取任务 207 pool->queue_front = (pool->queue_front + 1) % pool->queue_capacity; 208 pool->queue_size--; 209 210 //解锁提醒生产者 211 pthread_cond_signal(&pool->producer_wait_consumer); 212 pthread_mutex_unlock(&pool->mutex_pool); 213 214 //打印,busynum++ 215 printf("thread %ld start working...\n", pthread_self()); 216 pthread_mutex_lock(&pool->mutex_busy_num); 217 pool->busy_num++; 218 pthread_mutex_unlock(&pool->mutex_busy_num); 219 220 //执行函数,完成任务,销毁任务队列中这个数据(函数) 221 task.function(task.args); 222 free(task.args); 223 task.args = NULL; 224 225 //打印,busynum-- 226 printf("thread %ld end working\n", pthread_self()); 227 pthread_mutex_lock(&pool->mutex_busy_num); 228 pool->busy_num--; 229 pthread_mutex_unlock(&pool->mutex_busy_num); 230 231 if (pool->shutdown) 232 Thread_Exit(pool); 233 } 234 return NULL; 235} 236 237 238 239void* Manager(void* args) 240{ 241 ThreadPool* pool = (ThreadPool*)args; 242 while (!pool->shutdown) 243 { 244 //每隔三秒检查一次 245 sleep(3); 246 247 //取出线程池中任务的数量和当前的线程数量 248 pthread_mutex_lock(&pool->mutex_pool); 249 int queue_size = pool->queue_size; 250 int live_num = pool->live_num; 251 pthread_mutex_unlock(&pool->mutex_pool); 252 253 //取出忙的线程数量 254 pthread_mutex_lock(&pool->mutex_busy_num); 255 int busy_num = pool->busy_num; 256 pthread_mutex_unlock(&pool->mutex_busy_num); 257 258 //添加线程 259 //任务个数>存活线程个数 && 存活线程个数<最大线程数(消费者太少了,任务队列消化太慢 260 if (queue_size > live_num && live_num < pool->max_num) 261 { 262 pthread_mutex_lock(&pool->mutex_pool); 263 int counter = 0; 264 for (int i = 0; i < pool->max_num && 265 counter < NUMBER && 266 pool->live_num < pool->max_num; ++i) 267 { 268 if (pool->threadID[i] == 0) 269 { 270 pthread_create(&pool->threadID[i], NULL, Worker, pool); 271 counter++; 272 pool->live_num++; 273 } 274 } 275 pthread_mutex_unlock(&pool->mutex_pool); 276 } 277 278 //销毁线程 279 //忙的线程*2 < 存活的线程数 && 存活的线程>最小线程数 280 if (busy_num * 2 < live_num && live_num > pool->min_num) 281 { 282 pthread_mutex_lock(&pool->mutex_pool); 283 pool->exit_num = NUMBER; 284 pthread_mutex_unlock(&pool->mutex_pool); 285 286 //让工作的线程自杀 287 for (int i = 0; i < NUMBER; ++i) 288 { 289 pthread_cond_signal(&pool->consumer_wait_producer); 290 } 291 } 292 } 293 return NULL; 294} 295 296void Thread_Exit(ThreadPool* pool) 297{ 298 pthread_t tid = pthread_self(); 299 for (int i = 0; i < pool->max_num; ++i) 300 { 301 if (pool->threadID[i] == tid) 302 { 303 pool->threadID[i] = 0; 304 printf("thread %ld exiting...\n", tid); 305 break; 306 } 307 } 308 pthread_exit(NULL); 309}
2.3 测试部分
1#include "threadpool.h" 2 3void Task_Test(void* args) 4{ 5 int num = *(int*)args; 6 printf("thread %ld is working,number = %d\n", pthread_self(), num); 7 sleep(1); 8 return; 9} 10 11int main() 12{ 13 //创建线程池 14 ThreadPool* pool = Thread_Pool_Create(3, 10, 100); 15 for (int i = 0; i < 100; ++i) 16 { 17 int* num = (int*)malloc(sizeof(int)); 18 *num = i + 100; 19 Thread_Pool_Add(pool, Task_Test, num); 20 } 21 22 sleep(30); 23 24 Thread_Pool_Destroy(pool); 25 return 0; 26}
3. C++实现线程池
3.1 头文件
1#define _CRT_SECURE_NO_WARNINGS 2#pragma once 3#include<iostream> 4#include<string.h> 5#include<string> 6#include<pthread.h> 7#include<stdlib.h> 8#include<queue> 9#include<unistd.h> 10using namespace std; 11 12 13using callback = void(*)(void*); 14//任务的结构体 15template<typename T> 16struct Task 17{ 18 Task() 19 { 20 function = nullptr; 21 args = nullptr; 22 } 23 Task(callback fun, void* args) 24 { 25 function = fun; 26 this -> args = (T*)args; 27 } 28 callback function; 29 T* args; 30}; 31 32//任务队列 33template<typename T> 34class TaskQueue 35{ 36public: 37 TaskQueue() 38 { 39 pthread_mutex_init(&mutex,NULL); 40 } 41 42 ~TaskQueue() 43 { 44 pthread_mutex_destroy(&mutex); 45 } 46 47 //添加任务 48 void AddTask(Task<T> task) 49 { 50 pthread_mutex_lock(&mutex); 51 queue.push(task); 52 pthread_mutex_unlock(&mutex); 53 } 54 void AddTask(callback fun, void* args) 55 { 56 pthread_mutex_lock(&mutex); 57 Task<T> task(fun,args); 58 queue.push(task); 59 pthread_mutex_unlock(&mutex); 60 } 61 62 //取出一个任务 63 Task<T> TakeTask() 64 { 65 Task<T> task; 66 pthread_mutex_lock(&mutex); 67 if (queue.size() > 0) 68 { 69 task = queue.front(); 70 queue.pop(); 71 } 72 pthread_mutex_unlock(&mutex); 73 return task; 74 } 75 76 //获取当前队列中的任务个数 77 inline int GetTaskNum() 78 { 79 return queue.size(); 80 } 81private: 82 pthread_mutex_t mutex; //互斥锁 83 std::queue<Task<T>> queue; 84}; 85 86 87//线程池 88template<typename T> 89class ThreadPool 90{ 91public: 92 ThreadPool(int min , int max) 93 { 94 //实例化任务队列 95 taskqueue = new TaskQueue<T>; 96 97 //初始化线程池 98 min_num = min; 99 max_num = max; 100 busy_num = 0; 101 live_num = min; 102 103 //根据线程最大上限,给线程数组分配内存 104 threadID = new pthread_t[max]; 105 if (threadID == nullptr) 106 { 107 cout << "new threadID fail" << endl; 108 } 109 110 //初始化线程ID 111 memset(threadID, 0, sizeof(pthread_t) * max); 112 113 //初始化互斥锁和条件变量 114 if (pthread_mutex_init(&mutex_pool, NULL) != 0 || 115 pthread_cond_init(¬empty, NULL) != 0) 116 { 117 cout << "mutex or cond init fail" << endl; 118 } 119 120 //创建线程 121 for (size_t i = 0; i < min; ++i) 122 { 123 pthread_create(&threadID[i], NULL, Work, this); 124 cout << "create thread ID :" << to_string(threadID[i]) << endl; 125 } 126 pthread_create(&managerID, NULL, Manage, this); 127 128 } 129 130 ~ThreadPool() 131 { 132 shutdown = true; 133 //销毁管理者进程 134 pthread_join(managerID, NULL); 135 136 //唤醒消费者 137 for (int i = 0; i < live_num; ++i) 138 { 139 pthread_cond_signal(¬empty); 140 } 141 142 if (taskqueue) 143 { 144 delete taskqueue; 145 } 146 147 if (threadID) 148 { 149 delete[] threadID; 150 } 151 152 pthread_mutex_destroy(&mutex_pool); 153 pthread_cond_destroy(¬empty); 154 } 155 156 //添加任务 157 void Add_Task(Task<T> task) 158 { 159 if (shutdown) 160 return; 161 162 //添加任务,不需加锁,队列中有 163 taskqueue->AddTask(task); 164 165 //唤醒消费者 166 pthread_cond_signal(¬empty); 167 } 168 169 //获取忙线程个数 170 int Get_Busy_Num() 171 { 172 int busynum = 0; 173 pthread_mutex_lock(&mutex_pool); 174 busynum = busy_num; 175 pthread_mutex_unlock(&mutex_pool); 176 return busynum; 177 } 178 179 //获取存活线程个数 180 int Get_Live_Num() 181 { 182 int livenum = 0; 183 pthread_mutex_lock(&mutex_pool); 184 livenum = live_num; 185 pthread_mutex_unlock(&mutex_pool); 186 return livenum; 187 } 188 189private: 190 //工作的线程任务函数 191 static void* Work(void* args) 192 { 193 ThreadPool* pool = static_cast<ThreadPool*>(args); 194 195 while (true) 196 { 197 //访问任务队列加锁 198 pthread_mutex_lock(&pool->mutex_pool); 199 //判断任务队列是否为空,空了就堵塞 200 while (pool->taskqueue->GetTaskNum() == 0 && !pool->shutdown) 201 { 202 cout << "thread :" << to_string(pthread_self()) << " waiting..." << endl; 203 pthread_cond_wait(&pool->notempty, &pool->mutex_pool); 204 205 //解除后 判断是否要销毁进程 206 if (pool->exit_num > 0) 207 { 208 pool->exit_num--; 209 if (pool->live_num > pool->min_num) 210 { 211 pool->live_num--; 212 pthread_mutex_unlock(&pool->mutex_pool); 213 pool->Thread_Exit(); 214 } 215 } 216 } 217 218 //判断线程池是否要关闭了 219 if (pool->shutdown) 220 { 221 pthread_mutex_unlock(&pool->mutex_pool); 222 pool->Thread_Exit(); 223 } 224 225 //从任务队列取出任务 226 Task<T> task = pool->taskqueue->TakeTask(); 227 pool->busy_num++; 228 pthread_mutex_unlock(&pool->mutex_pool); 229 230 cout << "thread :" << to_string(pthread_self()) << " start working..." << endl; 231 task.function(task.args); 232 233 delete task.args; 234 task.args = nullptr; 235 236 //任务结束 237 cout << "thread :" << to_string(pthread_self()) << " end working..." << endl; 238 pthread_mutex_lock(&pool->mutex_pool); 239 pool->busy_num--; 240 pthread_mutex_unlock(&pool->mutex_pool); 241 242 } 243 return nullptr; 244 } 245 246 //管理者线程任务函数 247 static void* Manage(void* args) 248 { 249 ThreadPool* pool = static_cast<ThreadPool*>(args); 250 while (!pool->shutdown) 251 { 252 //5秒检测一次 253 sleep(5); 254 pthread_mutex_lock(&pool->mutex_pool); 255 int livenum = pool->live_num; 256 int busynum = pool->busy_num; 257 int queuesize = pool->taskqueue->GetTaskNum(); 258 pthread_mutex_unlock(&pool->mutex_pool); 259 260 const int NUMBER = 2; 261 //创建 262 if (queuesize > livenum && livenum < pool->max_num) 263 { 264 pthread_mutex_lock(&pool->mutex_pool); 265 int num = 0; 266 for (int i = 0; i < pool->max_num && 267 num < NUMBER && 268 pool->live_num < pool->max_num ; ++i) 269 { 270 if (pool->threadID[i] == 0) 271 { 272 pthread_create(&pool->threadID[i], NULL, Work, pool); 273 num++; 274 pool->live_num++; 275 } 276 } 277 pthread_mutex_unlock(&pool->mutex_pool); 278 } 279 280 //销毁 281 if (busynum * 2 < livenum && livenum > pool->min_num) 282 { 283 pthread_mutex_lock(&pool->mutex_pool); 284 pool->exit_num = NUMBER; 285 pthread_mutex_unlock(&pool->mutex_pool); 286 for (int i = 0; i < NUMBER; ++i) 287 { 288 pthread_cond_signal(&pool->notempty); 289 } 290 } 291 } 292 return nullptr; 293 } 294 295 void Thread_Exit() 296 { 297 pthread_t tid = pthread_self(); 298 for (int i = 0; i < max_num; ++i) 299 { 300 if (threadID[i] == tid) 301 { 302 cout << "thread :" << to_string(pthread_self()) << "exiting" << endl; 303 threadID[i] = 0; 304 break; 305 } 306 } 307 pthread_exit(NULL); 308 } 309private: 310 pthread_mutex_t mutex_pool; 311 pthread_cond_t notempty; 312 pthread_t* threadID; 313 pthread_t managerID; 314 TaskQueue<T>* taskqueue; 315 int min_num; 316 int max_num; 317 int busy_num; 318 int live_num; 319 int exit_num; 320 bool shutdown = false; 321 322};
3.2 测试部分
1#include"ThreadPool.h" 2 3void Task_Test(void* args) 4{ 5 int num = *(int*)args; 6 cout<<"thread :" << pthread_self() << " is working " << "number =" << num <<endl; 7 sleep(1); 8 return; 9} 10 11int main() 12{ 13 //创建线程池 14 ThreadPool<int> pool(3, 10); 15 for (int i = 0; i < 100; ++i) 16 { 17 int* num = new int(i+100); 18 pool.Add_Task(Task<int>(Task_Test,num)); 19 } 20 sleep(40); 21 return 0; 22} 23 24
以上只是基于C修改出对应于C++的代码
并且以上代码存在一个问题 输出的结果有时会因为线程原因出现混乱 可以通过加锁来解决,但锁的数量超过1就容易导致死锁问题,所以暂且搁置
4. C++11实现线程池
并非原创,摘于此处
4.1 头文件
1#pragma once 2#include<queue> 3#include<thread> 4#include<condition_variable> 5#include<atomic> 6#include<stdexcept> 7#include<future> 8#include<vector> 9#include<functional> 10 11namespace std 12{ 13 #define THREADPOOL_MAX_NUM 16 14 class threadpool 15 { 16 unsigned short _initsize; 17 using Task = function<void()>; 18 vector<thread> _pool; 19 queue<Task> _tasks; 20 mutex _lock; 21 mutex _lockGrow; 22 condition_variable _task_cv; 23 atomic<bool> _run{true}; 24 atomic<int> _spa_trd_num{0}; 25 26 public: 27 inline threadpool(unsigned short size = 4) 28 { 29 _initsize = size; 30 Add_Thread(size); 31 } 32 inline ~threadpool() 33 { 34 _run = false; 35 _task_cv.notify_all(); 36 for (thread& thread : _pool) 37 { 38 if (thread.joinable()) 39 thread.join(); 40 } 41 } 42 43 template<typename F,typename... Args> 44 auto commit(F&& f, Args&& ...args) -> future<decltype(f(args...)) > 45 { 46 if (!_run) 47 throw runtime_error{"commit auto stop"}; 48 using RetType = decltype(f(args...)); 49 auto task = make_shared<packaged_task<RetType()>>(bind(forward<F>(f), forward<Args>(args)...)); 50 future<RetType> future = task->get_future(); 51 { 52 lock_guard<mutex> lock{_lock}; 53 _tasks.emplace([task]() {(*task)(); }); 54 } 55 if (_spa_trd_num < 1 && _pool.size() < THREADPOOL_MAX_NUM) 56 Add_Thread(1); 57 _task_cv.notify_one(); 58 return future; 59 } 60 61 template<typename F> 62 void commit2(F&& f) 63 { 64 if (!_run) 65 return; 66 { 67 lock_guard<mutex> lock{_lock}; 68 _tasks.emplace(forward<F>(f)); 69 } 70 if (_spa_trd_num < 1 && _pool.size() < THREADPOOL_MAX_NUM) 71 Add_Thread(1); 72 _task_cv.notify_one(); 73 } 74 75 int idlCount() { return _spa_trd_num; } 76 int thrCount() { return _pool.size(); } 77 78 private: 79 void Add_Thread(unsigned short size) 80 { 81 if (!_run) 82 throw runtime_error{"Add_Thread stop"}; 83 unique_lock<mutex> lockgrow{_lockGrow}; 84 for (; _pool.size() < THREADPOOL_MAX_NUM && size > 0; --size) 85 { 86 _pool.emplace_back([this] 87 { 88 while (true) 89 { 90 Task task; 91 { 92 unique_lock<mutex> lock{_lock}; 93 _task_cv.wait(lock, [this] {return !_run || !_tasks.empty(); }); 94 if (!_run && _tasks.empty()) 95 return; 96 _spa_trd_num--; 97 task = move(_tasks.front()); 98 _tasks.pop(); 99 } 100 task(); 101 if (_spa_trd_num > 0 && _pool.size() > _initsize) 102 return; 103 { 104 unique_lock<mutex> lock{_lock}; 105 _spa_trd_num++; 106 } 107 } 108 }); 109 { 110 unique_lock<mutex> lock{_lock}; 111 _spa_trd_num++; 112 } 113 } 114 } 115 }; 116}
要使用pthread依赖库
4.2 测试部分
1#include"ThreadPool.hpp" 2#include<iostream> 3 4void fun1(int slp) 5{ 6 printf("fun1 %ld\n", std::this_thread::get_id()); 7 if (slp > 0) 8 { 9 printf("fun1 sleep %ld ========= %ld\n", slp, std::this_thread::get_id()); 10 std::this_thread::sleep_for(std::chrono::milliseconds(slp)); 11 } 12} 13 14struct gfun 15{ 16 int operator()(int n) 17 { 18 printf("gfun %ld\n", n, std::this_thread::get_id()); 19 return 42; 20 } 21}; 22 23class A 24{ 25public: 26 static int Afun(int n = 0) //函数必须是 static 的才能直接使用线程池 27 { 28 std::cout << n << "Afun " << std::this_thread::get_id() << std::endl; 29 return n; 30 } 31 32 static std::string Bfun(int n, std::string str, char c) 33 { 34 std::cout << n << "Bfun " << str.c_str() << " " << (int)c << " " << std::this_thread::get_id() << std::endl; 35 return str; 36 } 37}; 38 39int main() 40try { 41 std::threadpool executor{ 50 }; 42 std::future<void> ff = executor.commit(fun1, 0); 43 std::future<int> fg = executor.commit(gfun{}, 0); 44 //std::future<int> gg = executor.commit(A::Afun, 9999); //IDE提示错误,但可以编译运行 45 std::future<std::string> gh = executor.commit(A::Bfun, 9998, "mult args", 123); 46 std::future<std::string> fh = executor.commit([]()->std::string { std::cout << "hello, fh ! " << std::this_thread::get_id() << std::endl; return "hello,fh ret !\n"; }); 47 48 std::this_thread::sleep_for(std::chrono::seconds(1)); 49 std::cout << fg.get() << " " << fh.get().c_str() << " " << std::this_thread::get_id() << std::endl; 50 51 std::cout << " ======= fun1,55 ========= " << std::this_thread::get_id() << std::endl; 52 executor.commit(fun1, 55).get(); //调用.get()获取返回值会等待线程执行完 53 54 std::threadpool pool(4); 55 std::vector< std::future<int> > results; 56 57 for (int i = 0; i < 8; ++i) 58 { 59 results.emplace_back( 60 pool.commit([i] { 61 std::cout << "hello " << i << std::endl; 62 std::this_thread::sleep_for(std::chrono::seconds(3)); 63 std::cout << "world " << i << std::endl; 64 return i * i; 65 }) 66 ); 67 } 68 std::this_thread::sleep_for(std::chrono::seconds(15)); 69 for (auto&& result : results) 70 std::cout << result.get() << ' '; 71 std::cout << std::endl; 72 return 0; 73} 74catch (std::exception& e) 75{ 76 std::cout << "some error " << std::this_thread::get_id() << e.what() << std::endl; 77}
- 测试结果
