简单Linux C线程池

http://www.cnblogs.com/venow/archive/2012/11/22/2779667.html

　大多数的网络服务器，包括Web服务器都具有一个特点，就是单位时间内必须处理数目巨大的连接请求，但是处理时间却是比较短的。在传统的多线程服务器模型中是这样实现的：一旦有个请求到达，就创建一个新的线程，由该线程执行任务，任务执行完毕之后，线程就退出。这就是"即时创建，即时销毁"的策略。尽管与创建进程相比，创建线程的时间已经大大的缩短，但是如果提交给线程的任务是执行时间较短，而且执行次数非常频繁，那么服务器就将处于一个不停的创建线程和销毁线程的状态。这笔开销是不可忽略的，尤其是线程执行的时间非常非常短的情况。

　　线程池就是为了解决上述问题的，它的实现原理是这样的：在应用程序启动之后，就马上创建一定数量的线程，放入空闲的队列中。这些线程都是处于阻塞状态，这些线程只占一点内存，不占用CPU。当任务到来后，线程池将选择一个空闲的线程，将任务传入此线程中运行。当所有的线程都处在处理任务的时候，线程池将自动创建一定的数量的新线程，用于处理更多的任务。执行任务完成之后线程并不退出，而是继续在线程池中等待下一次任务。当大部分线程处于阻塞状态时，线程池将自动销毁一部分的线程，回收系统资源。

　　下面是一个简单线程池的实现，这个线程池的代码是我参考网上的一个例子实现的，由于找不到出处了，就没办法注明参考自哪里了。它的方案是这样的：程序启动之前，初始化线程池，启动线程池中的线程，由于还没有任务到来，线程池中的所有线程都处在阻塞状态，当一有任务到达就从线程池中取出一个空闲线程处理，如果所有的线程都处于工作状态，就添加到队列，进行排队。如果队列中的任务个数大于队列的所能容纳的最大数量，那就不能添加任务到队列中，只能等待队列不满才能添加任务到队列中。

　　主要由两个文件组成一个threadpool.h头文件和一个threadpool.c源文件组成。源码中已有重要的注释，就不加以分析了。

　　threadpool.h文件：

struct job
{void* (*callback_function)(void *arg);    //线程回调函数void *arg;                                //回调函数参数struct job *next;
};struct threadpool
{int thread_num;                   //线程池中开启线程的个数int queue_max_num;                //队列中最大job的个数struct job *head;                 //指向job的头指针struct job *tail;                 //指向job的尾指针pthread_t *pthreads;              //线程池中所有线程的pthread_tpthread_mutex_t mutex;            //互斥信号量pthread_cond_t queue_empty;       //队列为空的条件变量pthread_cond_t queue_not_empty;   //队列不为空的条件变量pthread_cond_t queue_not_full;    //队列不为满的条件变量int queue_cur_num;                //队列当前的job个数int queue_close;                  //队列是否已经关闭int pool_close;                   //线程池是否已经关闭
};//================================================================================================
//函数名：                   threadpool_init
//函数描述：                 初始化线程池
//输入：                    [in] thread_num     线程池开启的线程个数
//                         [in] queue_max_num  队列的最大job个数 
//输出：                    无
//返回：                    成功：线程池地址 失败：NULL
//================================================================================================
struct threadpool* threadpool_init(int thread_num, int queue_max_num);//================================================================================================
//函数名：                    threadpool_add_job
//函数描述：                  向线程池中添加任务
//输入：                     [in] pool                  线程池地址
//                          [in] callback_function     回调函数
//                          [in] arg                     回调函数参数
//输出：                     无
//返回：                     成功：0 失败：-1
//================================================================================================
int threadpool_add_job(struct threadpool *pool, void* (*callback_function)(void *arg), void *arg);//================================================================================================
//函数名：                    threadpool_destroy
//函数描述：                   销毁线程池
//输入：                      [in] pool                  线程池地址
//输出：                      无
//返回：                      成功：0 失败：-1
//================================================================================================
int threadpool_destroy(struct threadpool *pool);//================================================================================================
//函数名：                    threadpool_function
//函数描述：                  线程池中线程函数
//输入：                     [in] arg                  线程池地址
//输出：                     无  
//返回：                     无
//================================================================================================
void* threadpool_function(void* arg);

　　threadpool.c文件：

#include "threadpool.h"struct threadpool* threadpool_init(int thread_num, int queue_max_num)
{struct threadpool *pool = NULL;do {pool = malloc(sizeof(struct threadpool));if (NULL == pool){printf("failed to malloc threadpool!\n");break;}pool->thread_num = thread_num;pool->queue_max_num = queue_max_num;pool->queue_cur_num = 0;pool->head = NULL;pool->tail = NULL;if (pthread_mutex_init(&(pool->mutex), NULL)){printf("failed to init mutex!\n");break;}if (pthread_cond_init(&(pool->queue_empty), NULL)){printf("failed to init queue_empty!\n");break;}if (pthread_cond_init(&(pool->queue_not_empty), NULL)){printf("failed to init queue_not_empty!\n");break;}if (pthread_cond_init(&(pool->queue_not_full), NULL)){printf("failed to init queue_not_full!\n");break;}pool->pthreads = malloc(sizeof(pthread_t) * thread_num);if (NULL == pool->pthreads){printf("failed to malloc pthreads!\n");break;}pool->queue_close = 0;pool->pool_close = 0;int i;for (i = 0; i < pool->thread_num; ++i){pthread_create(&(pool->pthreads[i]), NULL, threadpool_function, (void *)pool);}return pool;    } while (0);return NULL;
}int threadpool_add_job(struct threadpool* pool, void* (*callback_function)(void *arg), void *arg)
{assert(pool != NULL);assert(callback_function != NULL);assert(arg != NULL);pthread_mutex_lock(&(pool->mutex));while ((pool->queue_cur_num == pool->queue_max_num) && !(pool->queue_close || pool->pool_close)){pthread_cond_wait(&(pool->queue_not_full), &(pool->mutex));   //队列满的时候就等待
    }if (pool->queue_close || pool->pool_close)    //队列关闭或者线程池关闭就退出
    {pthread_mutex_unlock(&(pool->mutex));return -1;}struct job *pjob =(struct job*) malloc(sizeof(struct job));if (NULL == pjob){pthread_mutex_unlock(&(pool->mutex));return -1;} pjob->callback_function = callback_function;    pjob->arg = arg;pjob->next = NULL;if (pool->head == NULL)   {pool->head = pool->tail = pjob;pthread_cond_broadcast(&(pool->queue_not_empty));  //队列空的时候，有任务来时就通知线程池中的线程：队列非空
    }else{pool->tail->next = pjob;pool->tail = pjob;    }pool->queue_cur_num++;pthread_mutex_unlock(&(pool->mutex));return 0;
}void* threadpool_function(void* arg)
{struct threadpool *pool = (struct threadpool*)arg;struct job *pjob = NULL;while (1)  //死循环
    {pthread_mutex_lock(&(pool->mutex));while ((pool->queue_cur_num == 0) && !pool->pool_close)   //队列为空时，就等待队列非空
        {pthread_cond_wait(&(pool->queue_not_empty), &(pool->mutex));}if (pool->pool_close)   //线程池关闭，线程就退出
        {pthread_mutex_unlock(&(pool->mutex));pthread_exit(NULL);}pool->queue_cur_num--;pjob = pool->head;if (pool->queue_cur_num == 0){pool->head = pool->tail = NULL;}else {pool->head = pjob->next;}if (pool->queue_cur_num == 0){pthread_cond_signal(&(pool->queue_empty));        //队列为空，就可以通知threadpool_destroy函数，销毁线程函数
        }if (pool->queue_cur_num == pool->queue_max_num - 1){pthread_cond_broadcast(&(pool->queue_not_full));  //队列非满，就可以通知threadpool_add_job函数，添加新任务
        }pthread_mutex_unlock(&(pool->mutex));(*(pjob->callback_function))(pjob->arg);   //线程真正要做的工作，回调函数的调用
        free(pjob);pjob = NULL;    }
}
int threadpool_destroy(struct threadpool *pool)
{assert(pool != NULL);pthread_mutex_lock(&(pool->mutex));if (pool->queue_close || pool->pool_close)   //线程池已经退出了，就直接返回
    {pthread_mutex_unlock(&(pool->mutex));return -1;}pool->queue_close = 1;        //置队列关闭标志while (pool->queue_cur_num != 0){pthread_cond_wait(&(pool->queue_empty), &(pool->mutex));  //等待队列为空
    }    pool->pool_close = 1;      //置线程池关闭标志pthread_mutex_unlock(&(pool->mutex));pthread_cond_broadcast(&(pool->queue_not_empty));  //唤醒线程池中正在阻塞的线程pthread_cond_broadcast(&(pool->queue_not_full));   //唤醒添加任务的threadpool_add_job函数int i;for (i = 0; i < pool->thread_num; ++i){pthread_join(pool->pthreads[i], NULL);    //等待线程池的所有线程执行完毕
    }pthread_mutex_destroy(&(pool->mutex));          //清理资源pthread_cond_destroy(&(pool->queue_empty));pthread_cond_destroy(&(pool->queue_not_empty));   pthread_cond_destroy(&(pool->queue_not_full));    free(pool->pthreads);struct job *p;while (pool->head != NULL){p = pool->head;pool->head = p->next;free(p);}free(pool);return 0;
}

　　测试文件main.c文件：

#include "threadpool.h"void* work(void* arg)
{char *p = (char*) arg;printf("threadpool callback fuction : %s.\n", p);sleep(1);
}int main(void)
{struct threadpool *pool = threadpool_init(10, 20);threadpool_add_job(pool, work, "1");threadpool_add_job(pool, work, "2");threadpool_add_job(pool, work, "3");threadpool_add_job(pool, work, "4");threadpool_add_job(pool, work, "5");threadpool_add_job(pool, work, "6");threadpool_add_job(pool, work, "7");threadpool_add_job(pool, work, "8");threadpool_add_job(pool, work, "9");threadpool_add_job(pool, work, "10");threadpool_add_job(pool, work, "11");threadpool_add_job(pool, work, "12");threadpool_add_job(pool, work, "13");threadpool_add_job(pool, work, "14");threadpool_add_job(pool, work, "15");threadpool_add_job(pool, work, "16");threadpool_add_job(pool, work, "17");threadpool_add_job(pool, work, "18");threadpool_add_job(pool, work, "19");threadpool_add_job(pool, work, "20");threadpool_add_job(pool, work, "21");threadpool_add_job(pool, work, "22");threadpool_add_job(pool, work, "23");threadpool_add_job(pool, work, "24");threadpool_add_job(pool, work, "25");threadpool_add_job(pool, work, "26");threadpool_add_job(pool, work, "27");threadpool_add_job(pool, work, "28");threadpool_add_job(pool, work, "29");threadpool_add_job(pool, work, "30");threadpool_add_job(pool, work, "31");threadpool_add_job(pool, work, "32");threadpool_add_job(pool, work, "33");threadpool_add_job(pool, work, "34");threadpool_add_job(pool, work, "35");threadpool_add_job(pool, work, "36");threadpool_add_job(pool, work, "37");threadpool_add_job(pool, work, "38");threadpool_add_job(pool, work, "39");threadpool_add_job(pool, work, "40");sleep(5);threadpool_destroy(pool);return 0;
}