线程池

痛定思痛。 2022-06-13 13:22 306阅读 0赞

1.所谓线程池，就是程序的初始化阶段，就预先创建一批线程，每个线程都做好准备干活；

2.然后有一个任务列表，一开始为空，当有任务来了，就往任务列表里面添加；这个任务列表由于那些线程们都会进行操作，所以需要做好同步工作。

3.任务列表里面有任务了，这时候那些等待的线程们就要抢活干了，怎么抢，使用各种线程同步手段（互斥量，临界区等），人品好的线程抢到任务后，从任务列表取出任务，就可以开始干活了。干完以后，就又继续回到初始等待状态，准备抢夺下一个任务。

/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/  
这样就好比你有一批小弟排队在那里等着，一旦有任务，他们会很守纪律的去抢着干，每个任务都会被一个小弟抢走，干完以后，小弟不用休息，继续等着抢下一个任务干活。这样当你的任务源源不断的到达，你的小弟们就一个个争先恐后的抢过来完成，绝不偷懒。  
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/

相反，如果不使用线程池，每次等到任务来了，再临时创建线程。这样就相当于每次有任务时，你再临时招聘一个小弟过来，小弟完成任务后，就回家了。然后下次再有任务，又招聘一个小弟过来，完成任务后，回家。相比线程池，中间招聘小弟的时间就要额外耗费时间和精力了（创建和销毁线程中，cpu的时间，内存的分配）。  
/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/

所以，使用线程池，就省去了哪些额外的线程开销，从而连续的完成所有的任务。当然，线程池中用于线程同步的操作同样也有一定的消耗，但这个消耗是相对小的。另外，还可以对线程池中的线程根据当前的任务量进行动态的调整，从而更好的节省相关资源。

下面代码是网上找到的一个例子：

threadpool.h

threadpool.cpp

/*
        Thread Pool implementation for unix / linux environments
        Copyright (C) 2008 Shobhit Gupta
    	
        This program is free software: you can redistribute it and/or modify
        it under the terms of the GNU General Public License as published by
        the Free Software Foundation, either version 3 of the License, or
        (at your option) any later version.
    
        This program is distributed in the hope that it will be useful,
        but WITHOUT ANY WARRANTY; without even the implied warranty of
        MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
        GNU General Public License for more details.
    
        You should have received a copy of the GNU General Public License
        along with this program.  If not, see <http://www.gnu.org/licenses/>.
    */
    
    #include <stdlib.h>
    #include "threadpool.h"
    
    using namespace std;
    
    pthread_mutex_t ThreadPool::mutexSync = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t ThreadPool::mutexWorkCompletion = PTHREAD_MUTEX_INITIALIZER;
    
    
    
    ThreadPool::ThreadPool()
    {
    	ThreadPool(2);
    }
    
    ThreadPool::ThreadPool(int maxThreads)
    {
       if (maxThreads < 1)  maxThreads=1;
      
       //mutexSync = PTHREAD_MUTEX_INITIALIZER;
       //mutexWorkCompletion = PTHREAD_MUTEX_INITIALIZER; 
       
       pthread_mutex_lock(&mutexSync);
       this->maxThreads = maxThreads;
       this->queueSize = maxThreads;
       //workerQueue = new WorkerThread *[maxThreads];
       workerQueue.resize(maxThreads, NULL);
       topIndex = 0;
       bottomIndex = 0;
       incompleteWork = 0;
       sem_init(&availableWork, 0, 0);
       sem_init(&availableThreads, 0, queueSize);
       pthread_mutex_unlock(&mutexSync);
    }
    
    void ThreadPool::initializeThreads()
    {
       for(int i = 0; i<maxThreads; ++i)
    	{
    		pthread_t tempThread;
    		pthread_create(&tempThread, NULL, &ThreadPool::threadExecute, (void *) this ); 
    		 //threadIdVec[i] = tempThread;
       }
    
    }
    
    ThreadPool::~ThreadPool()
    {
       workerQueue.clear();
    }
    
    
    
    void ThreadPool::destroyPool(int maxPollSecs = 2)
    {
    	while( incompleteWork>0 )
    	{
    	        //cout << "Work is still incomplete=" << incompleteWork << endl;
    		sleep(maxPollSecs);
    	}
    	cout << "All Done!! Wow! That was a lot of work!" << endl;
    	sem_destroy(&availableWork);
    	sem_destroy(&availableThreads);
            pthread_mutex_destroy(&mutexSync);
            pthread_mutex_destroy(&mutexWorkCompletion);
    
    }
    
    
    bool ThreadPool::assignWork(WorkerThread *workerThread)
    {
            pthread_mutex_lock(&mutexWorkCompletion);
    		incompleteWork++;
    		//cout << "assignWork...incomapleteWork=" << incompleteWork << endl;
    	pthread_mutex_unlock(&mutexWorkCompletion);
        
    	sem_wait(&availableThreads);
    	
    	pthread_mutex_lock(&mutexSync);
    		//workerVec[topIndex] = workerThread;
    		workerQueue[topIndex] = workerThread;
                    //cout << "Assigning Worker[" << workerThread->id << "] Address:[" << workerThread << "] to Queue index [" << topIndex << "]" << endl;
    		if(queueSize !=1 )
    			topIndex = (topIndex+1) % (queueSize-1);
    		sem_post(&availableWork);
    	pthread_mutex_unlock(&mutexSync);
    	return true;
    }
    
    bool ThreadPool::fetchWork(WorkerThread **workerArg)
    {
    	sem_wait(&availableWork);
    
    	pthread_mutex_lock(&mutexSync);
    		WorkerThread * workerThread = workerQueue[bottomIndex];
                    workerQueue[bottomIndex] = NULL;
    		*workerArg = workerThread;
    		if(queueSize !=1 )
    			bottomIndex = (bottomIndex+1) % (queueSize-1);
    		sem_post(&availableThreads);
    	pthread_mutex_unlock(&mutexSync);
        return true;
    }
    
    void *ThreadPool::threadExecute(void *param)
    {
    	WorkerThread *worker = NULL;
    	
    	while(((ThreadPool *)param)->fetchWork(&worker))
    	{
    		if(worker)
                    {
    			worker->executeThis();
                            //cout << "worker[" << worker->id << "]\tdelete address: [" << worker << "]" << endl;
                            delete worker;
                            worker = NULL;
                    }
    
    		pthread_mutex_lock( &(((ThreadPool *)param)->mutexWorkCompletion) );
                    //cout << "Thread " << pthread_self() << " has completed a Job !" << endl;
    	 	((ThreadPool *)param)->incompleteWork--;
    		pthread_mutex_unlock( &(((ThreadPool *)param)->mutexWorkCompletion) );
    	}
    	return 0;
    }

main.cpp