手写简易线程池-C语言实现

前置知识

1.C语言指针等语法
2.linux下的C语言多线程编程
3.队列，循环队列

一、为什么需要线程池

1.一般的创建线程的方式，创建一个执行完任务后就销毁。
2.若任务执行时间短，任务线程并发数量高，频繁创建线程和销毁线程需要时间。导致浪费系统资源，降低系统效率。

问题：所以说，有没有一种办法可以让线程可以重复使用，就是执行完一个任务，并不销毁，而是可以继续执行其他任务呢？

二、什么是线程池

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

在各个编程语言的语种中都有线程池的概念，并且很多语言中直接提供了线程池（~~但好像C语言没有，原始人~~ ），作为程序猿直接使用就可以了。但我们为了学习，还是得自己手搓一个。

三、线程池实现原理

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

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

通过线程池提供的API函数，将一个待处理的任务添加到任务队列，或者从任务队列中删除
已处理的任务会被从任务队列中删除
线程池的使用者，也就是调用线程池函数往任务队列中添加任务的线程就是生产者线程

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

线程池中维护了一定数量的工作线程, 他们的作用是是不停的读任务队列, 从里边取出任务并处理
工作的线程相当于是任务队列的消费者角色
如果任务队列为空, 工作的线程将会被阻塞 (使用条件变量/信号量阻塞)
如果阻塞之后有了新的任务, 由生产者将阻塞解除, 工作线程开始工作

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

它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测
当任务过多的时候, 可以适当的创建一些新的工作线程
当任务过少的时候, 可以适当的销毁一些工作的线程

四、编写思路：

先定义任务队列结构体，然后先线程池结构体，然后编写创建线程池函数，把对应的参数传递给线程池，再最后线程池地址。
编写工作者函数与管理者函数，工作函数首先看队列是否为空，是的话否需要阻塞线程，再看是否要关闭线程池，都不是那就取任务并执行并循环这个过程。管理者函数先取出线程池的一些变量，然后判断是否需要添加或删除线程。删除线程通过一个exitNum变量传递，让线程自己退出。
销毁线程池，把线程池关闭变量置为ture，唤醒所有的阻塞的工作线程，再等待管理线程释放，最后释放堆内存的地址。
剩下就是一些接口函数了，生产者函数，判断是否队列为空，是的则将任务放入队列，否则阻塞。添加线程池变量接口函数。

五、线程池C语言源码

threadpool.h

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <pthread.h> //线程库
#include <stdlib.h>  //malloc
#include <unistd.h>  //unix POXIS库
#include <stdio.h>   //printf
#include <string.h>  //memset

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

// 线程池，结构体
typedef struct ThreadPool_t
{
    Task_t *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_t;


//外置接口
//创建线程池
ThreadPool_t *threadPoolCreate(int min, int max, int queueSize);
//销毁线程池
int threadPoolDestroy(ThreadPool_t *pool);
//给线程池添加任务
void threadPoolAdd(ThreadPool_t* pool, void(*func)(void*), void* arg);
//获取线程池中的工作的线程的个数
int threadPoolBusyNum(ThreadPool_t* pool);
//获取线程池中活着的线程的个数
int threadPoolAliveNum(ThreadPool_t* pool);

#endif

threadpool.c

#include <pthread.h> //线程库
#include <stdlib.h>  //malloc
#include <unistd.h>  //unix std POSIX库

#include "threadpool.h"

#define NUMBER 2

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

ThreadPool_t *threadPoolCreate(int min, int max, int queueSize)
{
    // 堆中开辟空间用于存放 线程池
    ThreadPool_t *pool = (ThreadPool_t *)malloc(sizeof(ThreadPool_t));
    do
    {
        if (pool == NULL)
        {
            perror("malloc threadpool failed...\n");
            break;
        }

        //
        pool->threadIDs = (pthread_t *)malloc(sizeof(pthread_t) * max);
        if (pool->threadIDs == NULL)
        {
            printf("malloc threadIDs fail...\n");
            break;
        }
        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->mutexBusy, NULL) != 0 || pthread_cond_init(&pool->notEmpty, NULL) != 0 || pthread_cond_init(&pool->notFull, NULL) != 0)
        {
            printf("mutex or condition init fail...\n");
            break;
        }

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

        // 创建管理者线程
        pthread_create(&pool->managerID, NULL, manager, pool);
        // 创建工作者线程，并将pool结构体传入给worker线程，以便传入参数
        for (int i = 0; i < min; ++i)
        {
            pthread_create(&pool->threadIDs[i], NULL, worker, pool);
        }
        return pool;
    } while (0);

    // 如果malloc有一个失败，则释放成功malloc资源
    if (pool && pool->threadIDs)
        free(pool->threadIDs);
    if (pool && pool->taskQ)
        free(pool->taskQ);
    if (pool)
        free(pool);

    return NULL;
}

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

    // 关闭线程池
    pool->shutdown = 1;

    // 唤醒所有阻塞的消费者线程
    pthread_cond_broadcast(&pool->notEmpty);

    // pthread_mutex_lock(&pool->mutexPool);
    //  唤醒所有阻塞的消费者线程，让其自动退出线程
    //  for (int i = 0; i < pool->liveNum; ++i)
    //  {
    //      pthread_cond_signal(&pool->notEmpty);
    //  }
    //  pthread_mutex_unlock(&pool->mutexPool);

    printf("wake up all working threads...\n");

    while (pool->liveNum > 0)
    {
        usleep(1000);
    }

    // 阻塞回收管理者线程
    pthread_join(pool->managerID, NULL);
    printf("manager thread exit...\n");

    // 释放堆内存
    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;
    printf("threadpool destroy success...\n");
    return 0;
}

void threadPoolAdd(ThreadPool_t *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_t *pool)
{
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

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

void *worker(void *arg)
{
    ThreadPool_t *pool = (ThreadPool_t *)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_t 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);
        // 问题：如何把任务函数的返回值给到使用者？

        //  开始工作，arg参数是调用者给的，堆内存，用完之后及得free并指向NULL
        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_t *pool = (ThreadPool_t *)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++;
                    printf("threadpool add new thread...%d\n", 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_t *pool)
{
    pthread_t tid = pthread_self();
    pthread_mutex_lock(&pool->mutexPool); // 加锁
    for (int i = 0; i < pool->maxNum; ++i)
    {
        if (pool->threadIDs[i] == tid)
        {
            pool->threadIDs[i] = 0;
            pool->liveNum--;
            printf("threadExit() called, %ld exiting...\n", tid);
            break;
        }
    }
    pthread_mutex_unlock(&pool->mutexPool);
    pthread_exit(NULL);
}

测试代码main.c

#include "threadpool.h"

void taskFunc(void *arg)
{
    int num = *(int *)arg;
    printf("thread %ld is working, number = %d\n",
           pthread_self(), num);
    sleep(1);
}

int main()
{
    // 创建线程池
    ThreadPool_t *pool = threadPoolCreate(3, 10, 50);
    for (int i = 0; i < 50; i++)
    {
        int *num = (int *)malloc(sizeof(int));
        *num = i + 50;
        threadPoolAdd(pool, taskFunc, num);
    }

    sleep(12);

    threadPoolDestroy(pool);
    printf("threadpool destroy...\n");
    return 0;
}

BUG调试查询：

多写了一个 liveNum-- ，导致线程销毁不完全，程序一直退不出去。
一般什么问题出问题就对症下药，去仔细得去检查就行了。

成功的运行的结果：

六、优化思考

1.任务队列的任务是否应该有返回结果？答：可以在Taks_t 类型里加上返回参数与同步变量与锁，并由调用线程释放。以此来达到线程同步的效果，或者用回调函数等。
2.malloc，堆内存谁创建谁释放。
3.只读变量不用加锁
4.读写锁使用是否能优化性能？
5.队列是否能改为链表，更具有灵活性？

AI的建议：

建议增加的优化点：
可以添加线程池状态检查，防止重复初始化或销毁
建议添加任务队列为空或满时的等待超时机制
可以考虑添加任务优先级机制
建议添加线程池运行状态统计功能
可以考虑添加动态调整线程池大小的策略
其他建议：
建议添加详细的错误日志
可以考虑使用条件变量的超时等待来替代简单的sleep
建议添加线程池暂停/恢复功能
可以考虑添加任务取消机制
建议增加线程异常处理机制
性能优化建议：
可以使用无锁队列来减少锁竞争
考虑使用线程本地存储(TLS)来减少共享数据访问
可以实现任务批处理机制提高效率
考虑使用工作窃取算法来优化负载均衡

开源c语言线程池项目：
https://github.com/Pithikos/C-Thread-Pool

文章参考：手写线程池 – C语言版
作者: 苏丙榅
链接: https://subingwen.cn/linux/threadpool/#1-%E7%BA%BF%E7%A8%8B%E6%B1%A0%E5%8E%9F%E7%90%86