linux下线程池原理及实现

什么是线程池？

诸如web服务器、数据库服务器、文件服务器和邮件服务器等许多服务器应用都面向处理来自某些远程来源的大量短小的任务。构建服务器应用程序的一个过于简单的模型是：每当一个请求到达就创建一个新的服务对象，然后在新的服务对象中为请求服务。但当有大量请求并发访问时，服务器不断的创建和销毁对象的开销很大。所以提高服务器效率的一个手段就是尽可能减少创建和销毁对象的次数，特别是一些很耗资源的对象创建和销毁，这样就引入了“池”的概念，“池”的概念使得人们可以定制一定量的资源，然后对这些资源进行复用，而不是频繁的创建和销毁。

线程池是预先创建线程的一种技术。线程池在还没有任务到来之前，创建一定数量的线程，放入空闲队列中。这些线程都是处于睡眠状态，即均为启动，不消耗CPU，而只是占用较小的内存空间。当请求到来之后，缓冲池给这次请求分配一个空闲线程，把请求传入此线程中运行，进行处理。当预先创建的线程都处于运行状态，即预制线程不够，线程池可以自由创建一定数量的新线程，用于处理更多的请求。当系统比较闲的时候，也可以通过移除一部分一直处于停用状态的线程。

简单线程池的实现

1、线程池，顾名思义，就是要创建很多线程。创建线程的函数pthread_create()应该是最容易被想到的。有线程创建就要有线程退出pthread_exit()，在线程退出前，如果线程没有设置pthread_detach()属性，那么显然要回收线程资源pthread_join()。当然咯，可能要获取线程的ID值pthread_self()。

2、第一步创建了线程线程池linux，刚开始线程是不做事情的，初始化好了，就等待。等待当然不会是while(1)这种函数，因为那样太消耗CPU资源。容易想到的等待自然是使用条件变量的等待pthread_cond_wait()，这个函数干两件事情，第一件是对解除与形参mutex对应的互斥锁，然后是重新加锁，为的是在线程将任务放入任务队列的一个缓冲。任务放入完成后，再加锁，这样不会影响其他任务获取加锁的权利。因此，在调用该函数之前，会自然会想到加互斥锁。初始化互斥锁函数pthread_mutex_init()，反初始化互斥锁函数pthread_mutex_destroy()，加锁函数pthread_mutex_lock()，解锁函数pthread_mutex_unlock()，稍微再细化一点，可能会用到尝试解锁pthread_mutex_trylock()。

3.、实现了上面二步，一个线程池的框架就初步搭起来了。当然没法用，因为真正干事情的线程全部在等待中，注意不应该是超时的等待pthread_cond_timewait（）。要使处于阻塞状态的线程干事情，得用信号去唤醒它pthread_cond_signal（），“打鸟”的一个函数，开一枪，总会把这只鸟吵醒，但具体是那一只，看那只最先在那排队了（上面已经说了pthread_cond_wait（）函数的等待队列问题）。当然也可以想到“打鸟惊群”的函数pthread_cond_broadcast（），打一枪，无论打没打着，一群鸟都飞走了。

4、有了上面的基础，接下来就重点关注任务部分了。当然线程数量有限，上面已经说了，是固定的数目。因此任务大于线程数时，排队是难免的了。因此创建了一个任务队列，队列中的每一项代表一个任务。任务队列的节点最简单的模型就是一个处理任务的回掉函数void* (*callback_function)(void *arg)。指向函数的指针，参数是个指针，返回值也是个指针。具体的函数和参数则需要另外写函数定义。没次调用完线程处理完这个任务，就需要把它从任务队列中删除。进入任务队列的任务数也不能无限多，因此也设为一个比线程数稍微大个几个的一个固定值。

5、线程动态创建：一个线程退出后（在任务执行失败时，就有可能会退出），主线程要能检测到，然后动态创建一个新的线程，以维持线程池中线程总数不变。可以通过pthread_join()阻塞等待回收子线程资源，但是这就意味着主线程在阻塞状态下干不了其他工作，因此考虑使用线程信号，在子线程结束时，给主线程用pthread_kill()发送一个SIGUSR1信号，在主线程接收到此信号时，通过调用注册函数signal()或sigaction()函数注册的函数创建一个新的线程。

//ThreadPool设计    
void *thread_routine(void *args);    
class ThreadPool    
{    
    friend void *thread_routine(void *args);    
private:    
    //回调函数类型    
    typedef void *(*callback_t)(void *);    
    //任务结构体    
    struct task_t    
    {    
        callback_t run; //任务回调函数    
        void *args;     //任务函数参数    
    };    
public:    
    ThreadPool(int _maxThreads = 36, unsigned int _waitSeconds = 2);    
    ~ThreadPool();    
    //添加任务接口    
    void addTask(callback_t run, void *args);    
private:    
    void startTask();    
private:    
    Condition ready;                //任务准备就绪或线程池销毁通知    
    std::queue taskQueue; //任务队列    
    unsigned int maxThreads;        //线程池最多允许的线程数    
    unsigned int counter;           //线程池当前线程数    
    unsigned int idle;              //线程池空闲线程数    
    unsigned int waitSeconds;       //线程可以等待的秒数    
    bool         quit;              //线程池销毁标志    
};    

// 线程入口函数    
// 这其实就相当于一个消费者线程, 不断的消费任务(执行任务)    
void *thread_routine(void *args)    
{    
    //将子线程设置成为分离状态, 这样主线程就可以不用jion    
    pthread_detach(pthread_self());    
    printf("*thread 0x%lx is starting...\n", (unsigned long)pthread_self());    
    ThreadPool *pool = (ThreadPool *)args;    
    //等待任务的到来, 然后执行任务    
    while (true)    
    {    
        bool timeout = false;    
        pool->ready.lock();    
        //当处于等待的时候, 则说明空闲的线程多了一个    

        ++ pool->idle;    
        //pool->ready中的条件变量有三个作用:    
        // 1.等待任务队列中有任务到来    
        // 2.等待线程池销毁通知    
        // 3.确保当等待超时的时候, 能够将线程销毁(线程退出)    
        while (pool->taskQueue.empty() && pool->quit == false)    
        {    
            printf("thread 0x%lx is waiting...\n", (unsigned long)pthread_self());    
            //等待waitSeconds    
            if (0 != pool->ready.timedwait(pool->waitSeconds))    
            {    
                //如果等待超时    
                printf("thread 0x%lx is wait timeout ...\n", (unsigned long)pthread_self());    
                timeout = true;    
                //break出循环, 继续向下执行, 会执行到下面第1个if处    
                break;    
            }    
        }    
        //条件成熟(当等待结束), 线程开始执行任务或者是线程销毁, 则说明空闲线程又少了一个    
        -- pool->idle;    
        // 状态3.如果等待超时(一般此时任务队列已经空了)    
        if (timeout == true && pool->taskQueue.empty())    
        {    
            -- pool->counter;    
            //解锁然后跳出循环, 直接销毁线程(退出线程)    
            pool->ready.unlock();    
            break;    
        }    
        // 状态2.如果是等待到了线程的销毁通知, 且任务都执行完毕了    
        if (pool->quit == true && pool->taskQueue.empty())    
        {    
            -- pool->counter;    
            //如果没有线程了, 则给线程池发送通知    
            //告诉线程池, 池中已经没有线程了    
            if (pool->counter == 0)    
                pool->ready.signal();    
            //解锁然后跳出循环    
            pool->ready.unlock();    
            break;    
        }    
        // 状态1.如果是有任务了, 则执行任务    
        if (!(pool->taskQueue.empty()))    
        {    
            //从队头取出任务进行处理    
            ThreadPool::task_t *t = pool->taskQueue.front();    
            pool->taskQueue.pop();    
            //执行任务需要一定的时间    
            //解锁以便于其他的生产者可以继续生产任务, 其他的消费者也可以消费任务    
            pool->ready.unlock();    
            //处理任务    
            t->run(t->args);    
            delete t;    
        }    

    }    
    //跳出循环之后, 打印退出信息, 然后销毁线程    
    printf("thread 0x%lx is exiting...\n", (unsigned long)pthread_self());    
    pthread_exit(NULL);    
}    

//addTask函数    
//添加任务函数, 类似于一个生产者, 不断的将任务生成, 挂接到任务队列上, 等待消费者线程进行消费    
void ThreadPool::addTask(callback_t run, void *args)    
{    
    /** 1. 生成任务并将任务添加到"任务队列"队尾 **/    
    task_t *newTask = new task_t {run, args};    
    ready.lock();   //注意需要使用互斥量保护共享变量    
    taskQueue.push(newTask);    
    /** 2. 让线程开始执行任务 **/    
    startTask();    
    ready.unlock();//解锁以使任务开始执行    
}  

//线程启动函数    
void ThreadPool::startTask()    
{    
    // 如果有等待线程, 则唤醒其中一个, 让它来执行任务    
    if (idle > 0)    
        ready.signal();    
    // 没有等待线程, 而且当前先线程总数尚未达到阈值, 我们就需要创建一个新的线程    
    else if (counter < maxThreads)    
    {    
        pthread_t tid;    
        pthread_create(&tid, NULL, thread_routine, this);    
        ++ counter;    
    }    
}    

//析构函数    
ThreadPool::~ThreadPool()    
{    
    //如果已经调用过了, 则直接返回    
    if (quit == true)    
        return;    
    ready.lock();    
    quit = true;    
    if (counter > 0)    
    {    
        //对于处于等待状态, 则给他们发送通知,    
        //这些处于等待状态的线程, 则会接收到通知,    
        //然后直接退出    
        if (idle > 0)    
            ready.broadcast();    
        //对于正处于执行任务的线程, 他们接收不到这些通知,    
        //则需要等待他们执行完任务    
        while (counter > 0)    
            ready.wait();    
    }    
    ready.unlock();    
}    

（编辑：温州站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!