前言

概述

策略选择原则
线程池，线程，事件循环，非阻塞式，多路复用，异步（中断）都有其使用的场景
当线程本身的开销相对于线程任务执行开销而言是可以忽略不计的那么使用即时创建，即时销毁的线程策略比线程池更好

线程池原理
预先启动一些线程，线程无限循环从任务队列中获取一个任务进行执行，直到线程池被关闭。如果某个线程因为执行某个任务发生异常而终止，那么重新创建一个新的线程而已，如此反复。

线程池风险
1.同步错误，死锁，活锁，饥饿等是线程的并发风险
2.资源不足
3.并发错误：线程池和其它排队机制依靠使用wait() 和 notify()方法，这两个方法都难于使用。如果编码不正确，那么可能丢失通知，导致线程保持空闲状态
4.线程泄露：一种情形出现在任务抛出一个 RuntimeException 或一个 Error 时。如果池类没有捕捉到它们，就会泄露
5.请求过载

线程池大小配置参考：
如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU+1
如果是IO密集型任务，参考值可以设置为2*NCPU

创建一个具有动态伸缩性的线程池

线程池框架

线程池管理器:用于创建并管理线程池
工作线程: 线程池中实际执行的线程
任务接口: 尽管线程池大多数情况下是用来支持网络服务器，但是我们将线程执行的任务抽象出来，形成任务接口，从而使得线程池与具体的任务无关。
任务队列:线程池的概念具体到实现则可能是队列，链表之类的数据结构，其中保存执行线程。

我们的框架组成部分：CThreadManage，CThreadPool，CThread，CJob，CWorkerThread + 线程同步使用的类CThreadMutex和CCondition

线程池的时序很简单。CThreadManage直接跟客户端打交道，其接受需要创建的线程初始个数，并接受客户端提交的任务。这儿的任务是具体的非抽象的任务。CThreadManage的内部实际上调用的都是CThreadPool的相关操作。CThreadPool创建具体的线程，并把客户端提交的任务分发给CWorkerThread，CWorkerThread实际执行具体的任务。

系统组件

CThreadManage

CThreadManage是线程池与用户的直接接口，其屏蔽了内部的具体实现。

class CThreadManage
{
private:
    CThreadPool*    m_Pool;		//指向实际的线程池
    int          m_NumOfThread;	//初始创建时候允许创建的并发的线程个数
public:
    void     SetParallelNum(int num);
    CThreadManage();
    CThreadManage(int num);
    virtual ~CThreadManage();

    void    Run(CJob* job,void* jobdata);//调用CThreadPool的方法
    void    TerminateAll(void);			//调用CThreadPool的方法
};

CThreadManage::CThreadManage(){
    m_NumOfThread = 10;
    m_Pool = new CThreadPool(m_NumOfThread);
}
CThreadManage::CThreadManage(int num){
    m_NumOfThread = num;
    m_Pool = new CThreadPool(m_NumOfThread);
}
CThreadManage::~CThreadManage(){
    if(NULL != m_Pool)
    delete m_Pool;
}
void CThreadManage::SetParallelNum(int num){
    m_NumOfThread = num;
}
void CThreadManage::Run(CJob* job,void* jobdata){
    m_Pool->Run(job,jobdata);
}
void CThreadManage::TerminateAll(void){
    m_Pool->TerminateAll();
}

CThread

CThread是Linux中线程的包装，其封装了Linux线程最经常使用的属性和方法，它也是一个抽象类，是所有线程类的基类，具有一个接口Run。

class CThread
{
private:
    int          m_ErrCode;
    Semaphore    m_ThreadSemaphore;  //the inner semaphore, which is used to realize
    unsigned     long m_ThreadID;
    bool         m_Detach;       //The thread is detached
    bool         m_CreateSuspended;  //if suspend after creating
    char*        m_ThreadName;
    ThreadState m_ThreadState;      //the state of the thread
protected:
    void     SetErrcode(int errcode){m_ErrCode = errcode;}
    static void* ThreadFunction(void*);
public:
    CThread();
    CThread(bool createsuspended,bool detach);
    virtual ~CThread();
    virtual void Run(void) = 0;
    void     SetThreadState(ThreadState state){m_ThreadState = state;}

    bool     Terminate(void);    //Terminate the threa
    bool     Start(void);        //Start to execute the thread
    void     Exit(void);
    bool     Wakeup(void);

    ThreadState  GetThreadState(void){return m_ThreadState;}
    int      GetLastError(void){return m_ErrCode;}
    void     SetThreadName(char* thrname){strcpy(m_ThreadName,thrname);}
    char*    GetThreadName(void){return m_ThreadName;}
    int      GetThreadID(void){return m_ThreadID;}

    bool     SetPriority(int priority);
    int      GetPriority(void);
    int      GetConcurrency(void);
    void     SetConcurrency(int num);
    bool     Detach(void);
    bool     Join(void);
    bool     Yield(void);
    int      Self(void);
};

线程的状态分为四种，空闲、忙碌、挂起、终止(包括正常退出和非正常退出)。由于目前Linux线程库不支持挂起操作，因此，我们的此处的挂起操作类似于暂停。如果线程创建后不想立即执行任务，那么我们可以将其“暂停”，如果需要运行，则唤醒。有一点必须注意的是，一旦线程开始执行任务，将不能被挂起，其将一直执行任务至完毕。
线程类的相关操作均十分简单。线程的执行入口是从Start()函数开始，其将调用函数ThreadFunction，ThreadFunction再调用实际的Run函数，执行实际的任务。

CThreadPool

CThreadPool是线程的承载容器，一般可以将其实现为堆栈、单向队列或者双向队列。在我们的系统中我们使用STL Vector对线程进行保存。线程池类，其负责保存线程，释放线程以及调度线程。

CThreadPool::CThreadPool()
{
    m_MaxNum = 50;
    m_AvailLow = 5;
    m_InitNum=m_AvailNum = 10 ;
    m_AvailHigh = 20;

    m_BusyList.clear();
    m_IdleList.clear();
    for(int i=0;i<m_InitNum;i++){
    CWorkerThread* thr = new CWorkerThread();
    thr->SetThreadPool(this);
    AppendToIdleList(thr);
    thr->Start();
    }
}

CThreadPool::CThreadPool(int initnum)
{
    assert(initnum>0 && initnum<=30);
    m_MaxNum   = 30;
    m_AvailLow = initnum-10>0?initnum-10:3;
    m_InitNum=m_AvailNum = initnum ;
    m_AvailHigh = initnum+10;

    m_BusyList.clear();
    m_IdleList.clear();
    for(int i=0;i<m_InitNum;i++){
    CWorkerThread* thr = new CWorkerThread();
    AppendToIdleList(thr);
    thr->SetThreadPool(this);
    thr->Start();       //begin the thread,the thread wait for job
    }
}

CThreadPool::~CThreadPool()
{
   TerminateAll();
}

void CThreadPool::TerminateAll()
{
    for(int i=0;i < m_ThreadList.size();i++) {
    CWorkerThread* thr = m_ThreadList[i];
    thr->Join();
    }
    return;
}

CWorkerThread* CThreadPool::GetIdleThread(void)
{
    while(m_IdleList.size() ==0 )
    m_IdleCond.Wait();

    m_IdleMutex.Lock();
    if(m_IdleList.size() > 0 )
    {
    CWorkerThread* thr = (CWorkerThread*)m_IdleList.front();
    printf("Get Idle thread %dn",thr->GetThreadID());
    m_IdleMutex.Unlock();
    return thr;
    }
    m_IdleMutex.Unlock();

    return NULL;
}

//add an idle thread to idle list
void CThreadPool::AppendToIdleList(CWorkerThread* jobthread)
{
    m_IdleMutex.Lock();
    m_IdleList.push_back(jobthread);
    m_ThreadList.push_back(jobthread);
    m_IdleMutex.Unlock();
}

//move and idle thread to busy thread
void CThreadPool::MoveToBusyList(CWorkerThread* idlethread)
{
    m_BusyMutex.Lock();
    m_BusyList.push_back(idlethread);
    m_AvailNum--;
    m_BusyMutex.Unlock();

    m_IdleMutex.Lock();
    vector<CWorkerThread*>::iterator pos;
    pos = find(m_IdleList.begin(),m_IdleList.end(),idlethread);
    if(pos !=m_IdleList.end())
    m_IdleList.erase(pos);
    m_IdleMutex.Unlock();
}

void CThreadPool::MoveToIdleList(CWorkerThread* busythread)
{
    m_IdleMutex.Lock();
    m_IdleList.push_back(busythread);
    m_AvailNum++;
    m_IdleMutex.Unlock();

    m_BusyMutex.Lock();
    vector<CWorkerThread*>::iterator pos;
    pos = find(m_BusyList.begin(),m_BusyList.end(),busythread);
    if(pos!=m_BusyList.end())
    m_BusyList.erase(pos);
    m_BusyMutex.Unlock();

    m_IdleCond.Signal();
    m_MaxNumCond.Signal();
}

//create num idle thread and put them to idlelist
void CThreadPool::CreateIdleThread(int num)
{
    for(int i=0;i<num;i++){
    CWorkerThread* thr = new CWorkerThread();
    thr->SetThreadPool(this);
    AppendToIdleList(thr);
    m_VarMutex.Lock();
    m_AvailNum++;
    m_VarMutex.Unlock();
    thr->Start();       //begin the thread,the thread wait for job
    }
}

void CThreadPool::DeleteIdleThread(int num)
{
    printf("Enter into CThreadPool::DeleteIdleThreadn");
    m_IdleMutex.Lock();
    printf("Delete Num is %dn",num);
    for(int i=0;i<num;i++){
    CWorkerThread* thr;
    if(m_IdleList.size() > 0 ){
            thr = (CWorkerThread*)m_IdleList.front();
            printf("Get Idle thread %dn",thr->GetThreadID());
    }

    vector<CWorkerThread*>::iterator pos;
    pos = find(m_IdleList.begin(),m_IdleList.end(),thr);
    if(pos!=m_IdleList.end())
        m_IdleList.erase(pos);
    m_AvailNum--;
    printf("The idle thread available num:%d n",m_AvailNum);
    printf("The idlelist              num:%d n",m_IdleList.size());
    }
    m_IdleMutex.Unlock();
}
void CThreadPool::Run(CJob* job,void* jobdata)
{
    assert(job!=NULL);

    //if the busy thread num adds to m_MaxNum,so we should wait
    if(GetBusyNum() == m_MaxNum)
        m_MaxNumCond.Wait();

    if(m_IdleList.size()<m_AvailLow)
    {
    if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum )
        CreateIdleThread(m_InitNum-m_IdleList.size());
    else
        CreateIdleThread(m_MaxNum-GetAllNum());
    }

    CWorkerThread*  idlethr = GetIdleThread();
    if(idlethr !=NULL)
    {
    idlethr->m_WorkMutex.Lock();
    MoveToBusyList(idlethr);
    idlethr->SetThreadPool(this);
    job->SetWorkThread(idlethr);
    printf("Job is set to thread %d n",idlethr->GetThreadID());
    idlethr->SetJob(job,jobdata);
    }
}

在CThreadPool中存在两个链表，一个是空闲链表，一个是忙碌链表。Idle链表中存放所有的空闲进程，当线程执行任务时候，其状态变为忙碌状态，同时从空闲链表中删除，并移至忙碌链表中。在CThreadPool的构造函数中，我们将执行下面的代码:

for(int i=0;i<m_InitNum;i++)
    {
    CWorkerThread* thr = new CWorkerThread();
    AppendToIdleList(thr);
    thr->SetThreadPool(this);
    thr->Start();       //begin the thread,the thread wait for job
    }

在该代码中，我们将创建m_InitNum个线程，创建之后即调用AppendToIdleList放入Idle链表中，由于目前没有任务分发给这些线程，因此线程执行Start后将自己挂起。
事实上，线程池中容纳的线程数目并不是一成不变的，其会根据执行负载进行自动伸缩。为此在CThreadPool中设定四个变量：
m_InitNum：处世创建时线程池中的线程的个数。
m_MaxNum:当前线程池中所允许并发存在的线程的最大数目。
m_AvailLow:当前线程池中所允许存在的空闲线程的最小数目，如果空闲数目低于该值，表明负载可能过重，此时有必要增加空闲线程池的数目。实现中我们总是将线程调整为m_InitNum个。
m_AvailHigh：当前线程池中所允许的空闲的线程的最大数目，如果空闲数目高于该值，表明当前负载可能较轻，此时将删除多余的空闲线程，删除后调整数也为m_InitNum个。
m_AvailNum：目前线程池中实际存在的线程的个数，其值介于m_AvailHigh和m_AvailLow之间。如果线程的个数始终维持在m_AvailLow和m_AvailHigh之间，则线程既不需要创建，也不需要删除，保持平衡状态。因此如何设定m_AvailLow和m_AvailHigh的值，使得线程池最大可能的保持平衡态，是线程池设计必须考虑的问题。
线程池在接受到新的任务之后，线程池首先要检查是否有足够的空闲池可用。检查分为三个步骤：
(1)检查当前处于忙碌状态的线程是否达到了设定的最大值m_MaxNum，如果达到了，表明目前没有空闲线程可用，而且也不能创建新的线程，因此必须等待直到有线程执行完毕返回到空闲队列中。
(2)如果当前的空闲线程数目小于我们设定的最小的空闲数目m_AvailLow，则我们必须创建新的线程，默认情况下，创建后的线程数目应该为m_InitNum，因此创建的线程数目应该为( 当前空闲线程数与m_InitNum);但是有一种特殊情况必须考虑，就是现有的线程总数加上创建后的线程数可能超过m_MaxNum，因此我们必须对线程的创建区别对待。

if(GetAllNum()+m_InitNum-m_IdleList.size() < m_MaxNum ) 
        CreateIdleThread(m_InitNum-m_IdleList.size()); 
     else 
         CreateIdleThread(m_MaxNum-GetAllNum());

如果创建后总数不超过m_MaxNum，则创建后的线程为m_InitNum；如果超过了，则只创建( m_MaxNum-当前线程总数 )个。
(3)调用GetIdleThread方法查找空闲线程。如果当前没有空闲线程，则挂起；否则将任务指派给该线程，同时将其移入忙碌队列。
当线程执行完毕后，其会调用MoveToIdleList方法移入空闲链表中，其中还调用m_IdleCond.Signal()方法，唤醒GetIdleThread()中可能阻塞的线程。

CWorkerThread

CThread的派生类，是事实上的工作线程。在CThreadPool的构造函数中，我们创建了一定数量的CWorkerThread。一旦这些线程创建完毕，我们将调用Start()启动该线程。Start方法最终会调用Run方法。Run方法是个无限循环的过程。在没有接受到实际的任务的时候，m_Job为NULL，此时线程将调用Wait方法进行等待，从而处于挂起状态。一旦线程池将具体的任务分发给该线程，其将被唤醒，从而通知线程从挂起的地方继续执行。

class CWorkerThread:public CThread
{
private:
    CThreadPool*  m_ThreadPool;
    CJob*    m_Job;
    void*    m_JobData;

    CThreadMutex m_VarMutex;
    bool      m_IsEnd;
protected:
public:
    CCondition   m_JobCond;
    CThreadMutex m_WorkMutex;
    CWorkerThread();
    virtual ~CWorkerThread();
    void Run();
    void    SetJob(CJob* job,void* jobdata);
    CJob*   GetJob(void){return m_Job;}
    void    SetThreadPool(CThreadPool* thrpool);
    CThreadPool* GetThreadPool(void){return m_ThreadPool;}
};
CWorkerThread::CWorkerThread()
{
    m_Job = NULL;
    m_JobData = NULL;
    m_ThreadPool = NULL;
    m_IsEnd = false;
}
CWorkerThread::~CWorkerThread()
{
    if(NULL != m_Job)
    delete m_Job;
    if(m_ThreadPool != NULL)
    delete m_ThreadPool;
}

void CWorkerThread::Run()
{
    SetThreadState(THREAD_RUNNING);
    for(;;)
    {
    while(m_Job == NULL)
        m_JobCond.Wait();

    m_Job->Run(m_JobData);
    m_Job->SetWorkThread(NULL);
    m_Job = NULL;
    m_ThreadPool->MoveToIdleList(this);
    if(m_ThreadPool->m_IdleList.size() > m_ThreadPool->GetAvailHighNum())
    {
m_ThreadPool->DeleteIdleThread(m_ThreadPool->m_IdleList.size()-m_T
hreadPool->GetInitNum());
    }
    m_WorkMutex.Unlock();
    }
}
void CWorkerThread::SetJob(CJob* job,void* jobdata)
{
    m_VarMutex.Lock();
    m_Job = job;
    m_JobData = jobdata;
    job->SetWorkThread(this);
    m_VarMutex.Unlock();
    m_JobCond.Signal();
}
void CWorkerThread::SetThreadPool(CThreadPool* thrpool)
{
    m_VarMutex.Lock();
    m_ThreadPool = thrpool;
    m_VarMutex.Unlock();
}

当线程执行任务之前首先必须判断空闲线程的数目是否低于m_AvailLow，如果低于，则必须创建足够的空闲线程，使其数目达到m_InitNum个，然后将调用MoveToBusyList()移出空闲队列，移入忙碌队列。当任务执行完毕后，其又调用MoveToIdleList()移出忙碌队列，移入空闲队列，等待新的任务。
除了Run方法之外，CWorkerThread中另外一个重要的方法就是SetJob，该方法将实际的任务赋值给线程。当没有任何执行任务即m_Job为NULL的时候，线程将调用m_JobCond.Wait进行等待。一旦Job被赋值给线程，其将调用m_JobCond.Signal方法唤醒该线程。由于m_JobCond属于线程内部的变量，每个线程都维持一个m_JobCond，只有得到任务的线程才被唤醒，没有得到任务的将继续等待。无论一个线程何时被唤醒，其都将从等待的地方继续执行m_Job->Run(m_JobData)，这是线程执行实际任务的地方。
在线程执行给定Job期间，我们必须防止另外一个Job又赋给该线程，因此在赋值之前，通过m_VarMutex进行锁定， Job执行期间，其于的Job将不能关联到该线程；任务执行完毕，我们调用m_VarMutex.Unlock()进行解锁，此时，线程又可以接受新的执行任务。
在线程执行任务结束后返回空闲队列前，我们还需要判断当前空闲队列中的线程是否高于m_AvailHigh个。如果超过m_AvailHigh，则必须从其中删除(m_ThreadPool->m_IdleList.size()-m_ThreadPool->GetInitNum())个线程，使线程数目保持在m_InitNum个。

CJob

封装了任务的基本的属性和方法，其中最重要的是Run方法

class CJob
{
private:
    int      m_JobNo;        //The num was assigned to the job
    char*    m_JobName;      //The job name
    CThread  *m_pWorkThread;     //The thread associated with the job
public:
    CJob( void );
    virtual ~CJob();

    int      GetJobNo(void) const { return m_JobNo; }
    void     SetJobNo(int jobno){ m_JobNo = jobno;}
    char*    GetJobName(void) const { return m_JobName; }
    void     SetJobName(char* jobname);
    CThread *GetWorkThread(void){ return m_pWorkThread; }
    void     SetWorkThread ( CThread *pWorkThread ){
        m_pWorkThread = pWorkThread;
    }
    virtual void Run ( void *ptr ) = 0;
};
CJob::CJob(void)
:m_pWorkThread(NULL)
,m_JobNo(0)
,m_JobName(NULL)
{
}
CJob::~CJob(){
    if(NULL != m_JobName)
    free(m_JobName);
}
void CJob::SetJobName(char* jobname)
{
    if(NULL !=m_JobName)    {
        free(m_JobName);
        m_JobName = NULL;
    }
    if(NULL !=jobname)    {
        m_JobName = (char*)malloc(strlen(jobname)+1);
        strcpy(m_JobName,jobname);
    }
}

CThreadMutex

用于线程之间的互斥。

CCondition

条件变量的封装，用于线程之间的同步。

使用示例

至此我们给出了一个简单的与具体任务无关的线程池框架。使用该框架非常的简单，我们所需要的做的就是派生CJob类，将需要完成的任务实现在Run方法中。然后将该Job交由CThreadManage去执行。下面我们给出一个简单的示例程序

class CXJob:public CJob
{
public:
    CXJob(){i=0;}
    ~CXJob(){}
    void Run(void* jobdata)    {
        printf("The Job comes from CXJOBn");
        sleep(2);
    }
};

class CYJob:public CJob
{
public:
    CYJob(){i=0;}
    ~CYJob(){}
    void Run(void* jobdata)    {
        printf("The Job comes from CYJobn");
    }
};

main()
{
    CThreadManage* manage = new CThreadManage(10);
    for(int i=0;i<40;i++)
    {
        CXJob*   job = new CXJob();
        manage->Run(job,NULL);
    }
    sleep(2);
    CYJob* job = new CYJob();
    manage->Run(job,NULL);
    manage->TerminateAll();
}

CXJob和CYJob都是从Job类继承而来，其都实现了Run接口。CXJob只是简单的打印一句”The Job comes from CXJob”，CYJob也只打印”The Job comes from CYJob”，然后均休眠2秒钟。在主程序中我们初始创建10个工作线程。然后分别执行40次CXJob和一次CYJob。

用C实现的简单线程池

condition.h

#ifndef _CONDITION_H_
#define _CONDITION_H_

#include <pthread.h>

//封装一个互斥量和条件变量作为状态
typedef struct condition
{
    pthread_mutex_t pmutex;
    pthread_cond_t pcond;
}condition_t;

//对状态的操作函数
int condition_init(condition_t *cond);
int condition_lock(condition_t *cond);
int condition_unlock(condition_t *cond);
int condition_wait(condition_t *cond);
int condition_timedwait(condition_t *cond, const struct timespec *abstime);
int condition_signal(condition_t* cond);
int condition_broadcast(condition_t *cond);
int condition_destroy(condition_t *cond);

#endif

condition.c

#include "condition.h"

//初始化
int condition_init(condition_t *cond)
{
    int status;
    if((status = pthread_mutex_init(&cond->pmutex, NULL)))
        return status;
    
    if((status = pthread_cond_init(&cond->pcond, NULL)))
        return status;
    
    return 0;
}

//加锁
int condition_lock(condition_t *cond)
{
    return pthread_mutex_lock(&cond->pmutex);
}

//解锁
int condition_unlock(condition_t *cond)
{
    return pthread_mutex_unlock(&cond->pmutex);
}

//等待
int condition_wait(condition_t *cond)
{
    return pthread_cond_wait(&cond->pcond, &cond->pmutex);
}

//固定时间等待
int condition_timedwait(condition_t *cond, const struct timespec *abstime)
{
    return pthread_cond_timedwait(&cond->pcond, &cond->pmutex, abstime);
}

//唤醒一个睡眠线程
int condition_signal(condition_t* cond)
{
    return pthread_cond_signal(&cond->pcond);
}

//唤醒所有睡眠线程
int condition_broadcast(condition_t *cond)
{
    return pthread_cond_broadcast(&cond->pcond);
}

//释放
int condition_destroy(condition_t *cond)
{
    int status;
    if((status = pthread_mutex_destroy(&cond->pmutex)))
        return status;
    
    if((status = pthread_cond_destroy(&cond->pcond)))
        return status;
        
    return 0;
}

threadpool.h

#ifndef _THREAD_POOL_H_
#define _THREAD_POOL_H_

//线程池头文件

#include "condition.h"

//封装线程池中的对象需要执行的任务对象
typedef struct task
{
    void *(*run)(void *args);  //函数指针，需要执行的任务
    void *arg;              //参数
    struct task *next;      //任务队列中下一个任务
}task_t;


//下面是线程池结构体
typedef struct threadpool
{
    condition_t ready;    //状态量
    task_t *first;       //任务队列中第一个任务
    task_t *last;        //任务队列中最后一个任务
    int counter;         //线程池中已有线程数
    int idle;            //线程池中空闲线程数
    int max_threads;     //线程池最大线程数
    int quit;            //是否退出标志
}threadpool_t;


//线程池初始化
void threadpool_init(threadpool_t *pool, int threads);

//往线程池中加入任务
void threadpool_add_task(threadpool_t *pool, void *(*run)(void *arg), void *arg);

//摧毁线程池
void threadpool_destroy(threadpool_t *pool);

#endif

threadpool.c

#include "threadpool.h"
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <time.h>

//创建的线程执行
void *thread_routine(void *arg)
{
    struct timespec abstime;
    int timeout;
    printf("thread %d is startingn", (int)pthread_self());
    threadpool_t *pool = (threadpool_t *)arg;
    while(1)
    {
        timeout = 0;
        //访问线程池之前需要加锁
        condition_lock(&pool->ready);
        //空闲
        pool->idle++;
        //等待队列有任务到来 或者 收到线程池销毁通知
        while(pool->first == NULL && !pool->quit)
        {
            //否则线程阻塞等待
            printf("thread %d is waitingn", (int)pthread_self());
            //获取从当前时间，并加上等待时间， 设置进程的超时睡眠时间
            clock_gettime(CLOCK_REALTIME, &abstime);  
            abstime.tv_sec += 2;
            int status;
            status = condition_timedwait(&pool->ready, &abstime);  //该函数会解锁，允许其他线程访问，当被唤醒时，加锁
            if(status == ETIMEDOUT)
            {
                printf("thread %d wait timed outn", (int)pthread_self());
                timeout = 1;
                break;
            }
        }
        
        pool->idle--;
        if(pool->first != NULL)
        {
            //取出等待队列最前的任务，移除任务，并执行任务
            task_t *t = pool->first;
            pool->first = t->next;
            //由于任务执行需要消耗时间，先解锁让其他线程访问线程池
            condition_unlock(&pool->ready);
            //执行任务
            t->run(t->arg);
            //执行完任务释放内存
            free(t);
            //重新加锁
            condition_lock(&pool->ready);
        }
        
        //退出线程池
        if(pool->quit && pool->first == NULL)
        {
            pool->counter--;//当前工作的线程数-1
            //若线程池中没有线程，通知等待线程（主线程）全部任务已经完成
            if(pool->counter == 0)
            {
                condition_signal(&pool->ready);
            }
            condition_unlock(&pool->ready);
            break;
        }
        //超时，跳出销毁线程
        if(timeout == 1)
        {
            pool->counter--;//当前工作的线程数-1
            condition_unlock(&pool->ready);
            break;
        }
        
        condition_unlock(&pool->ready);
    }
    
    printf("thread %d is exitingn", (int)pthread_self());
    return NULL;
    
}


//线程池初始化
void threadpool_init(threadpool_t *pool, int threads)
{
    
    condition_init(&pool->ready);
    pool->first = NULL;
    pool->last =NULL;
    pool->counter =0;
    pool->idle =0;
    pool->max_threads = threads;
    pool->quit =0;
    
}


//增加一个任务到线程池
void threadpool_add_task(threadpool_t *pool, void *(*run)(void *arg), void *arg)
{
    //产生一个新的任务
    task_t *newtask = (task_t *)malloc(sizeof(task_t));
    newtask->run = run;
    newtask->arg = arg;
    newtask->next=NULL;//新加的任务放在队列尾端
    
    //线程池的状态被多个线程共享，操作前需要加锁
    condition_lock(&pool->ready);
    
    if(pool->first == NULL)//第一个任务加入
    {
        pool->first = newtask;
    }        
    else    
    {
        pool->last->next = newtask;
    }
    pool->last = newtask;  //队列尾指向新加入的线程
    
    //线程池中有线程空闲，唤醒
    if(pool->idle > 0)
    {
        condition_signal(&pool->ready);
    }
    //当前线程池中线程个数没有达到设定的最大值，创建一个新的线程
    else if(pool->counter < pool->max_threads)
    {
        pthread_t tid;
        pthread_create(&tid, NULL, thread_routine, pool);
        pool->counter++;
    }
    //结束，访问
    condition_unlock(&pool->ready);
}

//线程池销毁
void threadpool_destroy(threadpool_t *pool)
{
    //如果已经调用销毁，直接返回
    if(pool->quit)
    {
    return;
    }
    //加锁
    condition_lock(&pool->ready);
    //设置销毁标记为1
    pool->quit = 1;
    //线程池中线程个数大于0
    if(pool->counter > 0)
    {
        //对于等待的线程，发送信号唤醒
        if(pool->idle > 0)
        {
            condition_broadcast(&pool->ready);
        }
        //正在执行任务的线程，等待他们结束任务
        while(pool->counter)
        {
            condition_wait(&pool->ready);
        }
    }
    condition_unlock(&pool->ready);
    condition_destroy(&pool->ready);
}

main.c

#include "threadpool.h"
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>

void* mytask(void *arg)
{
    printf("thread %d is working on task %dn", (int)pthread_self(), *(int*)arg);
    sleep(1);
    free(arg);
    return NULL;
}

//测试代码
int main(void)
{
    threadpool_t pool;
    //初始化线程池，最多三个线程
    threadpool_init(&pool, 3);
    int i;
    //创建十个任务
    for(i=0; i < 10; i++)
    {
        int *arg = malloc(sizeof(int));
        *arg = i;
        threadpool_add_task(&pool, mytask, arg);
        
    }
    threadpool_destroy(&pool);
    return 0;
}

用C++实现的简单线程池

最后

以上就是活力鸡为你收集整理的线程池原理前言概述创建一个具有动态伸缩性的线程池用C实现的简单线程池用C++实现的简单线程池的全部内容，希望文章能够帮你解决线程池原理前言概述创建一个具有动态伸缩性的线程池用C实现的简单线程池用C++实现的简单线程池所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。