C语言简易版线程池

C语言版简易线程池

文章目录

C语言版简易线程池
- 背景
- 线程池的优点
- 线程池实现的图示
- 代码剖析
- - 结构体定义
  - 线程池初始化
  - 线程回调函数实现
  - 销毁所有线程
  - 添加任务功能
- 可能会出的错误
- 思考题
- 完整代码

背景

若同时有十万个客户端向服务器请求数据，那么我们所需要的资源：

posix标准的线程8M，则1G内存可以开128个线程，16G内存只能开16*128 = 2048个线程，则需要800G左右的内存，显然耗费硬件资源过大，所以需要一种解决方案。

线程池的优点

可以重用已存在的线程，避免线程太多，使得内存耗尽
避免创建和销毁线程的代价
任务与执行相分离

线程池实现的图示

线程池主要由三部分构成:任务队列、执行队列（线程队列）和管理组件

代码剖析结构体定义

任务队列和线程队列都是采用了双向链表来实现：

//任务类型
struct nTask{
    //任务实际工作函数
    void (*task_func)(void *arg); 
    void *user_data; //task_func的参数

    //任务队列用双向列表组织起来
    struct nTask *prev;
    struct nTask *next;
};


//线程（执行）类型
struct nWorker{
    pthread_t threadid;
    struct nManager *manager;
    int terminate; //是否终止的标识.1为终止

    struct nWorker *prev;
    struct nWorker *next;
};

链表的插入删除操作分别用宏实现：

//支持层对链表的操作
#define LIST_INSERT(item, list) do{ 
    item->prev = NULL;              
    item->next = list;              
    if((list) != NULL) (list)->prev = item; 
    (list) = item;                  
}while(0)

#define LIST_REMOVE(item, list) do{                         
    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;                         
}while(0)

**执行队列和任务队列之间的管理组件就是我们所谓的线程池，**结构体定义如下：

typedef struct nManager{
    struct nTask *tasks; //任务队列
    struct nWorker *workers;//执行队列

    pthread_mutex_t mutex;
    pthread_cond_t cond;
}ThreadPool;

线程池初始化

线程池初始化就是对除任务队列tasks外（任务时外界添加进来的）的其他成员进行初始化，在线程池中创建给定数量的线程：

//对线程池pool进行初始化，nWorker是线程池中线程数量(即聘请的柜员数量)
int nThreadPoolCreate(ThreadPool *pool, int numWorkers)
{
    if(pool == NULL)    return -1;
    if(numWorkers < 1)  numWorkers = 1;

    //pool必须要初始化为空，如果是垃圾值下面memcpy会出现段错误
    memset(pool, 0x00, sizeof(ThreadPool));

    //相当于pthread_cond_init()进行初始化
    pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
    memcpy(&pool->cond, &blank_cond, sizeof(pthread_cond_t));

    //pthread_mutex_init(&pool->mutex, NULL); // 初始化互斥锁，下方代码功能相同
    pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
    memcpy(&pool->mutex, &blank_mutex, sizeof(pthread_mutex_t));


    int i = 0;
    //创建线程
    for(i = 0;i < numWorkers; ++i)
    {
        struct nWorker *worker = (struct nWorker *)malloc(sizeof(struct nWorker));
        if(worker == NULL)
        {
            perror("malloc");
            return -2;
        }
        memset(worker, 0x00, sizeof(struct nWorker));
        worker->manager = pool;//方便线程回调函数的使用

        //创建线程，成功返回0
        //线程回调函数为nThreadPoolCallback
        //参数为：worker
        int ret = pthread_create(&worker->threadid, NULL, nThreadPoolCallback, worker);
        if(ret)
        {
            perror("pthread_create");
            free(worker);
            return -3;
        }

        //printf("LIST_INSERTn");
        //将线程插入线程链表(队列)头
        LIST_INSERT(worker, pool->workers);
    }

    return 0;
}

线程回调函数实现

线程回调函数的主要功能就是循环等待任务的到来，有任务就执行，没有任务就循环等待：

static void *nThreadPoolCallback(void *arg)
{
    //worker指向某一个线程
    struct nWorker *worker = (struct nWorker *)arg;
    //printf("nThreadPoolCallbackn");

    //有任务执行任务，没任务循环等待
    while(1)
    {
        //这里也能看出nWorker中添加了manager成员的用处，就不用再传nManager对象了
        pthread_mutex_lock(&worker->manager->mutex);
        //无任务时，阻塞等待
        while(worker->manager->tasks == NULL)
        {
            //worker->terminate非0表明所有线程都要销毁了
            if(worker->terminate)   break;  
            //条件等待 
            //会先解除互斥锁让其他线程能够执行，然后阻塞，
            //当被pthread_cond_signal唤醒时，则会解除阻塞并加互斥锁
            pthread_cond_wait(&worker->manager->cond, &worker->manager->mutex);
        } 

        if(worker->terminate)
        {
            //防止死锁
            pthread_mutex_unlock(&worker->manager->mutex);
            break;
        }

        //任务执行，将任务从任务队列中取出
        struct nTask *task = worker->manager->tasks; 
        LIST_REMOVE(task, worker->manager->tasks);
        
        pthread_mutex_unlock(&worker->manager->mutex);
        
        //执行任务
        task->task_func(task);
    }
    free(worker);

}

销毁所有线程

销毁所有线程就是让每一个线程的terminate属性为1，表明要退出，当对应线程回调函数执行到对于该属性判断的时候就会退出循环。

int nThreadPoolDestroy(ThreadPool *pool, int nWorker)
{
    struct nWorker *worker = NULL;
    for(worker = pool->workers;worker != NULL; worker = worker->next)
    {
        worker->terminate = 1; // 使得对应线程执行其回调函数就会退出
    }
    //保证任务执行的时候不会被直接打断（任务执行时也是用的这把互斥锁）
    pthread_mutex_lock(&pool->mutex);
    pthread_cond_broadcast(&pool->cond); // 广播唤醒所有等待任务的nWorker
    pthread_mutex_unlock(&pool->mutex);

    pool->workers = NULL;
    pool->tasks = NULL;
    return 0;
}

添加任务功能

添加任务就是将任务添加进任务链表，并且通过pthread_cond_signal唤醒其中一个线程去执行这个任务：

int nThreadPoolPushTask(ThreadPool *pool, struct nTask *task)
{
    //操作到任务队列和执行队列要加锁 （pool->tasks是共享资源）
    pthread_mutex_lock(&pool->mutex);
    //添加任务
    LIST_INSERT(task, pool->tasks);
    
    //通知回调函数有任务去执行任务
    pthread_cond_signal(&pool->cond); //唤醒一个线程中的wait

    pthread_mutex_unlock(&pool->mutex);
}

可能会出的错误

结构体变量没有进行清空，导致有值，解决方法：

ThreadPool     pool = {0}

或者

ThreadPool pool;
memset(&pool, 0x00, sizeof(ThreadPool));

主线程没有等待子线程执行任务结束就退出

解决办法：主线程出口位置添加 getchar()让用户输入一个字符再退出

思考题

了解CAS无锁算法，并写出自己心得体会

完整代码

#include 
#include 
#include 
#include 

//支持层对链表的操作
#define LIST_INSERT(item, list) do{ 
    item->prev = NULL;              
    item->next = list;              
    if((list) != NULL) (list)->prev = item; 
    (list) = item;                  
}while(0)

#define LIST_REMOVE(item, list) do{                         
    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;                         
}while(0)

//任务类型
struct nTask{
    //任务实际工作函数
    void (*task_func)(void *arg); 
    void *user_data; //task_func的参数

    //任务队列用双向列表组织起来
    struct nTask *prev;
    struct nTask *next;
};


//线程（执行）类型
struct nWorker{
    pthread_t threadid;
    struct nManager *manager;
    int terminate; //是否终止的标识.1为终止

    struct nWorker *prev;
    struct nWorker *next;
};

//执行队列和任务队列之间的管理组件
//这就是我们所谓的线程池，就是这个管理组件
typedef struct nManager{
    struct nTask *tasks; //任务队列
    struct nWorker *workers;//执行队列

    pthread_mutex_t mutex;
    pthread_cond_t cond;
}ThreadPool;


//线程回调函数，static表示本文件有效
//回调函数不等于任务！callback ！= task
//每个线程的任务就是循环等待任务到来并执行任务
static void *nThreadPoolCallback(void *arg)
{
    //worker指向某一个线程
    struct nWorker *worker = (struct nWorker *)arg;
    printf("nThreadPoolCallbackn");

    //有任务执行任务，没任务循环等待
    while(1)
    {
        //这里也能看出nWorker中添加了manager成员的用处，就不用再传nManager对象了
        pthread_mutex_lock(&worker->manager->mutex);
        //无任务时，阻塞等待
        while(worker->manager->tasks == NULL)
        {
            //worker->terminate表明所有线程都要销毁了
            if(worker->terminate)   break;  
             //条件等待 
            //会先解除互斥锁让其他线程能够执行，然后阻塞，
            //当被pthread_cond_signal唤醒时，则会解除阻塞并加互斥锁
            pthread_cond_wait(&worker->manager->cond, &worker->manager->mutex);
        } 

        if(worker->terminate)
        {
            //防止死锁
            pthread_mutex_unlock(&worker->manager->mutex);
            break;
        }

        //任务执行，将任务从任务队列中取出
        struct nTask *task = worker->manager->tasks; 
        LIST_REMOVE(task, worker->manager->tasks);
        
        pthread_mutex_unlock(&worker->manager->mutex);
        
        //执行任务
        task->task_func(task);
    }
    free(worker);

}

//对线程池pool进行初始化，numWorkers是线程池中线程数量(即聘请的柜员数量)
int nThreadPoolCreate(ThreadPool *pool, int numWorkers)
{
    if(pool == NULL)    return -1;
    if(numWorkers < 1)  numWorkers = 1;

    //pool必须要初始化为空，如果是垃圾值下面memcpy会出现段错误
    memset(pool, 0x00, sizeof(ThreadPool));

    //相当于pthread_cond_init()进行初始化
    pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
    memcpy(&pool->cond, &blank_cond, sizeof(pthread_cond_t));

    //pthread_mutex_init(&pool->mutex, NULL); // 初始化互斥锁，下方代码功能相同
    pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
    memcpy(&pool->mutex, &blank_mutex, sizeof(pthread_mutex_t));


    int i = 0;
    //创建线程
    for(i = 0;i < numWorkers; ++i)
    {
        struct nWorker *worker = (struct nWorker *)malloc(sizeof(struct nWorker));
        if(worker == NULL)
        {
            perror("malloc");
            return -2;
        }
        memset(worker, 0x00, sizeof(struct nWorker));
        worker->manager = pool;//方便操作

        //创建线程，成功返回0
        int ret = pthread_create(&worker->threadid, NULL, nThreadPoolCallback, worker);
        if(ret)
        {
            perror("pthread_create");
            free(worker);
            return -3;
        }

        printf("LIST_INSERTn");
        //将线程插入线程链表(队列)头
        LIST_INSERT(worker, pool->workers);
    }

    return 0;
}

int nThreadPoolDestroy(ThreadPool *pool, int nWorker)
{
    struct nWorker *worker = NULL;
    for(worker = pool->workers;worker != NULL; worker = worker->next)
    {
        worker->terminate = 1; // 使得对应线程执行其回调函数就会退出
    }
    //保证任务执行的时候不会被直接打断（任务执行时也是用的这把互斥锁）
    pthread_mutex_lock(&pool->mutex);
    pthread_cond_broadcast(&pool->cond); // 广播唤醒所有等待任务的nWorker
    pthread_mutex_unlock(&pool->mutex);

    pool->workers = NULL;
    pool->tasks = NULL;
    return 0;
}

//往任务队列添加任务
int nThreadPoolPushTask(ThreadPool *pool, struct nTask *task)
{
    //操作到任务队列和执行队列要加锁 （pool->tasks是共享资源）
    pthread_mutex_lock(&pool->mutex);
    //添加任务
    LIST_INSERT(task, pool->tasks);
    
    //通知回调函数有任务去执行任务
    pthread_cond_signal(&pool->cond); //唤醒一个线程中的wait

    pthread_mutex_unlock(&pool->mutex);
}

#if 1
//线程池初始化线程数量
#define THREADPOOL_INIT_COUNT   20
#define TASK_INIT_SIZE          1000

void task_entry(void *arg)
{
    struct nTask *task = (struct nTask *)arg;
    int idx = *(int *)task->user_data;

    printf("idx: %dn", idx); 
 
    free(task->user_data);
    free(task);
}

int main(void)
{
    ThreadPool pool = {0};
    //初始化线程池
    nThreadPoolCreate(&pool, THREADPOOL_INIT_COUNT);
    printf("nThreadPoolCreate -- finishn");

    int i = 0;
    //添加任务
    for(i = 0;i < TASK_INIT_SIZE;i++)
    {
        struct nTask *task = (struct nTask *)malloc(sizeof(struct nTask));
        if(task == NULL)
        {
            perror("malloc");
            exit(1);
        }
        memset(task, 0x00, sizeof(struct nTask));

        task->task_func = task_entry;
        task->user_data = malloc(sizeof(int));
        //先转换为int *类型，再取得指向的值
        *(int *)task->user_data = i;

        nThreadPoolPushTask(&pool, task);
    }

    //为了防止主线程在子线程之前退出
    //让用户输入一个字符
    getchar();
}

#endif

C语言简易版线程池

C/C++/C#相关栏目本月热门文章