【Linux】线程池实现

1??线程池概念

线程池是一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
线程池的应用场景：

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.

线程池示例：

1. 创建固定数量线程池，循环从任务队列中获取任务对象，
2. 获取到任务对象后，执行任务对象中的任务接口

2??线程池代码样例

以下为线程池代码：

#pragma once

#include <iostream>
#include <queue>
#include <pthread.h>
#include <ctime>


template<class T>
class ThreadPool
{
    private:
        std::queue<T> _q;//任务队列
        pthread_mutex_t _lock;
        pthread_cond_t _cond;//有任务时提醒线程执行任务
        static ThreadPool<T>* _instance;
        ThreadPool()
        {}   
    public:
        static ThreadPool<T>* getInstance()//单例模式,饿汉模式,静态成员
        {
            static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
            if(nullptr == _instance)//双if提高效率，因为只有第一次获取_instance时才会实例化，后续无需再加锁实例化_instance
            {
                pthread_mutex_lock(&mtx);
                if(nullptr == _instance)
                {
                    _instance = new ThreadPool<T>();
                }
                pthread_mutex_unlock(&mtx);
            }
            return _instance;
        }
        void MutexInit()
        {
            pthread_mutex_init(&_lock, nullptr);
        }
        void MutexLock()
        {
            pthread_mutex_lock(&_lock);
        }
        void MutexUnLock()
        {
            pthread_mutex_unlock(&_lock);
        }
        bool IsEmpty()
        {
            return _q.size() == 0 ? true : false;
        }
        void ThreadWait()
        {
            pthread_cond_wait(&_cond, &_lock);
        }
        void ThreadWakeUp()
        {
            pthread_cond_signal(&_cond);
        }
        void PopTask(T* out)//取任务
        {
            //此处不应加锁，应为取任务的时候是带着锁的，若此时申请锁，会出现死锁现象
            *out = _q.front();
            _q.pop();
        }
        //类内部的成员方法都有隐含的this参数，因此要加上static修饰
        static void* Routine(void* args)
        {
            ThreadPool<T>* tp = (ThreadPool<T>*) args;//接收this指针
            pthread_detach(pthread_self());//线程分离
            while(true)
            {
                tp->MutexLock();//加锁，访问临界区_q
                while(tp->IsEmpty())//任务队列为空，挂起等待
                {
                    tp->ThreadWait();
                }
                //到此处说明有任务
                T t;
                tp->PopTask(&t);
                tp->MutexUnLock();//退出临界区_q
                t();
            }
            return nullptr;
        }
        void ThreadPoolInit(int num)//初始化线程池
        {
            pthread_t p[num];
            for(int i = 0; i < num; i++)
            {
                pthread_create(p + i, nullptr, Routine, this);//将this指针作为参数传入
            }
        }
        void PushTask(const T& in)
        {
            //分配任务
            MutexLock();
            _q.push(in);
            MutexUnLock();
            ThreadWakeUp();//唤醒线程完成任务
        }
        ~ThreadPool()
        {
            pthread_mutex_destroy(&_lock);
        }
};


template<class T>
ThreadPool<T>* ThreadPool<T>::_instance = nullptr;

3??部分问题与细节

下面分享一些在编写该单例模式线程池代码遇到的一些问题与细节：

🔸类成员函数参数列表中隐含的this指针

我们在初始化线程池的这部分代码，需要创建若干线程来完成其所需要执行的任务，这些线程的例程函数形式为void *(*start_routine) (void *) ，其参数列表中仅有一个参数void*，而如果将这个例程函数定义成成员函数，会有一个隐含的this指针参数，导致形式不一致，因此需要将该例程函数用static修饰为静态的。
又因为静态成员函数只能访问静态成员变量，故我们需要在创建线程时将this指针通过参数传递给例程函数，这样才能在例程函数中使用this指针访问类中的成员变量。

🔸单例模式

我们这个线程池设计成了单例模式，并且采用的是饿汉模式，即服务启动后只有在用到线程池这个功能时才会创建对象。而在单例模式创建对象时，由于只有第一次创建对象时对象指针为nullptr，故判断是否要创建对象指针的时候可以在加锁之前再进行一次判断提高效率，而无需每次都要先加锁再判断。

🔸一个失误导致的bug

在线程取任务的接口设计时，我因为这里需要访问任务队列这个临界区给这个过程加上了锁，但是实际上在调用这个接口的时候其实线程就已经申请加了锁，而且两次申请的为同一把锁，就导致出现了线程在已经持有一把锁的情况下又去申请这把锁，从而产生了死锁。

4??调用线程池完成任务

任务类：
实现x 与 y 的+ - * / % 五种运算。

#pragma once
#include <iostream>

class Task//x op y = z
{
    private:
        int x;
        int y;
        char op;//+-*/%
    public:
        Task(){}
        Task(int _x, int _y, char _op)
        :x(_x),
        y(_y),
        op(_op)
        {}
        void operator()()
        {
            int z = -1;
            switch (op)
            {
            case /* constant-expression */'+':
                /* code */
                z = x + y;
                break;
            case '-':
                z = x - y;
                break;
            case '*':
                z = x * y;
                break;
            case '/':
                if(0 != y)
                    z = x / y;
                else
                    std::cout << "warning: div zero error" << std::endl;
                break;
            case '%':
                if(0 != y)
                    z = x % y;
                else
                    std::cout << "warning: div zero error" << std::endl;
                break;
            default:
                    std::cout << "unkonwn operator" << std::endl;
                break;
            }
            std::cout << "[" << pthread_self() << "] handler task: " << x << " " << op << " " << y << " = " << z << std::endl;
        }
        ~Task(){}
};

主函数：

#include "thread_pool.hpp"
#include "Task.hpp"
#include <string>
#include <unistd.h>

int main()
{
    srand((unsigned long)time(nullptr));
    ThreadPool<Task>* tp = ThreadPool<Task>::getInstance();
    tp->ThreadPoolInit(5);

    const std::string s = "+-*/%";
    while(true)
    {
        int x = rand() % 50;
        int y = rand() % 50;
        char op = s[rand() % 5];
        Task t(x, y, op);
        tp->PushTask(t);
        sleep(1);
    }
    return 0;
}

结果如上，左侧为线程池中的线程每隔一秒取出任务并执行，右侧为线程池的情况。