非常精簡的Linux線程池實(shí)現(xiàn)

非常精簡的線程池實(shí)現(xiàn)（一）——使用互斥鎖和條件變量線程池的含義跟它的名字一樣，就是一個(gè)由許多線程組成的池子。有了線程池，在程序中使用多線程變得簡單。我們不用再自己去操心線程的創(chuàng)建、撤銷、管理問題，有什么要消耗大量時(shí)間的任務(wù)通通直接扔到線程池里就好了，然后我們的主程序（主線程）可以繼續(xù)干自己的事去，線程池里面的線程會自動去執(zhí)行這些任務(wù)。另一方面，線程池提升了多線程程序的性能。我們不需要在大量任務(wù)需要執(zhí)行時(shí)現(xiàn)創(chuàng)建大量線程，然后在任務(wù)結(jié)束時(shí)又銷毀大量線程，因?yàn)榫€程池里面的線程都是現(xiàn)成的而且能夠重復(fù)使用。一個(gè)理想的線程池能夠合理地動態(tài)調(diào)節(jié)池內(nèi)線程數(shù)量，既不會因?yàn)榫€程過少而導(dǎo)致大量任務(wù)堆積，也不會因?yàn)榫€程過多了而增加額外的系統(tǒng)開銷。線程池看上去很神奇的樣子，那它是怎么實(shí)現(xiàn)的呢？線程這么虛渺在的東西也能像有形的物品一樣圈在一個(gè)池子里？在只知道線程池這個(gè)名字的時(shí)候，我心里的疑惑就是這樣的。其實(shí)線程池的原理非常簡單，它就是一個(gè)非常典型的生產(chǎn)者消費(fèi)者同步問題。如果不知道我說的這個(gè)問題也不要緊，我下面就解釋。根據(jù)剛才描述的線程池的功能，可以看出線程池至少有兩個(gè)主要動作，一個(gè)是主程序不定時(shí)地向線程池添加任務(wù)，另一個(gè)是線程池里的線程領(lǐng)取任務(wù)去執(zhí)行。且不論任務(wù)和執(zhí)行任務(wù)是個(gè)什么概念，但是一個(gè)任務(wù)肯定只能分配給一個(gè)線程執(zhí)行。這樣就可以簡單猜想線程池的一種可能的架構(gòu)了：主程序執(zhí)行入隊(duì)操作，把任務(wù)添加到一個(gè)隊(duì)列里面；池子里的多個(gè)工作線程共同對這個(gè)隊(duì)列試圖執(zhí)行出隊(duì)操作，這里要保證同一時(shí)刻只有一個(gè)線程出隊(duì)成功，搶奪到這個(gè)任務(wù)，其他線程繼續(xù)共同試圖出隊(duì)搶奪下一個(gè)任務(wù)。所以在實(shí)現(xiàn)線程池之前，我們需要一個(gè)隊(duì)列，我為這個(gè)線程池配備的隊(duì)列單獨(dú)放到了另一篇博客一個(gè)通用純隊(duì)列的實(shí)現(xiàn)中。這里的生產(chǎn)者就是主程序，生產(chǎn)任務(wù)（增加任務(wù)），消費(fèi)者就是工作線程，消費(fèi)任務(wù)（執(zhí)行、減少任

務(wù)）。因?yàn)檫@里涉及到多個(gè)線程同時(shí)訪問一個(gè)隊(duì)列的問題，所以我們需要互斥鎖來保護(hù)隊(duì)列，同時(shí)還需要條件變量來處理主線程通知任務(wù)到達(dá)、工作線程搶奪任務(wù)的問題。如果不熟悉條件變量，我在另一篇博客LinuxC語言多線程庫Pthread中條件變量的的正確用法逐步詳解中作了詳細(xì)說明。準(zhǔn)備工作都差不多了，可以開始設(shè)計(jì)線程池了。一個(gè)最簡單線程池應(yīng)該有什么功能呢？對于使用者來說，除了創(chuàng)建和銷毀線程池，最簡單的情況下只需要一個(gè)功能一一添加任務(wù)。對于線程池自己來說，最簡單的情況下不需要動態(tài)調(diào)節(jié)線程數(shù)量，不需要考慮線程同步、線程死鎖等等一大堆麻煩的問題。所以最后的線程池定義為：創(chuàng)建線程池時(shí)指定線程池中應(yīng)該固定包含多少工作線程，添加任務(wù)就是向線程池添加一個(gè)任務(wù)函數(shù)指針和任務(wù)函數(shù)需要的參數(shù)——這跟線程庫中的普通線程創(chuàng)建函數(shù)是一樣針和任務(wù)函數(shù)需要的參數(shù)——這跟線程庫中的普通線程創(chuàng)建函數(shù)是一樣的。根據(jù)這套線程池 ，我們使用線程池的應(yīng)用程序應(yīng)該是這個(gè)套路：#include<unistd.h>#include<pthread.h>void*test(void*arg){inti;for(i=0;i<5;i++){printf("tid:%ldtask:%ld\n",pthread_self(),(long)arg);fflush(stdout);sleep(2);}returnNULL;}intmain(){longi=0;thread_pool_tpool;pool=thread_pool_create(2);for(i=0;i<5;i++){thread_pool_add_task(pool,test,(void*)i);}puts("pressentertoterminate...");getchar();thread_pool_destroy(pool);return0;

上面這個(gè)測試程序向線程池添加了個(gè)相同的任務(wù)，每個(gè)任務(wù)耗時(shí)秒，但是線程池中只有個(gè)工作線呈，所以程序的運(yùn)行結(jié)果是兩個(gè)工作線程輪流把個(gè)任務(wù)挨個(gè)做完。顯示到屏幕上就是：前秒兩個(gè)工作線程輪流輸出自己的線程和當(dāng)前任務(wù)的任務(wù)號和，各輸出次；第二個(gè)秒兩個(gè)工作線程輪流輸出自己的線程和當(dāng)前任務(wù)的任務(wù)號和3……在這期間，主程序輸出“pressentertoterminate...”并等待用戶輸入，任何時(shí)候都可以按回車讓主程序繼續(xù)往下，這樣會強(qiáng)制終止所有工作線程并銷毀線程池，最后程序退出。test程序運(yùn)行效果截圖如下：最后就是線程池真正的實(shí)現(xiàn)了：treapoo.inetreapooin eee.inesti.in eptrea.str ttreapoonsineinttrea ontptreattreaseettassptreamtetoptreaonttasreaptreaonttasreastructtask{void*(*routine)(void*arg);void*arg;}；staticvoidcleanup(pthread_mutex_t*lock){pthread_mutex_unlock(lock);}staticvoid*worker(thread_pool_tpool){structtask*t;while(1){pthread_mutex_lock(&pool->lock);pthread_cleanup_push((void(*)(void*))cleanup,&pool->lock);while(queue_isempty(pool->tasks)){pthread_cond_wait(&pool->task_ready,&pool->lock);/*Aconditionwait(whethertimedornot)isacancellationpoint...aside-effectofactinguponacancellationrequestwhileinaconditionwaitisthatthemutexis(ineffect)re-acquiredbeforecallingthefirstcancellationcleanuphandler.*/}t=(structtask*)queue_dequeue(pool->tasks);pthread_cleanup_pop(0);pthread_mutex_unlock(&pool->lock);t->routine(t->arg);/*todo:reportreturnedvalue*/free(t);}returnNULL;}thread_pool_tthread_pool_create(unsignedintthread_count){unsignedinti;thread_pool_tpool=NULL;pool=(thread_pool_t)malloc(sizeof(structthread_pool));pool->thread_count=thread_count;pool->threads=(pthread_t*)malloc(sizeof(pthread_t)*thread_count);pool->tasks=queue_create();pthread_mutex_init(&pool->lock,NULL);pthread_cond_init(&pool->task_ready,NULL);for(i=0;i<thread_count;i++){pthread_create(pool->threads+i,NULL,(void*(*)(void*))worker,pool);}returnpool;}voidthread_pool_add_task(thread_pool_tpool,void*(*routine)(void*arg),void*arg){structtask*t;pthread_mutex_lock(&pool->lock);t=(structtask*)queue_enqueue(pool->tasks,sizeof(structtask));上面的 函數(shù)就是工作線程函數(shù)，所有的工作線程都在執(zhí)行著這個(gè)函數(shù)。它首先在互斥鎖和條件變量的保護(hù)下從任務(wù)隊(duì)列中取出一個(gè)任務(wù)，這個(gè)任務(wù)實(shí)際上是一個(gè)函數(shù)指針和調(diào)用函數(shù)所需的參數(shù)，所以執(zhí)行任務(wù)就很簡單了——用任務(wù)參數(shù)調(diào)用任務(wù)函數(shù)。函數(shù)返回以后，工作線程繼續(xù)去搶任務(wù)。這里沒有處理任務(wù)函數(shù)的返回值問題，理論上任務(wù)函數(shù)返回以后線程池應(yīng)該用某種機(jī)制通知主程序，然后主程序獲取通過某種手段獲取返回值，但這明顯不是一個(gè)最簡單的線程池需要操心的事。實(shí)際上，應(yīng)用程序可以通過全局變量或傳入的參數(shù)指針，加上額外的線程同步代碼解決返回值的通知和獲取問題。還有一點(diǎn)需要注意，最后線程池銷毀