基于线程池模型的讨论与完整代码演示

线程池引入的必要性：

在网络服务器中，包括大量的web服务器，它们都需要在单位时间内必须处理相当数目的接入请求以及数据处理。通常在传统多线程服务器中是这样实现的：一旦有个请求到达，就创建一个线程，由该线程执行任务，任务执行完毕后，线程就退出。这也就是通常所说的及时创建，及时销毁策略。在现代计算机中，尽管创建线程的时间已经大大缩短，但是如果提交给线程的任务是执行时间较短，而且次数非常的频繁，那么服务器就将处于一个不停创建于销毁线程的状态下，这将是一笔不小的开销。尤其是在线程执行的时间非常短的情况。

线程池就是为了解决上述问题的。

线程池概念

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

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

本程序由三个文件组成，分别是threadpoll.h,threadpoll.c,main.c,下面的代码已经在64位系统成功运行。

验证环境：

Distributor ID: Ubuntu

Description: Ubuntu 13.04

Release: 13.04

Codename: raring

线程池代码示例

threadpoll.h文件下的代码

#ifndef _THREADPOLL_H_
#define _THREADPOLL_H_
#include<pthread.h>
#include<unistd.h>
#include<stdlib.h>
#include<assert.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_t
pthread_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);
#endif

threadpoll.c

#include "threadpoll.h"
#include<stdio.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<stdio.h>
#include"threadpoll.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;
}

gcc main.c threadpoll.c -lpthread -othreadpoll

./threadpoll

ky@ky-S910-X31E:~/libz/623$ ./threadpoll
threadpool callback fuction : 1.
threadpool callback fuction : 2.
threadpool callback fuction : 8.
threadpool callback fuction : 5.
threadpool callback fuction : 3.
threadpool callback fuction : 13.
threadpool callback fuction : 6.
threadpool callback fuction : 17.
threadpool callback fuction : 18.
threadpool callback fuction : 19.
threadpool callback fuction : 20.
threadpool callback fuction : 21.
threadpool callback fuction : 22.
threadpool callback fuction : 23.
threadpool callback fuction : 24.
threadpool callback fuction : 25.
threadpool callback fuction : 26.
threadpool callback fuction : 27.
threadpool callback fuction : 28.
threadpool callback fuction : 29.
threadpool callback fuction : 30.
threadpool callback fuction : 31.
threadpool callback fuction : 32.
threadpool callback fuction : 33.
threadpool callback fuction : 34.
threadpool callback fuction : 35.
threadpool callback fuction : 36.
threadpool callback fuction : 37.
threadpool callback fuction : 38.
threadpool callback fuction : 39.
threadpool callback fuction : 40.
threadpool callback fuction : 9.
threadpool callback fuction : 11.
threadpool callback fuction : 12.
threadpool callback fuction : 14.
threadpool callback fuction : 7.
threadpool callback fuction : 15.
threadpool callback fuction : 16.
threadpool callback fuction : 10.
threadpool callback fuction : 4.

更多内容请前往：

http://wenku.baidu.com/p/helpylee