c++线程池实现方法

 #ifndef LOCKER_H_ #define LOCKER_H_ #include "pthread.h" class locker { public: locker(); virtual ~locker(); bool lock(); void unlock(); private: pthread_mutex_t   m_mutex; }; #endif /* LOCKER_H_ */ 

 #include "locker.h" locker::locker() { pthread_mutex_init(&m_mutex, 0); } locker::~locker() { pthread_mutex_destroy(&m_mutex); } bool locker::lock() { if(0 == pthread_mutex_lock(&m_mutex)) return true; return false; } void locker::unlock() { pthread_mutex_unlock(&m_mutex); } 

 #ifndef TASK_LIST_H_ #define TASK_LIST_H_ #include "list" #include "locker.h" #include "netinet/in.h" #include "semaphore.h" using namespace std; typedef void* (*THREAD_FUNC)(void*); // 线程池中运行的任务，对于下行任务，sin中包含目的地址信息 // parm0指向发出数据的对象，parm1指向数据，parm2为数据的长度 typedef struct { THREAD_FUNC func; void* parm0; void* parm1; void* parm2; } task_info; typedef list TASK_LIST; typedef list::iterator PTASK_LIST; class task_list { public: task_list(); virtual ~task_list(); void append_task(task_info* tsk); task_info* fetch_task(); private: TASK_LIST m_tasklist; locker m_lk; sem_t m_sem; }; #endif /* TASK_LIST_H_ */ 

 #include "task_list.h" task_list::task_list() { // Init Semaphore sem_init(&m_sem, 0, 0); m_tasklist.clear(); } task_list::~task_list() { while(!m_tasklist.empty()) { task_info* tr = m_tasklist.front(); m_tasklist.pop_front(); if(tr) delete tr; } // Destroy Semaphore sem_destroy(&m_sem); } void task_list::append_task(task_info* tsk) { // Lock before Modify the list m_lk.lock(); m_tasklist.push_back(tsk); m_lk.unlock(); // Increase the Semaphore sem_post(&m_sem); } task_info* task_list::fetch_task() { task_info* tr = NULL; sem_wait(&m_sem); m_lk.lock(); tr = m_tasklist.front(); m_tasklist.pop_front(); m_lk.unlock(); return tr; } 

 #ifndef THREAD_POOL_H_ #define THREAD_POOL_H_ #include "task_list.h" #include "pthread.h" #define DEFAULT_THREAD_COUNT  4 #define MAXIMUM_THREAD_COUNT  1000 class thread_pool { public: thread_pool(); virtual ~thread_pool(); int create_threads(int n = DEFAULT_THREAD_COUNT); void delete_threads(); void set_tasklist(task_list* plist); void del_tasklist(); protected: static void* thread_func(void* parm); task_info* get_task(); private: int       m_thread_cnt; pthread_t    m_pids[MAXIMUM_THREAD_COUNT]; task_list*   m_tasklist; }; #endif /* THREAD_POOL_H_ */ 

 #include "thread_pool.h" thread_pool::thread_pool() { m_thread_cnt = 0; m_tasklist = NULL; } thread_pool::~thread_pool() { delete_threads(); } task_info* thread_pool::get_task() { task_info* tr; if (m_tasklist) { tr = m_tasklist->fetch_task(); return tr; } return NULL; } void* thread_pool::thread_func(void* parm) { thread_pool *ptp = static_cast (parm); task_info *task; while (true) { task = ptp->get_task(); if (task) { (*task->func)(task); //delete task; //func负责释放task_info } } return NULL; } int thread_pool::create_threads(int n) { if (n > MAXIMUM_THREAD_COUNT) n = MAXIMUM_THREAD_COUNT; delete_threads(); for (int i = 0; i 

 #include "unistd.h" #include "stdio.h" #include "stdlib.h" #include "task_list.h" #include "thread_pool.h" void* fun(void *parm) { task_info* ptk = (task_info*)parm; pid_t tid = pthread_self(); int count = (int)ptk->parm0; printf("count=%d, tid=%d\n", count, tid); return NULL; } int main() { int count = 0; thread_pool tp; task_list tl; tp.create_threads(4 - 1); tp.set_tasklist(&tl); while (1) { task_info* pti = NULL; pti = (task_info *) malloc(sizeof(task_info)); pti->func = fun; pti->parm0 = (void *)count; tl.append_task(pti); count++; sleep(2); } // printf("hello,world\n"); return 0; } 

 bin_PROGRAMS=test test_SOURCES=test.cpp locker.h locker.cpp \ task_list.h task_list.cpp \ thread_pool.h thread_pool.cpp test_LDADD=-lpthread 

 count=0, tid=-1219888272 count=1, tid=-1219888272 count=2, tid=-1228280976 count=3, tid=-1236673680 count=4, tid=-1219888272 count=5, tid=-1228280976 count=6, tid=-1236673680 count=7, tid=-1219888272 count=8, tid=-1228280976 count=9, tid=-1236673680