汕头高端网站建设,北京市中海建设有限公司网站,如何做图片网站,株洲网站建设的企业一.线程池开发框架
我所开发的线程池由以下几部分组成#xff1a; 1.工作中的线程。也就是线程池中的线程#xff0c;主要是执行分发来的task。 2.管理线程池的监督线程。这个线程的创建独立于线程池的创建#xff0c;按照既定的管理方法进行管理线程池中的所有线程#xf…一.线程池开发框架
我所开发的线程池由以下几部分组成 1.工作中的线程。也就是线程池中的线程主要是执行分发来的task。 2.管理线程池的监督线程。这个线程的创建独立于线程池的创建按照既定的管理方法进行管理线程池中的所有线程主要任务是监听任务的到来唤醒线程池中的空闲线程分发任务如果任务增多动态的创建一批线程加入原来的线程池中进行工作适当的销毁线程减少系统开销。
这个线程池的开发涉及了以下几个数据结构、设计模式和软件结构 1.任务队列。整个框架有两个任务队列1等待任务队列以下简称wait_queue。2正在执行中任务队列以下简称doing_queue。队列采用先进先出的数据结构。当一个任务来到时会先被push到wait_queue监督线程会一直监督wait_queue中的元素一旦有任务便会pop wait_queue中的元素再push到doing_queue中。 2.单例设计模式。线程池的类被设计成单例模式防止一个程序中多次创建线程池对象出现紊乱现象用户只能调用静态方法初始化得到线程池的对象。 3.回调函数。回调函数的设计主要是为了能够把任务接口也就是需要线程去执行的任务通常是一个写好的函数方法提前初始化注册然后延迟调用。
下图是所用类的的大概结构图
程序整体结构如下
二.线程池开发具体实现
1.思路分析。 线程池顾名思义就是同时有数个线程处于待执行状态编码上的初始化的实现无非就是循环创建指定数量的线程然后等待任务的到来唤醒空闲线程。以下是ThreadPoll的类
class ThreadPool
{private:pthread_t *_thread; //线程池pthread_t *_thread_bak; //备用线程池当任务过多会自动创建pthread_t taskqueue_thread; //管理线程int u4sequence;int wait_time;int CANCEL_SIGNAL;bool F_improve_ThrdPoll; //备用线程池创建标志Mutex *mutex; //互斥锁CondThread *task_cond; //条件变量TaskFuncCallback callback; //声明回调函数即线程所需要执行的函数int _num_threads; //线程池数量//构造函数的实现为private属性禁止用户用构造函数初始化对象。ThreadPool(int num_threads):_num_threads(num_threads),F_improve_ThrdPoll(0),wait_time(0),u4sequence(0),CANCEL_SIGNAL(0){init(); //一些变量的创建ManagerThreadInit(); //创建管理线程ThreadPoolInit(num_threads);//初始化线程池}public:LVQueueTASK_QUEUE_T task_wait_queue;//创建任务等待队列LVQueueTASK_QUEUE_T task_doing_queue;//创建任务执行队列~ThreadPool(){delete(mutex);delete(task_cond);delete(_thread);delete(_thread_bak);}//用户通过调用此静态方法得到线程池的对象单例模式static ThreadPool* createThreadPool(int num){ static ThreadPool *_pool new ThreadPool(num);return _pool;}void init(){_thread new pthread_t[_num_threads];mutex new Mutex();task_cond new CondThread();}API_RETURN_TYPE_T ThreadPoolInit(int num_thr);//线程池初始化核心接口API_RETURN_TYPE_T run(); //线程执行函数API_RETURN_TYPE_T ManagerThreadInit();//管理线程初始化API_RETURN_TYPE_T managerThread();线程执行函数API_RETURN_TYPE_T wakeupThread(TaskFuncCallback p_func);//用户调用此接口唤醒线程执行任务参数为传入的任务执行函数地址API_RETURN_TYPE_T AutoComputeOptimumThreadNum(int wait_que_num,int _u4sequence);//一种自动计算需要增加多少线程到线程池当任务繁多时会调到。API_RETURN_TYPE_T ThreadJoin();//所有线程阻塞API_RETURN_TYPE_T ReleaseSubThreadPool();//释放备用线程池API_RETURN_TYPE_T DestroyThreadPool();//释放线程池
};12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
下面是cpp的实现
//线程池初始化的实现
API_RETURN_TYPE_T ThreadPool::ThreadPoolInit(int num_thr)
{printf(num %d.\n,num_thr); if(num_thr 0){return API_SUCCESS;}//设置创建线程的属性为DETACHED线程被释放后资源会被回收。pthread_attr_t attr;pthread_attr_init (attr);pthread_attr_setdetachstate (attr, PTHREAD_CREATE_DETACHED);int i 0;if(F_improve_ThrdPoll 1)//备用线程池创建的标志位初始化线程不会走这边{_thread_bak new pthread_t[num_thr];for(;i num_thr;i){if(RET_OK ! pthread_create((_thread_bak[i]), attr, thread_func, this)){return API_FAIL;}}return API_SUCCESS;}//create thread pool.for(;i num_thr;i){if(RET_OK ! pthread_create((_thread[i]), attr, thread_func, this)){return API_FAIL;}}pthread_attr_destroy (attr);return API_SUCCESS;
}
//创建的所有线程都会跑到这个线程函数最终指向run
void *thread_func(void *arg)
{ThreadPool *thread (ThreadPool*)arg;thread-run();
}
//线程池核心内容
API_RETURN_TYPE_T ThreadPool::run()
{//printf(this is run thread.\n);void *arg;while(1)//线程池内部一直在循环{printf (thread 0x%x begin\n, pthread_self ()); this-mutex-lock();//上锁if((CANCEL_SIGNAL 0) (task_doing_queue.length() _num_threads || F_improve_ThrdPoll 1) )//以上条件第一个是备用线程释放标志第二个是任务执行队列数量为0第三个是备用线程创建标志或的关系为了满足新增线程进入wait状态第一次这些条件都会满足{printf (thread 0x%x is waiting\n, pthread_self ()); this-task_cond-wait(mutex);//每次创建的新线程都会阻塞到这里执行完任务的线程也会阻塞在这里等待唤醒的signal虽然是阻塞在这里但是互斥锁已经是unlock状态了这是linux的机制。}usleep(200000);this-mutex-unlock();//解锁pthread_testcancel();//设置取消线程点pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL);//1头尾保护下面这段code保证在执行任务的时候屏蔽外部的线程取消信号if(callback ! NULL){callback(arg); //执行回调函数此时的回调函数应该指向当前任务执行函数的地址callback NULL;}task_doing_queue.popFront();//执行完任务任务执行队列出队列元素一个pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);//同1printf(wait len %d.\n,task_wait_queue.length());printf(thread 0x%x done.length() %d.\n,pthread_self (),task_doing_queue.length());}return API_SUCCESS;}//管理线程的初始化
API_RETURN_TYPE_T ThreadPool::ManagerThreadInit()
{//create manager threadpool thread.if(RET_OK ! pthread_create(taskqueue_thread, NULL, thread_task_queue, this)){ return API_FAIL;}return API_SUCCESS;
}
//管理线程的执行函数
void *thread_task_queue(void *arg)
{ ThreadPool *thread (ThreadPool*)arg;thread-managerThread();}
//管理线程的核心内容
API_RETURN_TYPE_T ThreadPool::managerThread()
{while(1){usleep(400000);printf(managerThread!.\n);this-mutex-lock();//上锁TASK_QUEUE_T task1; //初始化两个队列元素对象TASK_QUEUE_T task2;task1.sTask TASK_DOING;if(task_wait_queue.length() ! 0){//printf(len %d.\n,task_doing_queue.length());if(task_doing_queue.length() _num_threads)//只要任务执行队列的数目小于线程池中线程的总数就会执行{task2 task_wait_queue.popFront();//pop任务等待队列的元素并得到这个元素的对象callback task2.cTaskFunc;//获得任务的执行函数地址task_doing_queue.pushBack(task1);//将任务push到任务执行队列task_cond-signal();//发送信号唤醒一个空闲线程printf(signal cond.\n);}}//当人任务队列的等待任务数量大于线程池线程总数时会执行if(task_wait_queue.length() _num_threads F_improve_ThrdPoll 0){//通过简单的机制计算当前是否需要另外新增线程到线程池AutoComputeOptimumThreadNum(task_wait_queue.length(),u4sequence);F_improve_ThrdPoll 1;ThreadPoolInit(u4sequence);//如果需要新增线程u4sequence不为0. sleep(2);//缓冲线程创建}if(F_improve_ThrdPoll 1 ){//检测到备用线程池的创建while(task_wait_queue.length() 0 task_doing_queue.length() 0){//也就是当前任务等待队列和任务执行队列都没有任务时printf(no task!.\n);usleep(500000);wait_time;//计时等待一段时间if(wait_time NO_TASK_TIMEOUT){this-mutex-unlock();ReleaseSubThreadPool();//释放备用线程池printf(release!.\n);F_improve_ThrdPoll 2;wait_time 0;break;}}wait_time 0;}if(F_improve_ThrdPoll ! 2)this-mutex-unlock();}return API_SUCCESS;}//自动计算是否需要创建新的线程池的简单机制后续会结合读取当前CPU的使用率进一步优化此机制
API_RETURN_TYPE_T ThreadPool::AutoComputeOptimumThreadNum(int wait_que_num,int _u4sequence)
{if(wait_que_num 4*_num_threads){_u4sequence _num_threads;}else if(wait_que_num 2*_num_threads){_u4sequence _num_threads/2;}else{_u4sequence 0;}return API_SUCCESS;
}//释放备用线程池待优化API_RETURN_TYPE_T ThreadPool::ReleaseSubThreadPool()
{this-mutex-lock();CANCEL_SIGNAL 1;this-mutex-unlock();task_cond-broadcast();for(int i 0;i _num_threads;i){if(RET_OK ! pthread_cancel(_thread_bak[i])){return API_FAIL;}}this-mutex-lock();printf(4444.\n);CANCEL_SIGNAL 0;this-mutex-unlock();return API_SUCCESS;
}
//摧毁线程池,待优化
API_RETURN_TYPE_T ThreadPool::DestroyThreadPool()
{//first ,destroy manager thread.if(RET_OK ! pthread_cancel(taskqueue_thread)){return API_FAIL;}return API_SUCCESS;
}API_RETURN_TYPE_T ThreadPool::ThreadJoin()
{for(int i 0;i _num_threads;i){pthread_join(_thread[i],NULL);}pthread_join(taskqueue_thread,NULL);return API_SUCCESS;}
//用户调用此函数接口唤醒
API_RETURN_TYPE_T ThreadPool::wakeupThread(TaskFuncCallback p_func)
{printf(wakeupThread in .\n);this-mutex-lock();TASK_QUEUE_T task;task.cTaskFunc p_func;//将函数执行地址赋值到队列元素中task.sTask TASK_WAIT;if(task_wait_queue.length() MAX_TASK_NUM ){ this-task_wait_queue.pushBack(task); //push任务到等待任务队列中}else{//线程池数量过多此机制后续会优化printf(Current Thread Buffer is full!Please wait a moment!\n);this-mutex-unlock();return API_FAIL;}this-mutex-unlock();return API_SUCCESS;}
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249
下面新加的关于LVQueue的实现
#ifndef QUEUE_H_INCLUDED
#define QUEUE_H_INCLUDEDtemplate typename T
class LVQueue {friend struct Iterator;struct Item {T value;Item * next;Item * prev;Item(T v) : value(v), next(NULL), prev(NULL) {}};Item * head;Item * tail;int count;Item * remove(Item * p) {if (!p)return NULL;if (!p-prev)head p-next;elsep-prev-next p-next;if (!p-next)tail p-prev;elsep-next-prev p-prev;p-next NULL;p-prev NULL;count--;if (count 0) {head tail NULL;}return p;}void moveToHead(Item * item) {Item * p remove(item);if (head) {head-prev p;p-next head;head p;} else {head tail p;}count;}
public:struct Iterator {private:LVQueue * queue;Item * currentItem;public:Iterator(const Iterator v) {queue v.queue;currentItem v.currentItem;}Iterator(LVQueue * _queue) : queue(_queue), currentItem(NULL) {}T get() { return currentItem ? currentItem-value : T(); }void set(T value) { if (currentItem) currentItem-value value; }bool next() {if (!currentItem) {// first timecurrentItem queue-head;} else {// continuecurrentItem currentItem-next;}return currentItem ! NULL;}T remove() {if (!currentItem)return T();Item * next currentItem-next;Item * p queue-remove(currentItem);currentItem next;T res p-value;delete p;return res;}void moveToHead() {if (currentItem)queue-moveToHead(currentItem);}};public:Iterator iterator() { return Iterator(this); }LVQueue() : head(NULL), tail(NULL), count(0) {}~LVQueue() { clear(); }
// T operator [] (int index) {
// Item * p head;
// for (int i 0; i index; i) {
// if (!p)
// return
// }
// }int length() { return count; }void pushBack(T item) {Item * p new Item(item);if (tail) {tail-next p;p-prev tail;tail p;} else {head tail p;}count;}void pushFront(T item) {Item * p new Item(item);if (head) {head-prev p;p-next head;head p;} else {head tail p;}count;}T popFront() {if (!head)return T();Item * p remove(head);T res p-value;delete p;return res;}T popBack() {if (!tail)return T();Item * p remove(tail);T res p-value;delete p;return res;}void clear() {while (head) {Item * p head;head p-next;delete p;}head NULL;tail NULL;count 0;}
};#endif // LVQUEUE_H_INCLUDED
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
以下是简单的test程序ThreadPool *thread3 ThreadPool::createThreadPool(8);//得到线程池对象printf(task coming.\n);//test threadpoolfor(int i 0;i 15;i){thread3-wakeupThread(thread11_func);//每隔一秒唤醒线程thread11_func一个函数的地址sleep(1);thread3-wakeupThread(thread3_func);}
12345678910
下面是test程序运行的结果线程唤醒无一秒间隔
num 8.
task coming.
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
wakeupThread in .
thread 0xd528c700 begin
thread 0xd528c700 is waiting
thread 0xd4a8b700 begin
thread 0xd4a8b700 is waiting
thread 0xd428a700 begin
thread 0xd428a700 is waiting
thread 0xd5a8d700 begin
thread 0xd5a8d700 is waiting
thread 0xd628e700 begin
thread 0xd628e700 is waiting
thread 0xd6a8f700 begin
thread 0xd6a8f700 is waiting
thread 0xd7290700 begin
thread 0xd7290700 is waiting
thread 0xd7a91700 begin
thread 0xd7a91700 is waiting
managerThread!.
signal cond.
num 4.
thread 0xd2286700 begin
thread 0xd2a87700 begin
thread 0xd3288700 begin
thread 0xd3a89700 begin
thread 0xd2286700 is waiting
thread 0xd2a87700 is waiting
thread 0xd3288700 is waiting
thread 0xd3a89700 is waiting
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
signal cond.
managerThread!.
managerThread!.
managerThread!.
managerThread!.
managerThread!.
rate 4.82897
this is 0 task thread.
wait len 22.
thread 0xd528c700 done.length() 7.
thread 0xd528c700 begin
thread 0xd528c700 is waiting
managerThread!.
signal cond.
rate 4.64646
this is 1 task thread.
wait len 21.
thread 0xd4a8b700 done.length() 7.
thread 0xd4a8b700 begin
thread 0xd4a8b700 is waiting
managerThread!.
signal cond.
rate 4.64646
this is 2 task thread.
wait len 20.
thread 0xd428a700 done.length() 7.
thread 0xd428a700 begin
thread 0xd428a700 is waiting
managerThread!.
signal cond.
rate 4.25101
this is 3 task thread.
wait len 19.
thread 0xd5a8d700 done.length() 7.
thread 0xd5a8d700 begin
thread 0xd5a8d700 is waiting
managerThread!.
signal cond.
rate 4.23387
this is 4 task thread.
wait len 18.
thread 0xd628e700 done.length() 7.
thread 0xd628e700 begin
thread 0xd628e700 is waiting
managerThread!.
signal cond.
rate 4.04858
this is 5 task thread.
wait len 17.
thread 0xd6a8f700 done.length() 7.
thread 0xd6a8f700 begin
thread 0xd6a8f700 is waiting
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126
可以看到一次性唤醒了30个线程创建了8个线程的线程池后来通过优化计算又新增了4个线程到当前线程池中每唤醒一个线程执行任务大概是6s的时间执行完后又进入等待唤醒信号的状态。管理线程检测到当前所有线程都在执行便会阻塞当前signal行为直到有空余线程马上signal。
这些源代码还有一些数据类型的封装还没公布出来因为还在优化中所以准备等到优化完毕将会把完整的源代码交到GitHub上托管小弟资历尚浅如有出错的地方烦请不吝赐教。
