## **為什么要采用線程池？** 首先想一想，我們一般的服務器都是動態創建子線程來實現并發服務器的，比如每當有一個客戶端請求建立連接時我們就動態調用pthread\_create去創建線程去處理該連接請求。這種模式有什么缺點呢？ * 動態創建線程是比較費時的，這將到導致較慢的客戶響應。 * 動態創建的子線程通常只用來為一個客戶服務，這將導致系統上產生大量的細微線程，線程切換也會耗費CPU時間。 所以我們為了進一步提升服務器性能，可以采取“池”的思路，把線程的創建放在程序的初始化階段一次完成，這就避免了動態創建線程導致服務器響應請求的性能下降。 ## **線程池的設計思路** 1. 以單例模式設計線程池，保證線程池全劇唯一； 2. 在獲取線程池實例進行線程池初始化：線程預先創建+任務隊列創建； 3. 創建一個任務類，我們真實的任務會繼承該類，完成任務執行1. ``` #include <queue> #include <unistd.h> #include <pthread.h> #include <mutex> #include <vector> #include <stdio.h> typedef struct { int task_id; std::string task_name; }msg_t; class Task { public: Task(void* a = NULL): arg(a) { } void SetArg(void* a) { arg = a; } virtual int run()=0; protected: void* arg; }; class MyTask: public Task { public: int run() { msg_t* msg = (msg_t*)arg; //參數解析 printf("working thread[%lu] : task_id:%d task_name:%s\n", pthread_self(), msg->task_id, msg->task_name.c_str()); sleep(10); return 0; } }; class ThreadPool { private: std::queue<Task*> taskQueue;//任務隊列 bool isRunning; //線程池運行標志 pthread_t* pThreadSet; //指向線程id集合的指針 int threadsNum; //線程數目 pthread_mutex_t mutex; //互斥鎖 pthread_cond_t condition; //條件變量 ThreadPool(int num = 10):threadsNum(num) { printf ("creating threads %d pool ...\n ", num); isRunning = true; pthread_mutex_init(&mutex, NULL); pthread_cond_init(&condition, NULL); createThreads(); printf("created threads pool successfully!\n"); } void createThreads() { pThreadSet = (pthread_t *) malloc(sizeof(pthread_t) * threadsNum); for(int i = 0; i < threadsNum; ++i) { pthread_create(&pThreadSet[i], NULL, threadFunc, this); } } void clearThreads() { if (!isRunning) return; isRunning = false; pthread_cond_broadcast(&condition); printf("start closing ...\n"); for (int i = 0; i < threadsNum; i ++) { pthread_join(pThreadSet[i], NULL); printf("close thread %lu\n",pThreadSet[i]); } free(pThreadSet); pThreadSet = NULL; pthread_mutex_destroy(&mutex); pthread_cond_destroy(&condition); } void clearQueue() { while(!taskQueue.empty()) { delete(taskQueue.front()); taskQueue.pop(); } } static void* threadFunc(void* arg) { ThreadPool* p = (ThreadPool*)arg; while(p->isRunning) { Task* task = p->takeTask(); if (!task) { continue; } printf("take one ...\n"); task->run(); } } Task* takeTask() { Task* pTask = NULL; while(!pTask) { pthread_mutex_lock(&mutex); while(taskQueue.empty() && isRunning) { pthread_cond_wait(&condition, &mutex); } if (!isRunning) { pthread_mutex_unlock(&mutex); break; } else if (taskQueue.empty()) { pthread_mutex_unlock(&mutex); continue; } pTask = taskQueue.front(); taskQueue.pop(); pthread_mutex_unlock(&mutex); } return pTask; } public : void addTask(Task* pTask) { pthread_mutex_lock(&mutex); taskQueue.push(pTask); printf("one task is put into queue! Current queue size is %lu\n",taskQueue.size()); pthread_mutex_unlock(&mutex); pthread_cond_signal(&condition); } static ThreadPool* createThreadPool(int num = 10) { static ThreadPool* pThreadPoolInstance = new ThreadPool(num); return pThreadPoolInstance; } ~ThreadPool() { clearThreads(); clearQueue(); printf("thread pool is closed!\n"); } int getQueueSize() { pthread_mutex_lock(&mutex); int size = taskQueue.size(); pthread_mutex_unlock(&mutex); printf("current queue size is %d\n", size); return size; } int getThreadNum() { return threadsNum; } }; int main() { ThreadPool* pMyPool = ThreadPool::createThreadPool(5); //建立大小為5的線程池 char buf[32] = {0}; int num = 200; msg_t msg[num]; MyTask task_A[num]; //模擬生產者生產任務 for(int i=0;i<num;i++) { msg[i].task_id = i; sprintf(buf,"qq_task_%d",i); msg[i].task_name = buf; task_A[i].SetArg(&msg[i]); //設置函數參數 pMyPool->addTask(&task_A[i]); //任務入隊 sleep(1); } while(1) { //printf("there are still %d tasks need to process\n", pMyPool->getQueueSize()); if (pMyPool->getQueueSize() == 0) { printf("Now I will exit from main\n"); break; } sleep(1); } delete pMyPool; return 0; } ```