线程池创建销毁（c++销毁线程）

本文实例介绍了C 语言实现线程池，支持动态拓展和销毁，分享给大家供大家参考，具体内容如下

实现功能

1.初始化指定个数的线程
2.使用链表来管理任务队列
3.支持拓展动态线程
4.如果闲置线程过多,动态销毁部分线程

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 #include <stdio.h> #include <pthread.h> #include <stdlib.h> #include <signal.h> /*线程的任务队列由,函数和参数组成,任务由链表来进行管理*/ typedef struct thread_worker_s{ void *(*process)(void *arg); //处理函数 void *arg; //参数 struct thread_worker_s *next; }thread_worker_t; #define bool int #define true 1 #define false 0 /*线程池中各线程状态描述*/ #define THREAD_STATE_RUN 0 #define THREAD_STATE_TASK_WAITING 1 #define THREAD_STATE_TASK_PROCESSING 2 #define THREAD_STATE_TASK_FINISHED 3 #define THREAD_STATE_EXIT 4 typedef struct thread_info_s{ pthread_t id; int state; struct thread_info_s *next; }thread_info_t; static char* thread_state_map[] ={"创建","等待任务","处理中","处理完成","已退出"}; /*线程压缩的时候只有 0,1,2,4 状态的线程可以销毁*/ /*线程池管理器*/ #define THREAD_BUSY_PERCENT 0.5 /*线程:任务 = 1:2 值越小，说明任务多,增加线程*/ #define THREAD_IDLE_PERCENT 2 /*线程:任务 = 2:1 值大于1,线程多于任务,销毁部分线程*/ typedef struct thread_pool_s{ pthread_mutex_t queue_lock ; //队列互斥锁,即涉及到队列修改时需要加锁 pthread_cond_t queue_ready; //队列条件锁,队列满足某个条件，触发等待这个条件的线程继续执行,比如说队列满了,队列空了 thread_worker_t *head ; //任务队列头指针 bool is_destroy ; //线程池是否已经销毁 int num; //线程的个数 int rnum; ; //正在跑的线程 int knum; ; //已杀死的线程 int queue_size ; //工作队列的大小 thread_info_t *threads ; //线程组id,通过pthread_join(thread_ids[0],NULL) 来执行线程 pthread_t display ; //打印线程 pthread_t destroy ; //定期销毁线程的线程id pthread_t extend ; float percent ; //线程个数于任务的比例 rnum/queue_size int init_num ; pthread_cond_t extend_ready ; //如果要增加线程 }thread_pool_t; /*-------------------------函数声明----------------------*/ /** * 1.初始化互斥变量 * 2.初始化等待变量 * 3.创建指定个数的线程线程 */ thread_pool_t* thread_pool_create(int num); void *thread_excute_route(void *arg); /*调试函数*/ void debug(char *message,int flag){ if(flag) printf("%s\n",message); } void *display_thread(void *arg); /** * 添加任务包括以下几个操作 * 1.将任务添加到队列末尾 * 2.通知等待进程来处理这个任务 pthread_cond_singal(); */ int thread_pool_add_worker(thread_pool_t *pool,void*(*process)(void *arg),void *arg); //网线程池的队列中增加一个需要执行的函数，也就是任务 /** * 销毁线程池,包括以下几个部分 * 1.通知所有等待的进程 pthread_cond_broadcase * 2.等待所有的线程执行完 * 3.销毁任务列表 * 4.释放锁，释放条件 * 4.销毁线程池对象 */ void *thread_pool_is_need_recovery(void *arg); void *thread_pool_is_need_extend(void *arg); void thread_pool_destory(thread_pool_t *pool); thread_pool_t *thread_pool_create(int num){ if(num<1){ return NULL; } thread_pool_t *p; p = (thread_pool_t*)malloc(sizeof(struct thread_pool_s)); if(p==NULL) return NULL; p->init_num = num; /*初始化互斥变量与条件变量*/ pthread_mutex_init(&(p->queue_lock),NULL); pthread_cond_init(&(p->queue_ready),NULL); /*设置线程个数*/ p->num = num; p->rnum = num; p->knum = 0; p->head = NULL; p->queue_size =0; p->is_destroy = false; int i=0; thread_info_t *tmp=NULL; for(i=0;i<num;i++){ /*创建线程*/ tmp= (struct thread_info_s*)malloc(sizeof(struct thread_info_s)); if(tmp==NULL){ free(p); return NULL; }else{ tmp->next = p->threads; p->threads = tmp; } pthread_create(&(tmp->id),NULL,thread_excute_route,p); tmp->state = THREAD_STATE_RUN; } /*显示*/ pthread_create(&(p->display),NULL,display_thread,p); /*检测是否需要动态线程*/ //pthread_create(&(p->extend),NULL,thread_pool_is_need_extend,p); /*动态销毁*/ pthread_create(&(p->destroy),NULL,thread_pool_is_need_recovery,p); return p; } int thread_pool_add_worker(thread_pool_t *pool,void*(*process)(void*arg),void*arg){ thread_pool_t *p= pool; thread_worker_t *worker=NULL,*member=NULL; worker = (thread_worker_t*)malloc(sizeof(struct thread_worker_s)); int incr=0; if(worker==NULL){ return -1; } worker->process = process; worker->arg = arg; worker->next = NULL; thread_pool_is_need_extend(pool); pthread_mutex_lock(&(p->queue_lock)); member = p->head; if(member!=NULL){ while(member->next!=NULL) member = member->next; member->next = worker; }else{ p->head = worker; } p->queue_size ++; pthread_mutex_unlock(&(p->queue_lock)); pthread_cond_signal(&(p->queue_ready)); return 1; } void thread_pool_wait(thread_pool_t *pool){ thread_info_t *thread; int i=0; for(i=0;i<pool->num;i++){ thread = (thread_info_t*)(pool->threads+i); thread->state = THREAD_STATE_EXIT; pthread_join(thread->id,NULL); } } void thread_pool_destory(thread_pool_t *pool){ thread_pool_t *p = pool; thread_worker_t *member = NULL; if(p->is_destroy) return ; p->is_destroy = true; pthread_cond_broadcast(&(p->queue_ready)); thread_pool_wait(pool); free(p->threads); p->threads = NULL; /*销毁任务列表*/ while(p->head){ member = p->head; p->head = member->next; free(member); } /*销毁线程列表*/ thread_info_t *tmp=NULL; while(p->threads){ tmp = p->threads; p->threads = tmp->next; free(tmp); } pthread_mutex_destroy(&(p->queue_lock)); pthread_cond_destroy(&(p->queue_ready)); return ; } /*通过线程id,找到对应的线程*/ thread_info_t *get_thread_by_id(thread_pool_t *pool,pthread_t id){ thread_info_t *thread=NULL; thread_info_t *p=pool->threads; while(p!=NULL){ if(p->id==id) return p; p = p->next; } return NULL; } /*每个线程入口函数*/ void *thread_excute_route(void *arg){ thread_worker_t *worker = NULL; thread_info_t *thread = NULL; thread_pool_t* p = (thread_pool_t*)arg; //printf("thread %lld create success\n",pthread_self()); while(1){ pthread_mutex_lock(&(p->queue_lock)); /*获取当前线程的id*/ pthread_t pthread_id = pthread_self(); /*设置当前状态*/ thread = get_thread_by_id(p,pthread_id); /*线程池被销毁,并且没有任务了*/ if(p->is_destroy==true && p->queue_size ==0){ pthread_mutex_unlock(&(p->queue_lock)); thread->state = THREAD_STATE_EXIT; p->knum ++; p->rnum --; pthread_exit(NULL); } if(thread){ thread->state = THREAD_STATE_TASK_WAITING; /*线程正在等待任务*/ } /*线程池没有被销毁,没有任务到来就一直等待*/ while(p->queue_size==0 && !p->is_destroy){ pthread_cond_wait(&(p->queue_ready),&(p->queue_lock)); } p->queue_size--; worker = p->head; p->head = worker->next; pthread_mutex_unlock(&(p->queue_lock)); if(thread) thread->state = THREAD_STATE_TASK_PROCESSING; /*线程正在执行任务*/ (*(worker->process))(worker->arg); if(thread) thread->state = THREAD_STATE_TASK_FINISHED; /*任务执行完成*/ free(worker); worker = NULL; } } /*拓展线程*/ void *thread_pool_is_need_extend(void *arg){ thread_pool_t *p = (thread_pool_t *)arg; thread_pool_t *pool = p; /*判断是否需要增加线程,最终目的线程:任务=1:2*/ if(p->queue_size>100){ int incr =0; if(((float)p->rnum/p->queue_size) < THREAD_BUSY_PERCENT ){ incr = (p->queue_size*THREAD_BUSY_PERCENT) - p->rnum; /*计算需要增加线程个数*/ int i=0; thread_info_t *tmp=NULL; thread_pool_t *p = pool; pthread_mutex_lock(&pool->queue_lock); if(p->queue_size<100){ pthread_mutex_unlock(&pool->queue_lock); return ; } for(i=0;i<incr;i++){ /*创建线程*/ tmp= (struct thread_info_s*)malloc(sizeof(struct thread_info_s)); if(tmp==NULL){ continue; }else{ tmp->next = p->threads; p->threads = tmp; } p->num ++; p->rnum ++; pthread_create(&(tmp->id),NULL,thread_excute_route,p); tmp->state = THREAD_STATE_RUN; } pthread_mutex_unlock(&pool->queue_lock); } } //pthread_cond_signal(&pool->extend_ready); } pthread_cond_t sum_ready; /*恢复初始线程个数*/ void *thread_pool_is_need_recovery(void *arg){ thread_pool_t *pool = (thread_pool_t *)arg; int i=0; thread_info_t *tmp = NULL,*prev=NULL,*p1=NULL; /*如果没有任务了,当前线程大于初始化的线程个数*/ while(1){ i=0; if(pool->queue_size==0 && pool->rnum > pool->init_num ){ sleep(5); /*5s秒内还是这个状态的话就,销毁部分线程*/ if(pool->queue_size==0 && pool->rnum > pool->init_num ){ pthread_mutex_lock(&pool->queue_lock); tmp = pool->threads; while((pool->rnum != pool->init_num) && tmp){ /*找到空闲的线程*/ if(tmp->state != THREAD_STATE_TASK_PROCESSING){ i++; if(prev) prev->next = tmp->next; else pool->threads = tmp->next; pool->rnum --; /*正在运行的线程减一*/ pool->knum ++; /*销毁的线程加一*/ kill(tmp->id,SIGKILL); /*销毁线程*/ p1 = tmp; tmp = tmp->next; free(p1); continue; } prev = tmp; tmp = tmp->next; } pthread_mutex_unlock(&pool->queue_lock); printf("5s内没有新任务销毁部分线程,销毁了 %d 个线程\n",i); } } sleep(5); } } /*打印一些信息的*/ void *display_thread(void *arg){ thread_pool_t *p =(thread_pool_t *)arg; thread_info_t *thread = NULL; int i=0; while(1){ printf("threads %d,running %d,killed %d\n",p->num,p->rnum,p->knum); /*线程总数,正在跑的,已销毁的*/ thread = p->threads; while(thread){ printf("id=%ld,state=%s\n",thread->id,thread_state_map[thread->state]); thread = thread->next; } sleep(5); } }