这里给出一个简单的C++线程池包装类,该类具有的特点是:

1.线程池大小是固定的, 一创建后,就不具有伸缩特性. 一般建议是 CPU核心数的2倍或1倍.

2.简单但是很可靠.

3.资源占用极低. 在开启100个线程时, 4核CPU的占用率只有20%不到, 30个线程时, 占用7%以下.实践证明, 使用信号量和互斥锁进行多线程的同步是最合理高效的方法, 比sleep让出cpu要好很多.

后续需要增加的功能是, 线程个数的动态伸缩, 任务处理的超时机制. 当然, 在实际探索中发现高级功能的线程池并不能优雅退出, 这是困扰我很久的问题. 所以, 这里只给出最简单可靠的代码.

运行环境:

Ubuntu 14.04 64bit LTS

头文件thread_pool.h

#ifndef __THREAD_POOL_H__
#define __THREAD_POOL_H__#include 
#include 
#include 
#include 
#include using namespace std;class CJob{public:CJob(void* (*r)(void* arg), void* a): callback_routine(r), arg(a){}~CJob(){}void* (*callback_routine) (void* arg);void* arg;
};//fixed size thread pool
class CThreadPool{public:CThreadPool(int max_th_num);~CThreadPool();int pool_add_job(void* (*process)(void* arg), void* arg);pthread_mutex_t      queue_mutex;pthread_cond_t       queue_cond;list          queue_job;pthread_t*           thread_vec;int                  max_thread_num;int                  cur_queue_size;int                  shutdown;
};#endif

源文件thread_pool.cpp

#include "thread_pool.h"static void* thread_routine(void* arg){CThreadPool* pool = (CThreadPool*)arg;if(pool == NULL) return NULL;printf("starting thread [%lu]
", pthread_self());while(1){pthread_mutex_lock(&pool->queue_mutex);while (pool->cur_queue_size == 0 && !pool->shutdown){printf("thread [%lu] is waiting
", pthread_self());pthread_cond_wait(&pool->queue_cond, &pool->queue_mutex);}if(pool->shutdown){printf("thread [%lu] will exit
", pthread_self());pthread_mutex_unlock(&pool->queue_mutex);pthread_exit(NULL);}printf("thread [%lu] is starting to work
", pthread_self());assert (pool->cur_queue_size != 0);assert (!pool->queue_job.empty());pool->cur_queue_size--;CJob* job = pool->queue_job.front();pool->queue_job.pop_front();pthread_mutex_unlock(&pool->queue_mutex);(*job->callback_routine) (job->arg);delete job;}
}CThreadPool::CThreadPool(int max_th_num): cur_queue_size(0), shutdown(0), max_thread_num(max_th_num){pthread_mutex_init(&queue_mutex, NULL);pthread_cond_init(&queue_cond, NULL);thread_vec = new pthread_t[max_thread_num];for(int i = 0; i < max_thread_num; i++){pthread_create(&thread_vec[i], NULL, thread_routine, (void*)this);}
}CThreadPool::~CThreadPool(){if(shutdown) return;shutdown = 1;pthread_cond_broadcast(&queue_cond);for(int i=0; i < max_thread_num; i++)pthread_join(thread_vec[i], NULL);delete [] thread_vec;for(list::iterator it = queue_job.begin(); it != queue_job.end(); ++it)delete *it;queue_job.clear();pthread_mutex_destroy(&queue_mutex);pthread_cond_destroy(&queue_cond);
}int CThreadPool::pool_add_job(void* (*process)(void* arg), void* arg){CJob* job = new CJob(process, arg);pthread_mutex_lock(&queue_mutex);queue_job.push_back(job);cur_queue_size++;pthread_mutex_unlock(&queue_mutex);pthread_cond_signal(&queue_cond);return 0;
}

测试文件test_pool.cpp

//g++ -g test_pool.cpp thread_pool.cpp -o test_pool -lpthread
//
#include 
#include "thread_pool.h"void* job_process(void* arg){printf("thread [%lu], working on task [%d]
", pthread_self(), *(int*)arg);usleep(1000 * 1000);return NULL;
}int main(int argc, char* argv[]){CThreadPool* pool = new CThreadPool(100);int* jobNo = new int[10000];for(int i = 0; i < 10000; i++){jobNo[i] = i;pool->pool_add_job(job_process, &jobNo[i]);}usleep(100 * 1000 * 1000);delete pool;delete [] jobNo;return 0;
}

测试截图

欢迎批评指正.