基于的Linux线程池

/* 小规模的Linux线程池 written by luoxiongwei E-mail:[email protected] Welcom to report bugs! 下一步计划给工作队列中的任务添加优先级 实现一个优先队列。 */ #ifndef __THREADPOOLS #define __THREADPOOLS #include <sys/types.h> #include <pthread.h> #include <unistd.h> #include <vector> #include <cstdio> #include <cstdlib> #include <cstring> using std::vector; struct task//队列中的任务结点 { void (*handler)(void *);//任务函数 void * data;//任务数据 //int prority;//任务优先级 }; struct work_queue_t//工作队列 { struct work_queue_t * next ; struct work_queue_t * head ; struct work_queue_t * tail ; struct task * elment ; }; //工作队列操作 work_queue_t * init_work_queue();//初始化工作队列 void destory_work_queue();//销毁工作队列 work_queue_t * remove_task();//取队首任务 int add_tast(task * new_task);//在工作队列尾部添加任务 bool empty(); //线程池资源 void *thread_rounter(void *);//线程执行函数 struct work_queue_t * work_queue=NULL;//工作队列 int work_count=0;//工作队列中的元素个数 vector<pthread_t> threads;//记录所有线程的id pthread_cond_t pool_cond;//唤醒相应线程开始工作 pthread_mutex_t queue_lock;//保护工作队列,以及池中的互斥资源 int idles;//空闲线程数目 int busy;//正在执行任务的线程数目 int max_thread;//池中维持的最大线程数目 bool to_exit;//销毁线程池标志 bool validate;//线程池是否可用 //线程池操作,提供给用户操作的四个函数 void thread_pool(int init_threads_num);//建立线程池 void destroy_thread_pool();//销毁线程池 /*组装任务处理函数和处理函数参数，注意一旦该任务完成之后， 数据将会被系统释放掉，所以下一次向线程池加入任务时， 请重新组装。请保证处理函数是线程可重入的。 */ task * prod_task(void (*handler)(void *), void * data); int exec_task(task * new_task);//留给用户的调用接口 #endif //示例的工作函数 void task1(void *arg); int main() { //现在测试了池中有10个线程，12个任务。 //任务参数 char *ptr1="I am task one\n"; char *ptr2="I am task two\n"; char *ptr3="I am task three\n"; char *ptr4="I am task four\n"; char *ptr5="I am task five\n"; char *ptr6="I am task six\n"; char *ptr7="I am task seven\n"; char *ptr8="I am task eight\n"; char *ptr9="I am task nine\n"; char *ptr10="I am task ten\n"; char *ptr11="I am task eleven\n"; char *ptr12="I am task twltle\n"; //组装任务 task * task_p1=prod_task(task1,ptr1); task * task_p2=prod_task(task1,ptr2); task * task_p3=prod_task(task1,ptr3); task * task_p4=prod_task(task1,ptr4); task * task_p5=prod_task(task1,ptr5); task * task_p6=prod_task(task1,ptr6); task * task_p7=prod_task(task1,ptr7); task * task_p8=prod_task(task1,ptr8); task * task_p9=prod_task(task1,ptr9); task * task_p10=prod_task(task1,ptr10); task * task_p11=prod_task(task1,ptr11); task * task_p12=prod_task(task1,ptr12); thread_pool(10);//建立线程池,池中最多10个线程 //12个任务 exec_task(task_p1); exec_task(task_p2); exec_task(task_p3); exec_task(task_p4); exec_task(task_p5); exec_task(task_p6); exec_task(task_p7); exec_task(task_p8); exec_task(task_p9); exec_task(task_p10); exec_task(task_p11); exec_task(task_p12); /*延时，让所有任务执行完。也可以不延时， 但是可能任务还没有完成，线程池就被销毁。 */ sleep(20); destroy_thread_pool();//销毁线程池 return 0; } void task1(void *arg) { char *ptr=(char *)arg; write(STDOUT_FILENO,ptr,strlen(ptr)); } //提供给用户的外部接口之一，生成任务 task * prod_task(void (*handler)(void *), void * data) { struct task * new_task=new task; if(!new_task) return NULL; new_task->handler=handler; new_task->data=data; return new_task; } //工作队列的相应操作 work_queue_t * init_work_queue() { work_queue_t *work_queue_=new work_queue_t ; if(!work_queue_) return NULL; work_queue_->head=work_queue_->tail=NULL; work_queue_->next=NULL; work_queue_->elment=NULL; work_count=0; return work_queue_; } work_queue_t * remove_task()//取队首任务 { work_queue_t * temp; if(empty())//工作队列空 return NULL; temp=work_queue->head; work_queue->head=work_queue->head->next; work_count--; return temp; } //在工作队列尾部添加任务 int add_tast(task * new_task) { work_queue_t * temp=NULL; if(empty())//队列为空 { temp=new work_queue_t ; temp->next=NULL; temp->elment=new_task; work_queue->head=work_queue->tail=temp; work_count++; return 0; } else//队列不空 { temp=new work_queue_t ; temp->next=NULL; temp->elment=new_task; work_queue->tail->next=temp; work_queue->tail=temp; work_count++; return 0; } return -1;//添加不成功 } void destory_work_queue() { work_queue_t * temp; if(work_count!=0) { while(work_queue->head) { temp=work_queue->head; work_queue->head=work_queue->head->next; delete temp->elment; delete temp; } } } bool empty() { bool flage=false; if(work_count==0) flage=true; return flage; } //线程池的操作 //初始化线程池 void thread_pool(int init_threads_num) { pthread_mutex_init(&queue_lock,NULL);//初始化锁 if(pthread_cond_init(&pool_cond,NULL)!=0) return ; work_queue=init_work_queue();//工作队列 if(!work_queue) return ; max_thread=init_threads_num; idles=0; busy=0; to_exit=false; validate=true; } //销毁线程池和工作队列,条件变量，锁 void destroy_thread_pool() { pthread_mutex_lock(&queue_lock); //改变条件 to_exit=true; validate=false; //取消线程池中的所有线程 for(size_t i=0;i!=threads.size();++i) pthread_cancel(threads[i]); //销毁工作队列 destory_work_queue(); pthread_mutex_unlock(&queue_lock); //销毁锁和条件变量 pthread_mutex_destroy(&queue_lock); pthread_cond_destroy(&pool_cond); } int exec_task(task * new_task) { pthread_t tid; pthread_mutex_lock(&queue_lock); //将新任务添加至工作队列 if(add_tast(new_task)!=0) { pthread_mutex_unlock(&queue_lock); return -1; } pthread_cond_signal(&pool_cond);//工作队列中有任务了，唤醒一个阻塞线程 //是否新建线程，视条件而定 if(validate&&!to_exit&&(idles==0||(idles+busy)<max_thread)) { if(pthread_create(&tid,NULL,thread_rounter,NULL)!=0) { pthread_mutex_unlock(&queue_lock); return -1; } threads.push_back(tid);//记录线程id idles++;//多了一个空闲线程 pthread_mutex_unlock(&queue_lock); return 0; } pthread_mutex_unlock(&queue_lock); return 0; } void *thread_rounter(void *) { pthread_detach(pthread_self());//分离自己 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL); pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED,NULL);//到达取消点处才取消线程 work_queue_t * work_item=NULL; for(;;)//去检查工作队列中是否有需要处理的工作项目 { pthread_mutex_lock(&queue_lock); while(validate&&!to_exit&&((work_item=remove_task())==NULL))//工作队列为空,则一直等待 pthread_cond_wait(&pool_cond,&queue_lock);//这里将会是一个取消点 pthread_mutex_unlock(&queue_lock); if(validate&&!to_exit&&work_item->elment->data) { pthread_mutex_lock(&queue_lock); busy++;//该线程忙 idles--; pthread_mutex_unlock(&queue_lock); work_item->elment->handler(work_item->elment->data);//处理函数调�