【线程池C++实现】

操作系统：Ubuntu 16.04

运行方式：

        g++ -std=c++11 thread_pool_example.cpp -o thread_pool_example -lpthread

        ./thread_pool_example

        最近时间较紧，不能详细展开叙述。具体原理及代码逻辑可以参见注释，十分详细。日后一定补充完善。

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
using namespace std;

// 任务类
class Task
{
public:
    Task(int a, int b)
    {
        this->a = a;
        this->b = b;
    }

    void runTask()
    {
        cout << "task " << a << " + " << b << " = " << a + b << endl;
    }

private:
    int a, b;
};

// 线程池类
class ThreadPool
{
public:
    // 构造函数
    ThreadPool(int threadNum, int maxTask)
    {
        this->threadNum = threadNum;                   // threadNum 是线程池中线程的数量
        this->maxTask = maxTask;                       // maxTask 是求队列中最多允许的、等待处理的请求的数量
        threadsArr = nullptr;                          // 描述线程池的数组，其大小为 threadNum
        stop = false;                                  // 停止标志符
        if (pthread_mutex_init(&locker, nullptr) != 0) // 初始化互斥锁
            throw exception();
        if (sem_init(&sem, 0, 0) != 0) // 初始化信号量
            throw exception();

        if (threadNum <= 0 || maxTask <= 0)
            throw exception();

        threadsArr = new pthread_t[threadNum]; // 线程池申请空间 and 初始化
        if (!threadsArr)
            throw exception();

        
        for (int i = 0; i < threadNum; ++i)
        {
            if (pthread_create(threadsArr + i, nullptr, worker, this) != 0)
            {
                delete[] threadsArr;
                throw exception();
            }

            
            if (pthread_detach(threadsArr[i]))
            {
                delete[] threadsArr;
                throw exception();
            }
        }
    }

    // 析构函数
    ~ThreadPool()
    {
        delete[] threadsArr;
        sem_destroy(&sem);
        pthread_mutex_destroy(&locker);
        stop = true;
    }

    // 往请求队列中添加任务
    bool append(Task *task)
    {
        // 操作工作队列时，一定要加锁，因为它被所有线程共享
        pthread_mutex_lock(&locker);
        if (taskList.size() > maxTask)
        {
            pthread_mutex_unlock(&locker);
            return false;
        }

        taskList.push_back(task);
        pthread_mutex_unlock(&locker);
        sem_post(&sem); // v 操作，释放一个信号量，以此提醒run()有任务要处理
        return true;
    }

private:
    // 工作线程运行的函数，它不断从工作队列中取出任务并执行之，内部访问私有成员函数run，完成线程处理要求。
    static void *worker(void *arg)
    {
        ThreadPool *pool = (ThreadPool *)arg; // 将参数强转为线程池类，调用成员方法
        pool->run();                          // 调用类的动态函数 run()
        return pool;
    }

    // 取出工作队列头部任务执行
    void run()
    {
        while (!stop)
        {
            sem_wait(&sem);
            pthread_mutex_lock(&locker);
            if (taskList.empty())
            {
                pthread_mutex_unlock(&locker);
                continue;
            }

            Task *task = taskList.front();
            taskList.pop_front();
            pthread_mutex_unlock(&locker);
            if (!task)
                continue;
            task->runTask();
        }
    }

private:
    int threadNum;
    int maxTask;
    pthread_t *threadsArr;
    pthread_mutex_t locker;
    sem_t sem;
    list taskList;
    bool stop;
};

int main()
{
    cout << "=== Tasks Begin ===" << endl;

    // 构造线程池
    int threadNum = 2, maxTask = 10;
    ThreadPool *pool = new ThreadPool(threadNum, maxTask);

    // 这里把工作队列单独在主函数中构造出来，是为了后续的 delete，防止空指针的出现
    list tasks(10);
    for (int i = 0; i < 10; ++i)
    {
        Task *tmpTask = new Task(i, i);
        tasks.push_back(tmpTask);
        pool->append(tmpTask);
    }

    sleep(5);

    // 防止空指针出现
    for (int i = 0; i < 5; ++i)
    {
        delete tasks.front();
        tasks.front() = nullptr;
        tasks.pop_front();
    }

    cout << "=== Tasks End ===" << endl;
}