C++实现线程池一

线程池

当并发的线程数量增多，每个线程执行一个很短的任务之后销毁，频繁的创建和销毁线程就会降低系统效率。
线程池就是希望先一个线程执行完一个任务，不被销毁，去执行其他任务。线程池的任务处理过程中，将任务添加到队列，在创建线程后自动执行这些任务。

线程池的组成：

• 任务队列：存储需要处理的任务，由工作的线程来处理这些任务，
• 工作线程：读取任务队列，执行任务，充当消费者角色。
• 管理线程：周期性的管理工作线程，根据任务量，适当微调工作线程的个数。

---

C语言线程池

文件树：

liuchang@DESKTOP-LIUCHANG:~/codetest/THreads/cthreadpool$ tree
.
├── main.cpp
├── main.exe
├── Makefile
├── run.sh
├── threadpool.cpp
└── threadpool.h

0 directories, 6 files

Makefile

CC = g++ -std=c++14
CFLAGS = -g -Wall

src=$(wildcard *.cpp) #搜索当前目录下的cpp文件 # 受够繁文缛节了，不再生成中间文件，直接生成可执行文件
target=main.exe
$(target):$(src)
	$(CC) $(CFLAGS) $(src) -o $(target) -lpthread

.PHONY: clean
clean:
	rm -f main *.o

main.cpp

#include <iostream>
#include "threadpool.h"
#include <unistd.h>

using std::cout;
using std::endl;

void taskFunc(void* arg){
    int num = *(int*)arg;
    cout<<"thread "<< pthread_self() <<" is working, number = "<<num<<endl; 
    sleep(1);
}

int main(){
    
    // 创建线程池
    ThreadPool* pool = threadPoolCreate(3,10,100);// 做小运行线程数为3，最大运行线程数为10，线程池中队列大小为100
    for(int i=0;i<100;i++){
        int* num = (int*)malloc(sizeof(int));
        *num = i + 100;
        threadPoolAdd(pool, taskFunc, num);
    }
    sleep(30);

    threadPoolDestroy(pool);
    return 0;
}

threadpool.h

#pragma once

typedef struct ThreadPool ThreadPool;
// 创建线程池并初始化
ThreadPool* threadPoolCreate(int min,int max,int queueSize);

// 销毁线程池
int threadPoolDestroy(ThreadPool* pool);

// 给线程池加任务
void threadPoolAdd(ThreadPool* pool,void (*func)(void*), void* arg);

// 获取线程池中工作的线程的个数
int threadPoolBusyNum(ThreadPool* pool);

// 获取线程池中活着的线程的个数
int threadPoolAliveNum(ThreadPool* pool);


void* worker(void* arg);
void* manager(void* arg);
void threadExit(ThreadPool* pool);

threadpool.cpp

#include "threadpool.h"
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

const int NUMBER = 2;

// 任务结构体
typedef struct Task
{
    void (*function)(void *arg);
    void *arg;
} Task;

// 线程池结构体
struct ThreadPool
{
    // 任务队列
    Task *taskQ;
    int queueCapacity; // 容量
    int queueSize;     // 当前任务个数
    int queueFront;    // 队头-> 取数据
    int queueRear;     // 队尾-> 放数据

    pthread_t managerID;       // 管理者线程ID
    pthread_t *threadIDs;      // 工作者线程ID
    int minNum;                // 最小线程数目
    int maxNum;                // 最大线程数目
    int busyNum;               // 忙线程个数
    int liveNum;               // 存活线程个数
    int exitNum;               // 要杀死的线程个数
    pthread_mutex_t mutexpool; // 锁整个线程池
    pthread_mutex_t mutexBusy; // 锁busyNum
    pthread_cond_t notFull;    // 任务队列是否满了
    pthread_cond_t notEmpty;   // 任务队列是否空了

    int shutdown; // 是不是要销毁线程池，销毁为1，不销毁为0
};

ThreadPool *threadPoolCreate(int min, int max, int queueSize)
{
    ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));
    do
    {
        if (pool == NULL)
        {
            printf("malloc threadpool failed..\n");
            break;
        }

        pool->threadIDs = (pthread_t *)malloc(sizeof(pthread_t) * max);
        if (pool->threadIDs == NULL)
        {
            printf("malloc threadIDs failed..\n");
            break;
        }
        memset(pool->threadIDs, 0, max); // 线程全部初始化为0，存储线程的时候，根据线程id去判断，如果是无符号整型就是已经占用了，如果是0就还没有被占用
        pool->minNum = min;
        pool->maxNum = max;
        pool->busyNum = 0;
        pool->liveNum = min; // 和最小个数相等
        pool->exitNum = 0;

        if (pthread_mutex_init(&pool->mutexpool, NULL) != 0 ||
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpty, NULL) != 0 ||
            pthread_cond_init(&pool->notFull, NULL) != 0)
        {
            printf("mutex or cond init failed..\n");
            break;
        }

        // 任务队列
        pool->taskQ = (Task*)malloc(sizeof(Task)*queueSize);
        pool->queueCapacity = queueSize;
        pool->queueSize = 0;
        pool->queueFront = 0;
        pool->queueRear = 0;

        pool->shutdown = 0;

        pthread_create(&pool->managerID, NULL, manager, pool); // 管理线程调用管理函数
        for (int i = 0; i < min; i++)
        {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool); // 工作线程调用worker函数
        }
        return pool; //正常执行到这里
    } while (0);

    // 如果出现问题，就释放相关的数据和资源。
    if(pool && pool->threadIDs){
        free(pool->threadIDs);
    }
    if(pool && pool->taskQ){
        free(pool->taskQ);
    }
    if(pool)
        free(pool);

    return NULL;
}

void* worker(void* arg){
    ThreadPool* pool = (ThreadPool*) arg;
    while(1){
        pthread_mutex_lock(&pool->mutexpool); //对线程池上锁
        while(pool->queueSize == 0 && !pool->shutdown){
            // 阻塞工作线程
            pthread_cond_wait(&pool->notEmpty,&pool->mutexpool);
            // 判断是不是要销毁线程
            if (pool->exitNum>0)
            {
                pool->exitNum--;
                if (pool->liveNum > pool->minNum)
                {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexpool);
                    threadExit(pool);
                } 
            }
        }
        // 判断线程是否被关闭了
        if(pool->shutdown){
            pthread_mutex_unlock(&pool->mutexpool);
            threadExit(pool);
        }

        // 从任务队列中取出一个任务
        Task task;
        task.function = pool->taskQ[pool->queueFront].function;
        task.arg = pool->taskQ[pool->queueFront].arg;
        // 移动头结点
        pool->queueFront = (pool->queueFront+1)%pool->queueCapacity; //当队列头指针移动到尾部的时候，可以移动到头部，形成一个循环队列。
        pool->queueSize--;
        // 唤醒生产者
        pthread_cond_signal(&pool->notFull);
        pthread_mutex_unlock(&pool->mutexpool); // 用完以后解锁

        // 对工作线程加一，用到专门的锁
        printf("thread %ld start working...\n", pthread_self());
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum++;
        pthread_mutex_unlock(&pool->mutexBusy);

        task.function(task.arg);
        free(task.arg);
        task.arg = NULL;

        printf("thread %ld end working...\n", pthread_self());
        // 任务执行完成，--，用到专门的锁
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexBusy);
        
    }
    return NULL;
}

void* manager(void* arg){
    ThreadPool* pool = (ThreadPool*)arg;
    while (!pool->shutdown)
    {
        // 每隔3s检测一次
        sleep(3);
        
        // 取出线程池中任务的数量和当前线程的数量
        pthread_mutex_lock(&pool->mutexpool);
        int queueSize = pool->queueSize;
        int liveNum = pool->liveNum;
        pthread_mutex_unlock(&pool->mutexpool);

        // 取出忙线程个数。
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);

        // 添加线程
        // 任务的个数> 存活线程的个数 && 存活线程个数< 最大线程个数
        if(queueSize>liveNum && liveNum<pool->maxNum){
            pthread_mutex_lock(&pool->mutexpool);
            int counter = 0;
            for(int i =0 ; i<pool->maxNum && counter<NUMBER && pool->liveNum<pool->maxNum; ++i){
                if(pool->threadIDs[i]==0){
                    pthread_create(&pool->threadIDs[i],NULL,worker,pool);
                    counter++;
                    pool->liveNum++;
                }
            }
            pthread_mutex_unlock(&pool->mutexpool);
        }

        // 销毁线程
        // 忙的线程*2 < 存活的线程数 && 存活的线程>最小线程数
        if(busyNum*2 < liveNum && liveNum > pool->minNum){
            pthread_mutex_lock(&pool->mutexpool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexpool);
            // 让工作的线程自杀
            for (int i = 0; i < NUMBER; ++i)
            {
                pthread_cond_signal(&pool->notEmpty); //唤醒工作线程
            }
        }
    }
    
    return NULL;
}

void threadExit(ThreadPool* pool){
    pthread_t tid = pthread_self();
    for (int i = 0; i < pool->maxNum; ++i)
    {
        if (pool->threadIDs[i] == tid)
        {
            pool->threadIDs[i] = 0;// 修改为0以后，方便后续判断是否可以创建新线程
            printf("threadExit() called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

void threadPoolAdd(ThreadPool* pool,void (*func)(void*), void* arg){
    pthread_mutex_lock(&pool->mutexpool);
    while(pool->queueSize == pool->queueCapacity && !pool->shutdown){
        // 阻塞生产者线程
        pthread_cond_wait(&pool->notFull,&pool->mutexpool);
    }

    if(pool->shutdown){
        pthread_mutex_unlock(&pool->mutexpool);
        return;
    }
    // 添加任务
    pool->taskQ[pool->queueRear].function = func;
    pool->taskQ[pool->queueRear].arg = arg;
    pool->queueRear = (pool->queueRear+1)% pool->queueCapacity;
    pool->queueSize++;

    pthread_cond_signal(&pool->notEmpty);
    pthread_mutex_unlock(&pool->mutexpool);
}

int threadPoolBusyNum(ThreadPool* pool){
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

int threadPoolAliveNum(ThreadPool* pool)
{
    pthread_mutex_lock(&pool->mutexpool);
    int aliveNum = pool->liveNum;
    pthread_mutex_unlock(&pool->mutexpool);
    return aliveNum;
}

int threadPoolDestroy(ThreadPool* pool){
    if (pool == NULL)
    {
        return -1;
    }

    // 关闭线程池
    pool->shutdown = 1;
    // 阻塞回收管理者线程
    pthread_join(pool->managerID, NULL);

    // 唤醒阻塞的消费者线程
    for (int i = 0; i < pool->liveNum; ++i)
    {
        pthread_cond_signal(&pool->notEmpty);
    }

    // 释放堆内存
    if (pool->taskQ)
    {
        free(pool->taskQ);
    }
    if (pool->threadIDs)
    {
        free(pool->threadIDs);
    }

    pthread_mutex_destroy(&pool->mutexpool);
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);

    free(pool);
    pool = NULL;

    return 0;
}

---

参考列表：
https://subingwen.cn/linux/threadpool/