這里給出一個簡單的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 <stdio.h>
#include <pthread.h>
#include <assert.h>
#include <iostream>
#include <list>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<CJob*>          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]\n", pthread_self());while(1){pthread_mutex_lock(&pool->queue_mutex);while (pool->cur_queue_size == 0 && !pool->shutdown){printf("thread [%lu] is waiting\n", pthread_self());pthread_cond_wait(&pool->queue_cond, &pool->queue_mutex);}if(pool->shutdown){printf("thread [%lu] will exit\n", pthread_self());pthread_mutex_unlock(&pool->queue_mutex);pthread_exit(NULL);}printf("thread [%lu] is starting to work\n", 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<CJob*>::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 <unistd.h>
#include "thread_pool.h"void* job_process(void* arg){printf("thread [%lu], working on task [%d]\n", 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;
}

測試截圖

歡迎批評指正.

總結

以上是生活随笔為你收集整理的简单C++线程池包装类源码示例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。