Linux 執行緒池的概念與實作

執行緒池概念：一種執行緒使用模式，執行緒過多會帶來調度開銷，進而影響快取區域性和整體性能，而執行緒池維護著多個執行緒，等待著監督管理者分配可并發執行的任務，這避免了在處理短時間任務時創建與銷毀執行緒的代價，執行緒池不僅能夠保證內核的充分利用，還能防止過分調度，

應用場景：1、需要大量的執行緒來完成任務，且完成任務的時間比較短；2、對性能要求苛刻的應用；3、接受突發性的大量請求，但不至于使服務器因此產生大量執行緒的應用
不使用執行緒池的情況：若是一個資料請求的到來伴隨一個執行緒去創建，就會產生一定的風險以及一些不必要的消耗，
1、執行緒若不限制數量的創建，在峰值壓力下，執行緒創建過多，資源耗盡，有程式崩潰的風險；
2、處理一個短時間任務時，會有大量的資源用于執行緒的創建與銷毀成本上，
功能：執行緒池是使用了已經創建好的執行緒進行回圈處理任務，避免了大量執行緒的頻繁創建與銷毀的時間成本

如何實作一個執行緒池

執行緒池 = 大量執行緒 + 任務緩沖佇列

困難與解決方案：在創建執行緒時，都是伴隨創建執行緒的入口函式，一旦創建就無法改變，導致執行緒池進行任務處理的方式過于單一，靈活性太差，若任務佇列中的任務，不僅僅是單純的資料，而是包含處理任務方法在內的資料，這時候，執行緒池的執行緒是一條普通的執行流，只需要使用傳入的方法去處理資料即可，這樣子就可以提高執行緒池的靈活性

代碼實作流程：定義一個任務類Task，成員變數有要處理的資料_data和處理資料的方法_handler，成員函式有設定要處理資料和處理方式的函式setTask，還有一個處開始處理資料的函式run(創建執行緒時傳入的方法，由于創建執行緒必須有入口函式，這里用run封裝所有的處理方式，讓所有執行緒都將run置為入口函式，就提高了執行緒池的靈活性)，再定義一個執行緒池類ThreadPool，成員變數有定義執行緒池中執行緒的最大數量thr_max，一個任務緩沖佇列_queue，一個互斥量_mutex，用于實作對緩沖佇列的安全性，一個條件變數_cond，用于實作執行緒池中執行緒的同步，

threadpool.hpp檔案

//threadpool.hpp
#include <iostream>
#include <cstdio>
#include <queue>
#include <stdlib.h>
#include <pthread.h>
using namespace std;

typedef void (*handler_t)(int);
#define MAX_THREAD 5
//任務類
class ThreadTask
{
    public:
        ThreadTask()
        {

        }
        //將資料與處理方式打包在一起
        void setTask(int data, handler_t handler)
        {
            _data = data;
            _handler = handler;
        }
        //執行任務函式
        void run()
        {
            return _handler(_data);
        }
    private:
        int _data;//任務中處理的資料
        handler_t _handler;//處理任務方式
};

//執行緒池類
class ThreadPool
{
    public:
        ThreadPool(int thr_max = MAX_THREAD)
            :_thr_max(thr_max)
        {
            pthread_mutex_init(&_mutex, NULL);
            pthread_cond_init(&_cond, NULL);
            for (int i = 0; i < _thr_max; i++)
            {
                pthread_t tid;
                int ret = pthread_create(&tid, NULL, thr_start, this);
                if (ret != 0)
                {
                    printf("thread create error\n");
                    exit(-1);
                }
            }
        }
        ~ThreadPool()
        {
            pthread_mutex_destroy(&_mutex);
            pthread_cond_destroy(&_cond);
        }
        bool taskPush(ThreadTask &task)
        {
            pthread_mutex_lock(&_mutex);
            _queue.push(task);
            pthread_mutex_unlock(&_mutex);
            pthread_cond_signal(&_cond);
            return true;
        }
        //類的成員函式，有默認的隱藏引數this指標
        //置為static，沒有this指標，
        static void *thr_start(void *arg)
        {
            ThreadPool *p = (ThreadPool*)arg;
            while (1)
            {
                pthread_mutex_lock(&p->_mutex);
                while (p->_queue.empty())
                {
                    pthread_cond_wait(&p->_cond, &p->_mutex);
                }
                ThreadTask task;
                task =p-> _queue.front();
                p->_queue.pop();
                pthread_mutex_unlock(&p->_mutex);
                task.run();//任務的處理要放在解鎖之外
            }
            return NULL;
        }
    private:
        int _thr_max;//執行緒池中執行緒的最大數量
        queue<ThreadTask> _queue;//任務緩沖佇列
        pthread_mutex_t _mutex; //保護佇列操作的互斥量
        pthread_cond_t _cond; //實作從佇列中獲取結點的同步條件變數
};

main.cpp

//main.cpp
#include <unistd.h>
#include "threadpool.hpp"

//處理方法1
void test_func(int data)
{
    int sec = (data % 3) +1;
    printf("tid:%p -- get data:%d, sleep:%d\n", pthread_self(), data, sec);
    sleep(sec);
}

//處理方法2
void tmp_func(int data)
{
    printf("tid:%p -- tmp_func\n", pthread_self());
    sleep(1);
}

int main()
{
    ThreadPool pool;
    for (int i = 0; i < 10; i++)
    {
        ThreadTask task;
        if (i % 2 == 0)
        {
            task.setTask(i, test_func);
        }
        else
        {
            task.setTask(i, tmp_func);
        }
        pool.taskPush(task);
    }
    sleep(1000);
    return 0;
}

運行結果：執行緒池最多有5個執行緒，標注的每種顏色對應的是同一個執行緒，這樣子就能完成通過幾個執行緒，完成多個任務，而不是多個執行緒完成多個任務，創建和銷毀的時間開銷也節省了不少