C语言实现的线程池

系统管理员 2022-08-19 08:37 298阅读 0赞

**概念：**

如果一个程序每处理一个任务都需要创建一个线程来处理，假设创建线程的时间为T1，任务执行的时间为T2，线程销毁的时间为T3，那么线程的有效使用时间率为T2/(T1+T2+T3)，如果任务执行的时间非常短，那么线程的使用效率就会非常低，这对高并发的服务器性能来说是不能接受的，所以需要引入线程池概念。线程池简单点来说就是创建了很多工作线程，线程函数for(;;)循环，不停的从任务队伍获取任务并处理。

**注意点：**

1.线程同步问题，互斥锁，条件变量，信号量都可以用于线程同步，单纯的互斥锁只能轮询去处理数据，可能会空转进而造成CPU资源的浪费，信号量并不建议用([并发编程的十五条建议][Link 1])，所以使用**互斥锁+条件变量**来解决线程同步的问题，可参考[条件变量实现生产者消费者模型][Link 2]

**线程池的组成：**

1.任务队列：简单来说就是数据的容器，用来存放没有处理的任务，可以用队列，也可以用链表等数据结构

2.任务接口：具体点来说就是任务队列里面的数据要调用哪个函数来处理

3..线程管理：负责创建工作线程，销毁工作线程

**----- gcc pthread\_pool.c -lpthread -----**

#include <stdio.h>
    #include <unistd.h>
    #include <pthread.h>
    #include <stdlib.h>
    
    /* 任务,任务回调函数,任务队列 */
    typedef struct worker {  
        void* (*callback) (void* arg); /*任务回调函数*/
        void* arg;                     /*回调函数的参数*/  
        struct worker* next;           /*任务队列链表*/  
    } CThread_worker;
    
    /*回调函数*/
    static void* callback(void* arg) {
        printf("threadid:0x%x, working on task %d\n", pthread_self(), *(int*)arg);  
        sleep(1);
        return(NULL);
    }
    
    /*线程池结构*/  
    typedef struct {  
        pthread_mutex_t mutex;      /*互斥锁 */
        pthread_cond_t cond;        /*条件变量 */
      
        CThread_worker* queue_head; /*线程池的任务队列*/
      
        int shutdown;               /*是否摧毁线程池 0:不摧毁 1:摧毁 */
        pthread_t* threadid;        /*线程ID数组*/
        int max_thread_num;         /*线程池最大线程数*/  
        int cur_queue_size;         /*任务队列在任务数目*/
      
    } CThread_pool;
    
    /*线程函数*/
    void* thread_routine(void* arg);
    
    /*线程池实例*/
    static CThread_pool* pool = NULL; 
    
    /*线程池初始化*/
    void pool_init(int max_thread_num) {
    	/*一些列初始化*/
    	pool = (CThread_pool*) malloc(sizeof(CThread_pool));
    
    	pthread_mutex_init(&(pool->mutex), NULL);
        pthread_cond_init(&(pool->cond), NULL);
    
        pool->queue_head = NULL;
    
        pool->max_thread_num = max_thread_num;
    
        pool->shutdown = 0; /*0打开1关闭*/
    
        pool->cur_queue_size = 0;
    
    	pool->threadid = (pthread_t*) malloc(max_thread_num * sizeof (pthread_t));
    
    	/*创建工作线程*/
        int i = 0;  
        for (i=0; i<max_thread_num; ++i) {
            pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);
        }  
    
    }
    
    /*将任务加入队列*/
    int pool_add_worker(void* (*callback) (void* arg), void* arg) {
        /*构造一个新任务*/
        printf("pool add worker arg:%d\n", *(int*)arg);
        CThread_worker* newworker = (CThread_worker*) malloc(sizeof(CThread_worker));
        newworker->callback = callback;  
        newworker->arg = arg;  
        newworker->next = NULL; /*SET NULL*/
    
        pthread_mutex_lock(&(pool->mutex));
    
        /*将任务加入到任务队列中,也就是链表末端*/
        CThread_worker* worker = pool->queue_head;
        if (worker != NULL) {  
            while (worker->next != NULL)
                worker = worker->next;
            worker->next = newworker;
        }  
        else { 
            pool->queue_head = newworker;
        }
      	
      	/*是否需要唤醒线程*/
        int dosignal;
        if (pool->cur_queue_size == 0)
         	dosignal = 1;
    
        pool->cur_queue_size += 1; /*计数+1*/
    
        pthread_mutex_unlock(&(pool->mutex));
    
        /*需要叫醒工作线程*/
        if (dosignal)
        	pthread_cond_signal(&(pool->cond));
    
        return 0;
    }
    
    /*销毁线程池*/
    int pool_destroy() {
    	printf("pool destroy now\n");
    
    	/*启用关闭开关*/
        if (pool->shutdown)  
            return -1; /*防止两次调用*/
        pool->shutdown = 1;  
      
        /*唤醒所有等待线程*/  
        pthread_cond_broadcast(&(pool->cond));
      
        /*阻塞等待线程退出回收资源，还有另一种办法就是线程分离*/
        int i;  
        for (i=0; i<pool->max_thread_num; ++i)  
            pthread_join(pool->threadid[i], NULL);
        free(pool->threadid);
        pool->threadid = NULL;
      
        /*销毁任务队列*/  
        CThread_worker* head = NULL;  
        while (pool->queue_head != NULL) {  
            head = pool->queue_head;
            pool->queue_head = pool->queue_head->next;
            free(head);
            head = NULL;
        }
    
        /*销毁互斥锁与条件变量*/
        pthread_mutex_destroy(&(pool->mutex));
        pthread_cond_destroy(&(pool->cond));
          
        free(pool); 
        pool = NULL;
        printf("pool destroy end\n");
        return 0;  
    }  
    
    /*工作线程函数*/
    void* thread_routine(void* arg) {
        printf("starting threadid:0x%x\n", pthread_self());  
        
        for (; ;) {
    
            pthread_mutex_lock(&(pool->mutex));  
            /*任务队列为空时wait唤醒,当销毁线程池时跳出循环*/ 
            while (pool->cur_queue_size == 0 && !pool->shutdown) {
                printf("threadid:0x%x is waiting\n", pthread_self());  
                pthread_cond_wait(&(pool->cond), &(pool->mutex));  
            }  
      
            /*线程池要销毁了*/  
            if (pool->shutdown) {
                pthread_mutex_unlock(&(pool->mutex));  
                printf("threadid:0x%x will exit\n", pthread_self());  
                pthread_exit(NULL);  
            }
      		
      		/*开始执行任务*/
            printf("threadid:0x%x is starting to work\n", pthread_self());
              
            /*等待队列长度减去1，并取出链表中的头元素*/  
            pool->cur_queue_size -= 1;  
            CThread_worker* worker = pool->queue_head;  
            pool->queue_head = worker->next;  
            pthread_mutex_unlock(&(pool->mutex));  
      
            /*调用回调函数，执行任务*/  
            (*(worker->callback)) (worker->arg);  
            free(worker);  
            worker = NULL;  
        }
        return(NULL);
    }
    
    
    /*测试*/
    int main(int argc, char const *argv[])
    {	
        pool_init(2); /*创建n个线程*/  
          
        /*添加n个任务*/
        int* workingnum = (int*) malloc(sizeof(int) * 10); /* 一定要动态创建 */
        int i;  
        for (i=0; i<5; ++i) {
        	workingnum[i] = i;
            pool_add_worker(callback, &workingnum[i]);
        }  
        
        sleep(5); /*等待所有任务完成*/ 
         
        pool_destroy(); /*销毁线程池*/ 
        free(workingnum);
        workingnum = NULL;
        return 0; 
    }

**测试结果：**

![20160108015045658][]

[Link 1]: http://blog.csdn.net/solstice/article/details/5915355
[Link 2]: http://blog.csdn.net/daiyudong2020/article/details/50465664
[20160108015045658]: /images/20220731/33e6f7cef2a24f0cad81ededea8e10b8.png