C++自定义实现线程池

本C++线程池由C语言版改编而来

1.任务执行类或结构体

线程池中的线程主要执行的对象

class Task{
public:
    void (*func)(void* arg);
    void* arg;
};

2.线程池类或结构体

线程池类中的成员属性

// 线程池类或结构体
class ThreadPoolAttr{
public:
    Task* taskQueue; // 任务队列
    int queueCapacity; // 任务队列容量
    int queueSize;   // 任务队列当前大小
    int front; // 入队下标
    int rear;  // 出队下标
    pthread_t managerId; // 管理者线程数量
    pthread_t *wordThreadIDs; // 工作线程
    int wtMinNum; // 工作线程最小数量  
    int wtMaxNum; // 工作线程最大数量
    int wtCurNum; // 当前线程数量
    int wtbusyNum; // 处于工作中的线程数量
    int wtExitNum; // 需要清除的线程数量
    bool is_exit; // 退出标志
    pthread_mutex_t mutexPool; // 线程池互斥变量
    pthread_mutex_t mutexBusy; 
    pthread_cond_t condEmpty;  // 任务队列为空，组塞
    pthread_cond_t condFull;   // 任务队列满了，阻塞
};

3.完整代码

#include 
using namespace std;
#include 
#include 
#include 
#include 
#include 
#define EMPTY_STATE 0
#define REDUCE_THREAD 2 // 每次减少两个
#define INCREASE_THREAD 2 // 每次增加两个
#define MAX_THREAD_NUM 20 // 最大线程数

class Task{
public:
    void (*func)(void* arg);
    void* arg;
};

// 线程池类或结构体
class ThreadPoolAttr{
public:
    Task* taskQueue; // 任务队列
    int queueCapacity; // 任务队列容量
    int queueSize;   // 任务队列当前大小
    int front; // 入队下标
    int rear;  // 出队下标
    pthread_t managerId; // 管理者线程数量
    pthread_t *wordThreadIDs; // 工作线程
    int wtMinNum; // 工作线程最小数量  
    int wtMaxNum; // 工作线程最大数量
    int wtCurNum; // 当前线程数量
    int wtbusyNum; // 处于工作中的线程数量
    int wtExitNum; // 需要清除的线程数量
    bool is_exit; // 退出标志
    pthread_mutex_t mutexPool; // 线程池互斥变量
    pthread_mutex_t mutexBusy; 
    pthread_cond_t condEmpty;  // 任务队列为空，组塞
    pthread_cond_t condFull;   // 任务队列满了，阻塞
};


class ThreadPool{
private:
    ThreadPoolAttr* pool;
public:
     
    ThreadPool(int queueCapacity, int wtMinNum, int wtMaxNum){
        pool = new ThreadPoolAttr;
        if(pool == NULL){
            cout << "create pool fail" << endl;
            return;
        }
        pool->queueCapacity = queueCapacity;
        pool->wtMinNum = wtMinNum;
        pool->wtMaxNum = wtMaxNum;
        pool->wtCurNum = wtMaxNum;
        pool->queueSize = 0;
        pool->wtbusyNum = 0;
        pool->wtExitNum = 0;
        pool->is_exit = false;
        pthread_mutex_init(&pool->mutexPool, NULL);
        pthread_mutex_init(&pool->mutexBusy, NULL);
        pthread_cond_init(&pool->condEmpty, NULL);
        pthread_cond_init(&pool->condFull, NULL);
        pool->taskQueue = new Task[queueCapacity]; // 创建任务队列数组
        pool->wordThreadIDs = (pthread_t*)malloc(wtMaxNum*sizeof(pthread_t)); // 创建工作线程
        for(int i = 0; i < wtMaxNum; i++){
            pool->wordThreadIDs[i] = 0;
        }
         // 创建管理线程
        pthread_create(&pool->managerId, NULL, manager, pool);
        // 创建工作线程
        for(int i = 0; i < pool->wtMaxNum; i++){
            pthread_create(&pool->wordThreadIDs[i], NULL, worker, pool);
        }

    }
     
    void addTask(void (*func)(void* arg), void *arg){
        // 加线程池锁
        pthread_mutex_lock(&pool->mutexPool);
        // 任务队列为满状态
        if(pool->queueSize == pool->queueCapacity){
            pthread_cond_wait(&pool->condFull, &pool->mutexPool); // 队列满则阻塞
        }
        int rear = pool->rear;
        pool->taskQueue[rear].func = func;
        pool->taskQueue[rear].arg = arg;
        pool->queueSize++; // 现存任务数量加1
        pool->rear = (pool->rear+1)%pool->queueCapacity; // 下标后移
        pthread_cond_signal(&pool->condEmpty); // 唤醒不为空
        // 解锁
        pthread_mutex_unlock(&pool->mutexPool);
    }
    
    bool release(){
        pool->is_exit = true;
    }
    // 析构函数释放内存
    ~ThreadPool(){
        // if(pool == NULL){
        //     return;
        // }
        for(int i = 0, j = 0; i < pool->wtMaxNum; i++){
            while(pool->wordThreadIDs[i] != 0){
                // j = 0;
                // while(j++ < 10000);
                pthread_join(pool->wordThreadIDs[i], NULL);
            }
        }
        pthread_join(pool->managerId, NULL);
        // 释放任务内存
        if(pool->taskQueue){
            delete pool->taskQueue;
        }
        // 释放线程id内存
        free(pool->wordThreadIDs);
        pthread_mutex_destroy(&pool->mutexPool);
        pthread_mutex_destroy(&pool->mutexBusy);
        pthread_cond_destroy(&pool->condEmpty);
        pthread_cond_destroy(&pool->condFull);
        delete pool;
    }
private:
    // 工作线程的主要作用是将将任务队列中的任务取出，然后执行任务
    static void* worker(void* arg){
        ThreadPoolAttr* pool = (ThreadPoolAttr*)arg; // 类型强转
        while(!pool->is_exit){
            pthread_mutex_lock(&pool->mutexPool);
            if(pool->queueSize == EMPTY_STATE){
                pthread_cond_wait(&pool->condEmpty, &pool->mutexPool); // 队列为空则阻塞
            }
            // 判断是否有退出线程
            if(pool->wtExitNum != 0 || pool->is_exit){
                pthread_t cur_id = pthread_self(); // 获取当前线程id
                for(int i = 0; i < pool->wtMaxNum; i++){
                    if(pool->wordThreadIDs[i] == cur_id){
                        pool->wordThreadIDs[i] = 0;
                        break;
                    }
                }
                if(pool->wtExitNum != 0){
                    pool->wtExitNum--; // 退出量减1
                }
                cout << "线程id:"<< pthread_self() << " 被释放！！！" << endl;
                pthread_mutex_unlock(&pool->mutexPool);
                pthread_exit(NULL);
            }
            int front = pool->front;
            void (*func)(void* func_arg) = pool->taskQueue[front].func;
            void* func_arg = pool->taskQueue[front].arg;
            pool->front = (pool->front + 1)%pool->queueCapacity;
            pool->queueSize--; // 任务总数减1
            pthread_cond_signal(&pool->condFull); // 唤醒满队列
            // 解锁
            pthread_mutex_unlock(&pool->mutexPool);

            pthread_mutex_lock(&pool->mutexBusy);
            pool->wtbusyNum++; // 工作中线程加1
            pthread_mutex_unlock(&pool->mutexBusy);
            func(func_arg); // 执行任务
            pthread_mutex_lock(&pool->mutexBusy);
            pool->wtbusyNum--; // 工作中线程减1
            pthread_mutex_unlock(&pool->mutexBusy);
        }
        return NULL;
    }
    static void* manager(void* arg){
        ThreadPoolAttr* pool = (ThreadPoolAttr*)arg;
        while(1){
            sleep(1);
            pthread_mutex_lock(&pool->mutexPool);
            // cout << "管理线程:" << pthread_self() << endl;
            // 减少线程数量
            if(pool->wtbusyNum*2 < pool->wtCurNum && pool->wtCurNum - REDUCE_THREAD > pool->wtMinNum){
                pool->wtExitNum = REDUCE_THREAD;
                pool->wtCurNum -= REDUCE_THREAD;
                for(int i = 0; i < REDUCE_THREAD; i++){
                    pthread_cond_signal(&pool->condEmpty); // 唤醒空阻塞
                }
            }
            // 增加线程数量
            if(pool->queueSize > pool->wtCurNum && pool->wtCurNum + INCREASE_THREAD <= pool->wtMaxNum){
                int n = INCREASE_THREAD;
                for(int i = 0; i < pool->wtMaxNum; i++){
                    if(pool->wordThreadIDs[i] == 0 && n > 0){
                        pthread_create(&pool->wordThreadIDs[i], NULL, worker, pool);
                        n--;
                    }
                    if(n == 0){
                        break;
                    }
                }
                pool->wtCurNum += INCREASE_THREAD; 
            }
            pthread_mutex_unlock(&pool->mutexPool);
            if(pool->is_exit){
                pthread_cond_broadcast(&pool->condEmpty); // 唤醒全部阻塞线程
                pthread_cond_broadcast(&pool->condFull); // 唤醒全部阻塞线程
                break;
            }
        }
        return NULL;
    }
};
void add_test(void* arg){
    cout << "线程id:" << pthread_self() << endl;
    sleep(2);
}
int main()
{
    ThreadPool *threadpool = new ThreadPool(100, 5, 10);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    threadpool->addTask(add_test, NULL);
    cout << "Hello World" << endl;
    sleep(10); // 等待10秒 ，注意添加 unistd.h 头文件
    threadpool->release(); // 释放线程池
    delete threadpool;
    return 0;
}

4.执行结果