线程池的基本概念

线程池的基本概念

不管线程池是什么东西！但是我们必须知道线程池被搞出来的目的就是：提高程序执行效率而设计出来的；

了解了线程池的目的后：我们就可以开始理解线程池：

首先回答一个问题：为什么会有线程池？

呃呃，我这么问就很奇怪，因为线程池是什么我都没说，怎么会知道为什么会有线程池呢？所以我打算带大家去思考一个场景：

当我们的程序中：有一批任务到来时候（通常该任务都是从网络来的），我们就会创建一堆线程去处理这一批任务；

虽然说创建线程的成本开销并不大，但是这里有个问题：当我们任务来到时候，你才去创建线程去处理这个任务，你不觉得这样很慢吗？

是否我们可以换个思路：假如我们有一种手段：使得任务一到来，就可以马上有线程去处理这批任务，这样是不是相对于前面等线程来到，再创建线程去处理时候快得多；

所以说：线程池就是基于上面的思路设计的；线程池就是：预先创建好一大批线程，同时线程池维护一个队列，来存放到来的任务，当队列中一旦有任务时候，预先创建好的一大批线程就可以并发处理这一批任务了；

我们抽象出一个模型：

任务派发者是谁？ 是生产者；

任务存储的队列是什么？是一个容器，数组，链表，只要是可以存放产品（数据）的东西即可；

拿任务去处理的是谁？是消费者；

所以说：线程池本质就是一个生产者消费者的模型；

而我们线程池只需要关注两个点：一个存放任务的队列，和消费队列任务的消费者即可；而生产者暂时不用关注，因为生产者是你外部搞出任务丢给线程池去使用；那么什么时候可以关心生产者呢？

也就是当我们去使用线程池的时候咯；这不就是妥妥的生产者消费者模型嘛！

线程池实现的基本思路：

在各个编程语言的语种中都有线程池的概念，并且很多语言中直接提供了线程池，作为程序猿直接使用就可以了，下面给大家介绍一下线程池的实现原理：

线程池的组成主要分为 3 个部分，这三部分配合工作就可以得到一个完整的线程池：

任务队列，存储需要处理的任务，由工作的线程来处理这些任务

通过线程池提供的 API 函数，将一个待处理的任务添加到任务队列，或者从任务队列中删除；

已处理的任务会被从任务队列中删除；

线程池的使用者，也就是调用线程池函数往任务队列中添加任务的线程就是生产者线程；

工作的线程（任务队列任务的消费者） ，N个

线程池中维护了一定数量的工作线程，他们的作用是是不停的读任务队列，从里边取出任务并处理

工作的线程相当于是任务队列的消费者角色；

如果任务队列为空，工作的线程将会被阻塞 (使用条件变量 / 信号量阻塞)；

如果阻塞之后有了新的任务，由生产者将阻塞解除，工作线程开始工作；

管理者线程（不处理任务队列中的任务），1个

它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测；

当任务过多的时候，可以适当的创建一些新的工作线程；

当任务过少的时候，可以适当的销毁一些工作的线程；

线程池的代码

1.任务队列的任务结构体

对于任务队列：

里面存放的都是函数指针，该函数指针指向的就是处理任务的函数；

同时还要维护一个任务函数的形参；

typedef struct Task
{
    void (*function)(void *args); //任务的函数指针
    void *args; //任务函数的形参
} Task;

2. 线程池的定义

线程池里面最重要的是：

一个任务队列；
多个消费者线程IDs;
一个管理者线程ID;
管理线程池的锁；
管理任务队列是否为满和空的条件变量；

还有一些其他的辅助成员变量；

struct ThreadPool
{
    Task *taskQ; //任务队列
    /*对于一个任务队列：我们需要知道以下信息*/
    int queueCapacity; //队列的容量
    int queueSize;     //当前任务的个数
    int queueFront;    //队头取任务
    int queueRear;     //队尾放任务

    /*有了任务队列后，还要有管理任务队列的线程和从任务队列拿任务的线程*/
    pthread_t managerID; //管理者线程
    /*设置为指针的目的：工作线程有多个*/
    pthread_t *threadIDs; //工作线程（也就是消费者）

    /*对于工作线程我们要知道以下这几个消息方便管理*/
    int minNum;  //最少的工作线程数
    int maxNum;  //最多的工作线程数
    int busyNum; //正在工作的线程数，也就是正在获取任务处理的线程
    int liveNum; //存货的工作线程数（也就是被唤醒的线程，却没有资格去获取任务的线程）
    int exitNum; //销毁的工作线程数（因为可能工作线程存在，但是却不工作，我们需要杀掉一些不必要的线程）

    /*  由于任务队列为临界资源：
        工作线程（消费者）可能有多个会同时竞争该资源
        同时多生产者线程之间（也就是往任务队列放任务的线程）也会竞争该资源
        所以我们要保证互斥访问线程池的任务队列
    */
    pthread_mutex_t mutexpool;    //锁整个线程池
    pthread_mutex_t mutexbusyNum; //锁增在工作线程的数量
    /*由于任务队列满，或者为空：
      生产者和消费者都需要阻塞
      所以需要条件变量，来保证
    */
    pthread_cond_t notFull;  //判断线程池是否为满
    pthread_cond_t notEmpty; //判断线程池是否为空

    /*辅助成员主要判断该线程池是否还在工作*/
    int shutdown; //判断是否需要销毁线程池，是0不销毁，是1销毁
};

线程池的头文件声明

#pragma once
#include < pthread.h >
#include < string.h >
#include < unistd.h >
#include < malloc.h >
#include< stdio.h >

typedef struct ThreadPool ThreadPool; //线程池结构体,这里声明的原因是结构体定义在线程池源文件中

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

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

//给线程池添加任务
void threadPoolAdd(ThreadPool* pool,void(*functions)(void*),void* args);

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

//获取线程池存活的线程的个数
int threadLiveNum (ThreadPool* pool);
//工作线程
void* worker (void* args);
//管理线程
void* manager (void* args);

//线程退出函数
void threadExit(ThreadPool* pool);

线程池的源文件

#include"thread_pool.h"

const int WORK_THREAD_NUMBER = 2; //管理者线程要添加的工作线程个数,和销毁的线程个数
/*
线程池：首先要有个任务队列，在C语言中，
任务队列是需要自己定义的，C++中可以直接使用容器queue
*/
//任务队列存放的任务就是一个函数指针
typedef struct Task
{
    void (*function)(void *args);
    void *args;
} Task;

//再搞出一个线程池

struct ThreadPool
{
    Task *taskQ; //任务队列
    /*对于一个任务队列：我们需要知道以下信息*/
    int queueCapacity; //队列的容量
    int queueSize;     //当前任务的个数
    int queueFront;    //队头取任务
    int queueRear;     //队尾放任务

    /*有了任务队列后，还要有管理任务队列的线程和从任务队列拿任务的线程*/
    pthread_t managerID; //管理者线程
    /*设置为指针的目的：工作线程有多个*/
    pthread_t *threadIDs; //工作线程（也就是消费者）

    /*对于工作线程我们要知道以下这几个消息方便管理*/
    int minNum;  //最少的工作线程数
    int maxNum;  //最多的工作线程数
    int busyNum; //正在工作的线程数，也就是正在获取任务处理的线程
    int liveNum; //存货的工作线程数（也就是被唤醒的线程，却没有资格去获取任务的线程）
    int exitNum; //销毁的工作线程数（因为可能工作线程存在，但是却不工作，我们需要杀掉一些不必要的线程）

    /*  由于任务队列为临界资源：
        工作线程（消费者）可能有多个会同时竞争该资源
        同时多生产者线程之间（也就是往任务队列放任务的线程）也会竞争该资源
        所以我们要保证互斥访问线程池的任务队列
    */
    pthread_mutex_t mutexpool;    //锁整个线程池
    pthread_mutex_t mutexbusyNum; //锁增在工作线程的数量
    /*由于任务队列满，或者为空：
      生产者和消费者都需要阻塞
      所以需要条件变量，来保证
    */
    pthread_cond_t notFull;  //判断线程池是否为满
    pthread_cond_t notEmpty; //判断线程池是否为空

    /*辅助成员主要判断该线程池是否还在工作*/
    int shutdown; //判断是否需要销毁线程池，是0不销毁，是1销毁
};

//************************************************************************************************

/*由于我们的线程池被创建出来时候，就必须保证存在的，
    所以我们返回值要设计为指针类型，不能是赋值拷贝的形式
    并且如何考虑线程池需要传入什么参数初始化呢？
*/
ThreadPool *threadPoolCreate(int min, int max, int queueSize)
{
    //先搞出一个线程池
    ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));
    do // do while(0)的设计是为了，假设开辟线程池，消费者线程IDs,任务队列空间失败，可以直接跳出循环统一处理释放空间
    {
        if (pool == NULL)
        {
            printf("malloc threadPool is failedn");
            break;
        }
        //搞出线程池后开始初始化里面的数据成员

        //首先先搞出消费者线程出来
        pool- >threadIDs = (pthread_t *)malloc(sizeof(pthread_t) * max);
        if (pool- >threadIDs == NULL)
        {
            printf("malloc threadIDs is failedn");
            /*如果没有do while(0)的设计，这里直接返回，那么前面的pool内存池的空间没有被释放，这就会内存泄漏了*/
            // return NULL;

            //基于上面的注释考虑，这里设计break;退出dowhile(0)然后处理
            break;
        }
        //初始化消费者线程ID
        /*这么做的目的是：在管理者线程中可以通过判断线程ID是否为0，来说明该消费者线程是否被占用*/
        memset(pool- >threadIDs, 0, sizeof(pthread_t) * max);
        //初始化线程池的其他成员属性
        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- >mutexpool, NULL) != 0 ||
            pthread_cond_init(&pool- >notEmpty, NULL) != 0 ||
            pthread_cond_init(&pool- >notFull, NULL) != 0)
        {
            perror("mutex or condition failed:");
        }
        //初始化任务队列
        pool- >taskQ = (Task *)malloc(sizeof(Task) * queueSize);
        if (pool- >taskQ == NULL)
        {
            printf("malloc taskQ is failedn");
            break;
        }
        pool- >queueCapacity = queueSize;
        pool- >queueSize = 0;
        pool- >queueFront = 0;
        pool- >queueRear = 0;
        //刚开始不关闭线程池
        pool- >shutdown = 0;

        //创建管理者线程和消费者线程
        pthread_create(&pool- >managerID, NULL, manager, (void *)pool);
        int i = 0;
        for (; i < min; ++i)
        {
            /*消费线程需要消费的是任务，
            也就是taskQ,而taskQ又是pool的一个成员属性
            所以传参时候，我们传入pool就可以获得taskQ了
            */
            pthread_create(&pool- >threadIDs[i], NULL, worker, (void *)pool);
        }

        //创建成功初始化后，那么就可以把线程池返回去了
        return pool;
    } while (0);
    //如果break出来，那么就是异常的开辟空间失败，要释放资源
    if (pool)
        free(pool);
    if (pool && pool- >threadIDs)
        free(pool- >threadIDs);
    if (pool && pool- >taskQ)
        free(pool- >taskQ);

    return NULL;
}

//判断任务队列是否为空
static int taskQIsEmpty(ThreadPool *pool)
{
    return pool- >queueSize == 0;
}
//判断线程池是否还工作
static int isShutDown(ThreadPool *pool)
{
    return pool- >shutdown == 1 ? 1 : 0;
}

//消费者线程
void *worker(void *args)
{
    ThreadPool *pool = (ThreadPool *)args;
    /*设计为死循环是：消费者要不断从任务队列拿任务来处理*/
    while (1)
    {
        pthread_mutex_lock(&pool- >mutexpool);
        //消费数据之前，要判断任务队列是否为空,空就需要挂起该线程
        while (taskQIsEmpty(pool) && !isShutDown(pool))
        {
            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 (isShutDown(pool))
        {
            pthread_mutex_unlock(&pool- >mutexpool);
            threadExit(pool);
        }
        //开始消费者拿任务
        Task task;                                              //保存任务的变量
        task.function = pool- >taskQ[pool- >queueFront].function; //获取到任务队列的任务，就是一个函数指针
        task.args = pool- >taskQ[pool- >queueFront].args;           //获取任务队列任务的函数指针参数

        //控制任务队列的指针移动
        pool- >queueFront++;
        pool- >queueFront %= pool- >queueCapacity;
        pool- >queueSize--;

        pthread_mutex_unlock(&pool- >mutexpool);
         //唤醒生产者
        pthread_cond_signal(&pool- >notFull);

        //拿到任务后就是处理任务

        // 1.处理任务前，先处理busyNum
        pthread_mutex_lock(&pool- >mutexbusyNum);
        pool- >busyNum++;
        pthread_mutex_unlock(&pool- >mutexbusyNum);

        // 2. 这里处理任务就是调用任务函数
        task.function(task.args);
        //任务处理完就释放参数的空间
        free(task.args);
        task.args = NULL;

        printf("thread %ld ending working ... n", pthread_self());
        // 3.处理完任务对其busyNum操作
        pthread_mutex_lock(&pool- >mutexbusyNum);
        pool- >busyNum--;
        pthread_mutex_unlock(&pool- >mutexbusyNum);
    }
}
//管理者线程
/*
主要是管理创建线程和销毁线程

*/
void *manager(void *args)
{
    ThreadPool *pool = (ThreadPool *)args;
    //只要线程池没关闭，那么管理者线程就一直工作
    while (!isShutDown(pool))
    {
        //自己定制的检查策略：我设置每个三秒检测
        sleep(3);

        //取出线程池任务的数量和消费者的工作线程数量
        pthread_mutex_lock(&pool- >mutexpool);
        int queueSize = pool- >queueSize;
        int liveNum = pool- >liveNum;
        pthread_mutex_unlock(&pool- >mutexpool);

        //获取忙的消费者线程数量
        pthread_mutex_lock(&pool- >mutexbusyNum);
        int busyNum = pool- >busyNum;
        pthread_mutex_unlock(&pool- >mutexbusyNum);

        //开始管理线程
        // 1.添加消费者线程
        /*制定添加规则（也是自己设定的）
            任务的个数 > 存活的线程个数 && 存活的线程个数 < 最大的线程个数
        */
        if (queueSize > liveNum && liveNum < pool- >maxNum)
        {
            pthread_mutex_lock(&pool- >mutexpool); //这个锁主要是操作了liveNum这个资源

            int counter = 0; // counter表示要添加的消费者线程数量
            //遍历 消费者线程IDs数组，看看哪个位置可以放入新添加的线程
            int i = 0;
            for (; i < pool- >maxNum &&
                   counter < WORK_THREAD_NUMBER &&
                   pool- >liveNum < pool- >maxNum;
                 i++)
            {
                //为0表示消费者线程数组的位置可以放入线程ID
                if (pool- >threadIDs[i] == 0)
                {
                    pthread_create(&pool- >threadIDs[i], NULL, worker, pool);
                    counter++;
                    liveNum++;
                }
            }
            pthread_mutex_unlock(&pool- >mutexpool);
        }

        //由于线程过多，可能要进行销毁
        // 2. 销毁消费者线程
        /*
             销毁线程的策略：
              存活的线程数量 >忙的线程数量*2 && 存活线程数量 >最小线程数量
        */
        if (liveNum > busyNum * 2 && liveNum > pool- >minNum)
        {
            pthread_mutex_lock(&pool- >mutexpool);
            pool- >exitNum = WORK_THREAD_NUMBER;
            pthread_mutex_unlock(&pool- >mutexpool);

            //让工作者线程去自杀
            /*如何让他自杀呢？
              由于线程池有多余的消费者线程不工作
              我们可以通过唤醒消费者线程，让他去自己消亡
            */
            int i = 0;
            for (; i < WORK_THREAD_NUMBER; i++)
            {
                pthread_cond_signal(&pool- >notEmpty);
            }
        }
    }
}

//线程退出函数
void threadExit(ThreadPool *pool)
{
    pthread_t tid = pthread_self();
    int i = 0;
    //遍历消费者线程的线程个数，找到退出线程的ID
    for (; i < pool- >maxNum; i++)
    {
        if (pool- >threadIDs[i] == tid)
        {
            pool- >threadIDs[i] = 0;
            printf("threadExit()消费者线程 :%ld exit...n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}
static int taskQisFull(ThreadPool* pool)
{
    return pool- >queueCapacity == pool- >queueSize;
}
//给线程池添加任务
void threadPoolAdd(ThreadPool* pool,void(*function)(void*),void* args)
{
    pthread_mutex_lock(&pool- >mutexpool); 
    //生产者线程：任务队列满要阻塞自己
    while(taskQisFull(pool) && !isShutDown(pool))
    {
        pthread_cond_wait(&pool- >notFull,&pool- >mutexpool);
    }
    if(isShutDown(pool))
    {
        pthread_mutex_unlock(&pool- >mutexpool);
        return ;
    }

    //添加任务
    pool- >taskQ[pool- >queueRear].function = function;
    pool- >taskQ[pool- >queueRear].args = args;

    pool- >queueRear++;
    pool- >queueRear %= pool- >queueCapacity;
    pool- >queueSize++;

    pthread_mutex_unlock(&pool- >mutexpool); 
    //唤醒work线程：
    pthread_cond_signal(&pool- >notEmpty);
}

//获取线程池工作线程的个数
int threadBusyNum (ThreadPool* pool)
{
    pthread_mutex_lock(&pool- >mutexbusyNum);
    int busyNum = pool- >busyNum;
    pthread_mutex_unlock(&pool- >mutexbusyNum);
    return busyNum;

}

//获取线程池存活的线程的个数
int threadLiveNum (ThreadPool* pool)
{
    pthread_mutex_lock(&pool- >mutexpool);
    int liveNum = pool- >liveNum;
    pthread_mutex_unlock(&pool- >mutexpool);
    return liveNum;
}

//销毁线程池
int threadPoolDestroy(ThreadPool* pool)
{
    if(pool == NULL)
    {
        return -1;
    }
    //关闭线程池
    pool- >shutdown = 1;


    //唤醒阻塞的消费者
    //存活的线程有多少就唤醒多少
    int i = 0;
    for(;i < pool- >liveNum;i++)
    {
        pthread_cond_signal(&pool- >notEmpty);
    }
    pthread_join(pool- >managerID,NULL);

    //释放资源
    if(pool- >taskQ )
        free(pool- >taskQ);
    if(pool- >threadIDs)
        free(pool- >threadIDs);

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

    free(pool);
    pool = NULL;

    return 0;

}

线程池测试代码

#include"thread_pool.h"

//任务处理函数
void taskFunction(void* args)
{
    int num = *(int*)args;
    printf("thread: %ld is working,number:%dn",pthread_self(),num);
    sleep(1);
}
int main()
{
    //创建线程池
    ThreadPool* pool = threadPoolCreate(3,10,20);

    //往线程池里面放任务
    int i = 0;
    for(; i< 20; i++)
    {
        int *num = (int*)malloc(sizeof(int));
        *num = i+1;
        threadPoolAdd(pool,taskFunction,(void*)num);
    }

    sleep(10);

    threadPoolDestroy(pool);
    return 0;
}

测试线程池结果

由于我的测试代码：只搞了3个工作线程（消费者线程），任务队列大小为20，并且搞了20个任务队列进去，所以线程池就会有三个工作线程在抢夺任务工作！