[C++知识库]C语言实现线程池：什么？你是小白？我也是，一起肝了

为了在Linux C聊天室项目里植入线程池，特地肝了线程池。后面完善了聊天室也会分享出来。现将代码分享，备注思路详细，本小白的代码小白应该也能看懂！最后，欢迎纠错指正！

直接上代码

/*************************************************************************
	> File Name: Btree.cpp
	> Author: 念念
	> Mail: 2845906049@qq.com 
	> Created Time: 2021年08月21日 星期六 17时42分40秒
    > Function: 创建线程池并使用线程池打印1~100
 ************************************************************************/

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

//两个宏函数，节点插入删除操作
#define LL_ADD(item,list)   \
    item->prev = NULL;    \
    item->next = list;    \
    if(list != NULL)list->prev = item;    \
    list = item;  \

#define LL_REMOVE(item,list)  \
    if(item->prev != NULL) item->prev->next = item->next;    \
    if(item->next != NULL) item->next->prev = item->prev;    \
    if(list == item) list = item->next;    \
    item->prev = item->next = NULL;    \

//线程结构体
struct NWORKER{

    pthread_t thread; //线程
    struct NMANAGER *pool; //线程池指针
    int num;  //线程编号
    int terminate;  // 停止标志，terminate置1线程退出

    struct NWORKER *prev;
    struct NWORKER *next;
};

//任务结构体
struct NJOB{
    
    void (*func)(void *arg);  //工作函数，使用void* arg是为了接口能适配不同的参数
    void *user_data;  //传入的数据
    int num; //任务序号

    struct NJOB *prev;
    struct NJOB *next;
};

//线程池管理结构体
typedef struct NMANAGER{
    
    struct NWORKER *workers;  //线程队列
    struct NJOB *jobs;  //任务队列

    int jobs_num;

    pthread_cond_t jobs_cond;  //条件变量
    pthread_mutex_t jobs_mutex;  //锁

}nThreadPool;

//测试代码为打印1~100，这里用全局变量count表示
static int count = 0;

//线程回调函数
static void *nThreadCallback(void *arg)
{
    struct NWORKER *worker = (struct NWORKER*)arg;

    while(1)
    {
        pthread_mutex_lock(&worker->pool->jobs_mutex);  //上锁
        while(NULL == worker->pool->jobs) //任务队列为空则持续等待
        {   
            if(worker->terminate) break; //terminate置为1终止线程，先跳出
            pthread_cond_wait(&worker->pool->jobs_cond,&worker->pool->jobs_mutex); //持续等待有任务信号

        }
        if(worker->terminate) //terminate置为1终止线程
        {
            pthread_mutex_unlock(&worker->pool->jobs_mutex); //解锁
            break;
        }

        struct NJOB *job = worker->pool->jobs;  //提取任务
        LL_REMOVE(job,worker->pool->jobs);  //从任务队列拿出

        

        count++;
        job->user_data = &count;  //在这里为传入打印的count赋值，在函数内部就不需要再加锁和条件变量了

        printf("\033[35mcallBack：打印：%d \n\033[0m",count);
        printf("callBack：线程 %d 收到新任务 %d ,开始执行\n",worker->num,job->num);
        job->func((void *)job);  //执行工作函数
        printf("callBack：线程 %d 完成任务 %d \n\n",worker->num,job->num);
        usleep(500000); //没别的意思，就是一步步的看执行过程
        pthread_mutex_unlock(&worker->pool->jobs_mutex); //解锁
        usleep(100000);  //如果这里没加睡眠就一直是同一个线程在占有资源执行，让他休息一下不跟其他资源抢占
        //不知道其他解决办法，求大佬赐教
    }

    printf("线程退出， ptid：%ld\n",pthread_self());
    free(worker); //释放线程内存
    pthread_exit(NULL);  //终止
}

int nThreadPoolCreate(nThreadPool *pool,int numWorkers)
{

    if(numWorkers < 1) numWorkers = 1;  //校验参数

    if(NULL == pool) return -1;
    memset(pool, 0, sizeof(nThreadPool));

    pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
    memcpy(&pool->jobs_cond, &blank_cond, sizeof(pthread_cond_t));

    pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
    memcpy(&pool->jobs_mutex, &blank_mutex, sizeof(pthread_mutex_t));

    pool->jobs_num = 0;

    for(int i = 1; i <= numWorkers; i++)
    {
        struct NWORKER *worker = (struct NWORKER*)malloc(sizeof(struct NWORKER));
        if(worker == NULL)
        {
            perror("malloc");
            return -1;
        }

        memset(worker, 0, sizeof(struct NWORKER));
        worker->pool = pool; //绑定线程池
        worker->num = i;  //线程编号

        int ret = pthread_create(&worker->thread,NULL,nThreadCallback, worker);
        if(ret)
        {
            perror("pthread_create");
            free(worker);
            return -3;
        }
        
        LL_ADD(worker,pool->workers);  //加入到线程队列
    }
    return 0;
}

int nThreadPoolDestroy(nThreadPool *pool)
{

    struct NWORKER *worker = NULL;

    for(worker = pool->workers; worker != NULL; worker = worker->next)
    {
        worker->terminate = 1;  //所有线程退出标志置为一，所有线程退出
    }

    pthread_mutex_lock(&pool->jobs_mutex);
    pthread_cond_broadcast(&pool->jobs_cond);
    pthread_mutex_unlock(&pool->jobs_mutex);

}

//加任务
void nThreadPoolPush(nThreadPool *pool,struct NJOB *job)
{
    pthread_mutex_lock(&pool->jobs_mutex);

    LL_ADD(job,pool->jobs);
    printf("\n线程池里添加了一个新任务, 任务号：%d\n",job->num);
    pthread_cond_signal(&pool->jobs_cond);  //发送有任务信号

    pthread_mutex_unlock(&pool->jobs_mutex);
}


//以上为API
//以下为测试代码

#if 1

// static pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;;
// static pthread_cond_t count_cond = PTHREAD_COND_INITIALIZER;


void func(void *arg)
{   
    struct NJOB *job = (struct NJOB *)arg;
    int count = *(int *)job->user_data;

    printf("\033[36mfunc：");
    printf("  打印count：%d ,任务号为：%d\n\033[0m",count,job->num);
}

int main()
{
    nThreadPool *pool = (nThreadPool *)malloc(sizeof(nThreadPool));
    
    if(0 != nThreadPoolCreate(pool,10)) //创建一个线程池
    {
        perror("nThreadPoolCreate");
        exit(1);
    }

    for(int i = 1; i <= 100; i++)
    {
        struct NJOB *job = (struct NJOB *)malloc(sizeof(struct NJOB));
        job->num = i;
        job->func = func;
        nThreadPoolPush(pool,job); //将任务加入执行队列
    }

    while(pool->jobs != NULL);  //等待，直到任务全部完成
     usleep(100000);//这里适当sleep能保证所有任务完成才开始销毁线程池
    printf("\n\n\n所有任务已完成！\n");
    nThreadPoolDestroy(pool);  //销毁线程池
    usleep(100000);//这里适当sleep能保证在程序结束前所有线程正常退出，用pthread_join更严谨
    // pthread_mutex_destroy(&count_mutex);
    // pthread_cond_destroy(&count_cond);
    return 0;
}

#endif

#Ubuntu下运行结果（截取部分）：