可扩/减容线程池C语言原理讲解及代码实现

Posted 狱典司

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了可扩/减容线程池C语言原理讲解及代码实现相关的知识，希望对你有一定的参考价值。

可扩/减容线程池C语言原理讲解及代码实现

一、线程池模型原理分析

客户端相当于生产者，服务端相当于消费者。

线程池相比较于传统的多线程模型，每次创建线程和销毁线程的开销是可以被节省的。

故放弃每来一个请求就创建一个线程的机制，采用一次性批量产生线程的方式。

所谓线程“池”，这个池是一个虚拟的概念，指的是产生的线程被保存在的可访问地址。

当没有任务时，线程池里的线程都阻塞（pthread_cond_wait() ）在任务队列不为NULL的条件变量上;

当有客户端发来了任务/请求，server需要去唤醒池中的线程：

两个函数：

  1. pthread_cond_signal()：默认唤醒（阻塞在对应条件变量上的）一个线程。
  2. pthread_cond_broadcast()：会唤醒所有（阻塞在对应条件变量上的）的线程，即产生惊群效应

被唤醒的线程从任务队列中取出任务并完成处理后，重新回到线程池中等待任务，即等待被唤醒。

可以用Unbuntu查看一个线程池的例子：

打开火狐浏览器，并在shell中输入ps aux | grep firefox：

取出进程ID，然后在shell中输入 ps -L [PID]，可以看到虽然我们只开了一个浏览器，但是后台已经一次性启动了许多线程，这里用的就是线程池：

NLWP标识线程数，这里可以看到线程数是38。

1. 1多路IO转接和线程池的比较

  • 多路IO转接处理的是客户端怎么去和服务器建立连接和接受请求的问题；

  • 而线程池处理的是服务器拿到请求之后，要怎么处理的问题。

故线程池可以和多路IO转接配合使用以提高效能。

1.2 线程池线程的创建与维护所需要的参数

在一开始，需要有初步的线程数，即初始值，如：thread_init_num = 38;

但是显然38个线程是不够用的，因为存在访问高峰的情况；当访问高峰来到时，势必要增加线程数，但并不能够无限的增加线程，因为：① 一个进程能够打开的线程数量是有限的 ② 线程虽然和进程共享地址空间，但其栈是要独立的；

故需要一个值来限制最多开启的线程数，如：thread_max_num = 500;

“win系统默认的栈空间是1M(1MiB)大小,而Linux默认下栈空间常见的是8M或10M。

此外，为了判断线程池扩容和瘦身的最佳时机，还需要两个值：

  1. 记录当前的忙线程数的值：thread_busy_num;
  2. 记录当前总共的/存活的线程数的值：thread_live_num;

根据这两个值的比例来判判定是否扩容or瘦身；

扩容时需要一个扩容步长：thread_step，即一次扩容多少线程。

既然存在访问高峰就一定存在访问波谷，若服务器仍维护着高峰时期的线程数，那么就会带来不必要的资源开销（尤其是内存开销），故此时需要瘦身，瘦身步长可以和扩容步长一样，也可以单独设置。

关于线程池扩容or瘦身的行为，一个关键的问题是由谁来计算thread_busy_num和thread_live_num的比值，

让server的主进程来做这件事是不合理的，server主进程的主要任务是监听连接，之所以使用线程池的目的就是不让server主进程分心，将数据处理的工作”外包“给线程池，所以应该额外有一个管理者线程，来维护线程池。

二、线程池描述结构体

typedef struct 
    void *(*function)(void *);          /* 函数指针，回调函数 */
    void *arg;                          /* 上面函数的参数 */
 threadpool_task_t;                    /* 各子线程任务结构体 */


/* 描述线程池相关信息 */

struct threadpool_t 
    pthread_mutex_t lock;               /* 用于锁住本结构体 */    
    pthread_mutex_t thread_counter;     /* 记录忙状态线程个数的锁 -- busy_thr_num */

    pthread_cond_t queue_not_full;      /* 当任务队列满时，添加任务的线程阻塞，等待此条件变量 */
    pthread_cond_t queue_not_empty;     /* 任务队列里不为空时，通知等待任务的线程 */

    pthread_t *threads;                 /* 存放线程池中每个线程的tid，数组结构 */
    pthread_t adjust_tid;               /* 存管理线程tid */
    threadpool_task_t *task_queue;      /* 任务队列(数组首地址) */

    int min_thr_num;                    /* 线程池最小线程数 */
    int max_thr_num;                    /* 线程池最大线程数 */
    int live_thr_num;                   /* 当前存活线程个数 */
    int busy_thr_num;                   /* 忙状态线程个数 */
    int wait_exit_thr_num;              /* 要销毁的线程个数 */

    int queue_front;                    /* task_queue队头下标 */
    int queue_rear;                     /* task_queue队尾下标 */
    int queue_size;                     /* task_queue队中实际任务数 */
    int queue_max_size;                 /* task_queue队列可容纳任务数上限 */

    int shutdown;                       /* 标志位，线程池使用状态，true或false */
;

实际使用的线程池可以根据业务需求来增加或者减少结构体的相关属性。

三、线程池主函数架构

int main(void)

    /*threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size);*/

    threadpool_t *thp = threadpool_create(3,100,100);   /*创建线程池，池里最小3个线程，最大100，队列最大100*/
    													/* 该线程池结构体是所有线程共用的 */
    printf("pool inited");

    //int *num = (int *)malloc(sizeof(int)*20);
    int num[20], i;
    for (i = 0; i < 20; i++) 
        num[i] = i;
        printf("add task %d\\n",i);
        
        /*int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg) */

        threadpool_add(thp, process, (void*)&num[i]);   /* 向线程池中添加任务 */
    

    sleep(10);                                          /* 等子线程完成任务，模拟处理其他任务所占用的cpu时间 */
    threadpool_destroy(thp);							/* 线程池销毁 */

    return 0;

main()函数架构：

  1. 创建线程池
  2. 模拟向线程池添加任务… 借助回调处理任务
  3. 销毁线程池

四、线程池创建

//threadpool_create(3,100,100);  
threadpool_t *threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size)

    int i;
    threadpool_t *pool = NULL;          /* 线程池 结构体 */

    /* 申请线程池结构体空间 */	
    do 
        if((pool = (threadpool_t *)malloc(sizeof(threadpool_t))) == NULL)   
            printf("malloc threadpool fail");
            break;                                      /*跳出do while*/
        

        pool->min_thr_num = min_thr_num;				/* 最小线程数 */
        pool->max_thr_num = max_thr_num;				/* 最大线程数 */
        pool->busy_thr_num = 0;							/* 当前忙线程数 */
        pool->live_thr_num = min_thr_num;               /* 活着的线程数 初值=最小线程数 */
        pool->wait_exit_thr_num = 0;					/* 等待被销毁的线程数 */
        pool->queue_size = 0;                           /* 有0个产品，队列的实际任务数 */
        pool->queue_max_size = queue_max_size;          /* 最大任务队列数 */
        pool->queue_front = 0;							/* 任务队列队头指针 */
        pool->queue_rear = 0;							/* 任务多列队尾指针 */
        pool->shutdown = false;                         /* 不关闭线程池 */

        /* 根据最大线程上限数， 给工作线程数组开辟空间, 并清零 */
        pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num); 
        if (pool->threads == NULL) 
            printf("malloc threads fail");
            break;
        
        memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num);

        /* 给任务队列开辟空间 */
        pool->task_queue = (threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size);
        if (pool->task_queue == NULL) 
            printf("malloc task_queue fail");
            break;
        

        /* 初始化互斥琐、条件变量（锁的默认属性传NULL） */
        if (pthread_mutex_init(&(pool->lock), NULL) != 0
                || pthread_mutex_init(&(pool->thread_counter), NULL) != 0
                || pthread_cond_init(&(pool->queue_not_empty), NULL) != 0
                || pthread_cond_init(&(pool->queue_not_full), NULL) != 0)
        
            printf("init the lock or cond fail");
            break;
        

        /* 启动 min_thr_num 个 work thread */
        for (i = 0; i < min_thr_num; i++) 
            /* threadpool_thread是线程的回调函数地址 */
            pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);   /*pool指向当前线程池,作为线程回调函数的参数 void* args*/
            /* 即每个线程启动时会将线程池属性作为参数传递给线程的回调函数，注意！传递的是地址！ */
            printf("start thread 0x%x...\\n", (unsigned int)pool->threads[i]);
        
        
        /*  创建1个管理者线程，adjust_thread是管理者线程的回调函数 。 */
        pthread_create(&(pool->adjust_tid), NULL, adjust_thread, (void *)pool);     /* 创建管理者线程 */

        return pool;

     while (0);

    threadpool_free(pool);      /* 前面代码调用失败时，释放poll存储空间 */

    return NULL;

pthreadpool_create()函数架构:

  1. 创建线程池结构体指针
  2. 初始化线程池结构体（N个成员变量）
  3. 创建N个任务线程
  4. 创建1个管理者线程
  5. 失败时，销毁开辟的所有空间

五、向线程池中添加任务

先简要介绍一下条件变量和互斥锁的配合使用方式：

pthread_cond_XXX()函数与互斥锁mutex的配合：

  1. pthread_cond_wait()
    用于阻塞当前线程，等待别的线程使用pthread_cond_signal()或pthread_cond_broadcast来唤醒它 pthread_cond_wait() 必须与pthread_mutex配套使用。pthread_cond_wait()函数一进入wait状态就会自动release mutex（解锁）。当其他线程通过pthread_cond_signal()或pthread_cond_broadcast，把该线程唤醒，使pthread_cond_wait()通过（返回）时，该线程又自动获得该mutex（上锁）。
  2. pthread_cond_signal函数的作用是发送一个信号给另外一个正在处于阻塞等待状态的线程,使其脱离阻塞状态,继续执行.如果没有线程处在阻塞等待状态,pthread_cond_signal也会成功返回。
    使用pthread_cond_signal一般不会有“惊群现象”产生，他最多只给一个线程发信号。假如有多个线程正在阻塞等待着这个条件变量的话，那么是根据各等待线程优先级的高低确定哪个线程接收到信号开始继续执行。如果各线程优先级相同，则根据等待时间的长短来确定哪个线程获得信号。但无论如何一个pthread_cond_signal调用最多发信一次。
  3. 但是pthread_cond_signal在多处理器上可能同时唤醒多个线程，当你只能让一个线程处理某个任务时，其它被唤醒的线程就需要继续
    wait，而且规范要求pthread_cond_signal至少唤醒一个pthread_cond_wait上的线程，其实有些实现为了简单在单处理器上也会唤醒多个线程。另外，某些应用，如线程池，pthread_cond_broadcast唤醒全部线程，但我们通常只需要一部分线程去做执行任务，所以其它的线程需要继续wait。所以强烈推荐对pthread_cond_wait()使用while循环来做条件判断（下面的任务线程回调函数中就有体现）。

向线程池中添加任务

/* 线程池中的线程，模拟处理业务 */
void *process(void *arg)

    printf("thread 0x%x working on task %d\\n ",(unsigned int)pthread_self(),(int)arg);
    sleep(1);                           //模拟 工作
    printf("task %d is end\\n",(int)arg);

    return NULL;


/* 向线程池中 添加一个任务 */
//调用： threadpool_add(thp, process, (void*)&num[i]);   /* 向线程池中添加任务 process: 小写---->大写*/
/* 在这个函数里面把任务写道任务队列中 */ 
int threadpool_add(threadpool_t *pool, void*(*function)(void *arg), void *arg)
 
	/*先对线程池结构体本身上锁*/ 
    pthread_mutex_lock(&(pool->lock));

    /* ==为真，队列已经满， 调wait阻塞 */
    while ((pool->queue_size == pool->queue_max_size) && (!pool->shutdown)) 
        pthread_cond_wait(&(pool->queue_not_full), &(pool->lock));
    

    if (pool->shutdown) 
        pthread_cond_broadcast(&(pool->queue_not_empty));
        pthread_mutex_unlock(&(pool->lock));
        return 0;
    

    /* 清空 工作线程 调用的回调函数 的参数arg -- 在任务队列的队尾 */
    if (pool->task_queue[pool->queue_rear].arg != NULL) 
        pool->task_queue[pool->queue_rear].arg = NULL;
    

    /*添加任务到任务队列里*/
    pool->task_queue[pool->queue_rear].function = function;
    pool->task_queue[pool->queue_rear].arg = arg;
    pool->queue_rear = (pool->queue_rear + 1) % pool->queue_max_size;       /* 队尾指针移动, 模拟环形 */
    pool->queue_size++;

    /*添加完任务后，队列不为空，唤醒线程池中 等待处理任务的线程*/
    pthread_cond_signal(&(pool->queue_not_empty));
    pthread_mutex_unlock(&(pool->lock));

    return 0;

具体的业务可以在void *process(void *arg)函数中自定义。

六、子线程回调函数

6.1 管理线程回调函数

/* 管理线程 */
void *adjust_thread(void *threadpool)

    int i;
    threadpool_t *pool = (threadpool_t *)threadpool;
    while (!pool->shutdown) 

        sleep(DEFAULT_TIME);                                    /* 对扩容和减容的操作一般没有那么紧迫，不需要让管理线程一直参与cpu的争夺，定时苏醒对线程池进行管理即可 */

        pthread_mutex_lock(&(pool->lock));						/* 锁住线程池 */
        int queue_size = pool->queue_size;                      /* 任务数 */
        int live_thr_num = pool->live_thr_num;                  /* 存活线程数 */
        pthread_mutex_unlock(&(pool->lock));					/* 解锁线程池 */

        pthread_mutex_lock(&(pool->thread_counter));			/* 锁住忙线程数 */
        int busy_thr_num = pool->busy_thr_num;                  /* 忙着的线程数 */
        pthread_mutex_unlock(&(pool->thread_counter));			/* 解锁忙线程数 */

        /* 1. 创建新线程 算法： 任务数大于最小线程池个数, 且存活的线程数少于最大线程个数时 如：30>=10 && 40<100*/
        if (queue_size >= MIN_WAIT_TASK_NUM && live_thr_num < pool->max_thr_num) 
            pthread_mutex_lock(&(pool->lock));  
            int add = 0;

            /*一次增加 DEFAULT_THREAD 个线程*/
            for (i = 0; i < pool->max_thr_num && add < DEFAULT_THREAD_VARY
                    && pool->live_thr_num < pool->max_thr_num; i++) 
                if (pool->threads[i] == 0 || !is_thread_alive(pool->threads[i])) 
                    pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
                    add++;
                    pool->live_thr_num++;
                
            

            pthread_mutex_unlock(&(pool->lock));
        

        /* 2.  销毁多余的空闲线程 算法：忙线程X2 小于 存活的线程数 且 存活的线程数 大于 最小线程数时*/
        if ((busy_thr_num * 2) < live_thr_num  &&  live_thr_num > pool->min_thr_num) 

            /* 一次销毁DEFAULT_THREAD个线程, 隨機10個即可 */
            pthread_mutex_lock(&(pool->lock));
            pool->wait_exit_thr_num = DEFAULT_THREAD_VARY;      /* 要销毁的线程数 设置为10 */
            pthread_mutex_unlock(&(pool->lock));

            for (i = 0; i < DEFAULT_THREAD_VARY; i++) 
                /* 通知处在空闲状态的线程, 他们会自行终止*/
                /* 所谓通知处在空闲状态的线程，就是唤醒阻塞在queue_not_empty条件变量上的线程 */ 
                /* pthread_cond_signal()的功能是一次唤醒一个进程 */
                pthread_cond_signal(&(pool->queue_not_empty));
            
        
    
    return NULL;

adjust_thread()函数架构：

  1. 接收回调参数 void *arg --> pool结构体

  2. 循环 DEFAULT_TIME 苏醒一次

    对扩容和减容的操作一般没有那么紧迫，不需要让管理线程一直参与cpu的争夺，定时苏醒对线程池进行管理即可。

  3. 配合互斥锁的使用，获取【任务数】、【存活线程数】和 【忙线程数】

  4. 根据既定算法，使用上述三个变量，计算并判断是否进行扩容或减容操作（即是否应该创建or销毁现称此中指定步长的线程）。

6.2 任务线程回调函数

/* 线程池中各个工作线程 */
void *threadpool_thread(void *threadpool)/* 参数是指向线程池结构体的指针  */ 

    threadpool_t *pool = (threadpool_t *)threadpool; /* 还原出线程池结构体pool */ 
    threadpool_task_t task;							 /* 定义一个任务结构体 */

    while (true) 
        /* Lock must be taken to wait on conditional variable */
        /*刚创建出线程，等待任务队列里有任务，否则阻塞等待任务队列里有任务后再唤醒接收任务*/
        
        
        /* 在线程访问公用的线程池结构体pool 之前先上互斥锁*/
        pthread_mutex_lock(&(pool->lock));

        /*queue_size == 0 说明没有任务，调 wait 阻塞在条件变量上, 若有任务，跳过该while*/
        while ((pool->queue_size == 0) && (!pool->shutdown))   
            printf("thread 0x%x is waiting\\n", (unsigned int)pthread_self());
            
            /* 阻塞等待条件变量queue_not_empty 并解开 lock互斥锁，直到条件变量满足则上lock锁并执行  */ 
            pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));

            /*清除指定数目的空闲线程，如果要结束的线程个数大于0，结束线程*/
            /* wait_exit_thr_num 表示要销毁的线程 */ 
            if (pool->wait_exit_thr_num > 0) 
                pool->wait_exit_thr_num--;

                /*如果线程池里线程个数大于最小值时可以结束当前线程*/
                /* 注意该if是在上一个if里的 */
                /* live_thr_num表示当前存活的线程个数 */
                if (pool->live_thr_num > pool->min_thr_num) 
                    printf("thread 0x%x is exiting\\n", (unsigned int)pthread_self());
                    pool->live_thr_num--;
                    
                    
                    /* 在解锁之后退出 */ 
                    /* 使用pthread_mutex的颗粒度尽量小；
					* 特别注意while，如果解锁写在while的最尾，上锁写在while的最前，
					* 那么其他线程几乎不可能抢到这把锁 */ 
                    pthread_mutex_unlock(&(pool->lock));

					/* 处在threadpool_thread回调函数中，若这里exit之后的代码就不会执行了 */
                    pthread_exit(NULL);
                
            
        

        /*如果指定了true，要关闭线程池里的每个线程，自行退出处理---销毁线程池*/
        if (pool->shutdown) 
            pthread_mutex_unlock(&(pool->lock));
            printf("thread 0x%x is exiting\\n", (unsigned int)pthread_self());
            pthread_detach(pthread_self());  /* 将自己分离，资源被自动回收，主要是回收资源 */ 
            pthread_exit(NULL);     /* detach之后线程并没有死，还需要pthread_exit()使得线程自行结束 */
        

        /*从任务队列里获取任务, 是一个出队操作*/
        task.function = pool->task_queue[pool->queue_front].function; //取回调函数 
        task.arg = pool->task_queue以上是关于可扩/减容线程池C语言原理讲解及代码实现的主要内容，如果未能解决你的问题，请参考以下文章
 

 用Unix / C实现基于自动扩/减容线程池+epoll反应堆检测沉寂用户模型的服务器框架（含源码）


 用Linux / C实现基于自动扩/减容线程池+epoll反应堆检测沉寂用户模型的服务器框架（含源码）


 一个Linux下C线程池的实现


 基于C++11实现的高效线程池及工作原理


 基于C++11实现的高效线程池及工作原理


 c++11线程池的实现原理及回调函数的使用