用条件变量和互斥锁去管理线程池

   今天终于克服众多Bug 搞出了一个简单的线程池的应用（思想来自一篇外文Blog），在初始化线程池之后，可以实现向其中投放任务，多个线程完成大于线程数目的任务量，当然每个任务只能一次被一个线程执行。当一个线程完成一个任务后回去检测当前的剩余任务，从而继续执行剩余任务。大体的内容实现就是这样，下面仔细说一下整个流程问题。

  首先谈到线程池，顾名思义我们需要预先建立多个线程，但是怎么去管理这些线程才是重点。下面的结构可以作为线程池的一个管理结构：

typedef struct {
	pthread_cond_t  ready_event;
	pthread_mutex_t task_queue_lock;
	Work_task *task_head;
	pthread_t *threadid_link;
	int max_thread_num;
	int current_queue_size;
	int destory_flag;
}Thread_pool;

   其中的条件变量和互斥锁是为了后来多个线程访问任务所需要用到的控制条件。threadid_link 是一个pthread_t 类型的链，用来管理线程ID，max_thread_num是当前线程池可以产生的最大线程数量，current_queue_size 是当前线程池中还未处理的任务数量。destory_flag 是一个判断标记，检查是否需要销毁线程池。其中的task_head 是一个Work_task类型的指针，用来管理对应投放到当前线程池的任务链，可以用先面的结构表示一个任务：

typedef struct work_queue{
	void *(*task_process)(void *arg);
	void *arg;
	struct work_queue *next;
}Work_task;

Task_process 是一个参数为arg的函数指针，next为链表的next域指针。
这样我们表示了一个线程池的管理结构，先面我们就需要去初始化这个线程池，开始我们定义一个全局的线程池结构的变量

Thread_pool *thread_pool=NULL;

   初始化我们的线程池结构其实是对thread_pool的数据进行初始化，设定条件变量，互斥锁之类的以及最重要的建立指定数量max_thread_num个线程：

void Thread_pool_init(int max_thread_num){
	int num;
	thread_pool=(Thread_pool *)malloc(sizeof(Thread_pool));
	pthread_mutex_init(&(thread_pool->task_queue_lock),NULL);
	pthread_cond_init(&(thread_pool->ready_event),NULL);
	thread_pool->task_head=NULL;
	thread_pool->destory_flag=0;
	thread_pool->max_thread_num=max_thread_num;
	thread_pool->current_queue_size=0;
	thread_pool->threadid_link=(pthread_t *)malloc(sizeof(pthread_t)*max_thread_num);
	for(num=0;num<max_thread_num;num++){
		pthread_create(&(thread_pool->threadid_link[num]),NULL,thread_routine,NULL);
	}
}

   上面初始化线程池时，创建线程就需要让线程去调用对应的事物函数thread_routine(),在这里面我们首先需要检测当前线程池中的任务数量（没有任务下，当前对应的线程就需要等待，通过条件变量来控制，当下一次投放任务进来时会调用pthread_cond_signal唤醒这个线程），还有当前线程池所出的一个状态，是否需要销毁，执行对应的操作，当然这只是预先判断，下来我们要做的就是让当前的可用进程执行对应的任务，我们之前说过，线程池中的所有任务都存放在task_head所指一条向的任务链（函数链表）里面，每次通过加锁机制保证当前线程取得的是task_head这条链中的第一个任务，然后执行对应的任务

(*(work->task_process))(work->arg);

之后需要释放当前的任务。当然需要对线程池中的任务数量做出调整（减1）.对应的thread_routine()是：

void *thread_routine(void *arg){
	printf("[ starting thread %lu ]\n",pthread_self());
	while(1){
		pthread_mutex_lock(&(thread_pool->task_queue_lock));
		while(thread_pool->current_queue_size==0 && !thread_pool->destory_flag){
			printf("[ task_queue is null ,thread %lu is waitting ! ]\n",pthread_self());
			pthread_cond_wait(&(thread_pool->ready_event),&(thread_pool->task_queue_lock));
		}
		if(thread_pool->destory_flag){
			pthread_mutex_unlock(&(thread_pool->task_queue_lock));
			printf("[ thread %lu will got over now !]\n",pthread_self());
			pthread_exit(NULL);
		}
		printf("[ thread %lu is starting work ]\n",pthread_self());
		assert(thread_pool->current_queue_size!=0);
		assert(thread_pool->task_head!=NULL);
		thread_pool->current_queue_size--;
		Work_task *work=thread_pool->task_head;
		thread_pool->task_head=work->next;
		pthread_mutex_unlock(&(thread_pool->task_queue_lock));
		(*(work->task_process))(work->arg);
		free(work);
		work=NULL;
	}
}

   在此，我们通过互斥锁的机制保证了线程的任务执行流程，现在我们关心的是既然线程需要执行任务，那么怎么向线程池中投放任务，即我们需要一个函数实现任务投放，我们可以声明一个函数：

Void Add_task_to_threadpool(void *(*task_process)(void *arg),void *arg);

   我们之前说过线程池中的任务管理是通过一个Work_task类型的单链表实现的，我们每次去遍历这条任务链。过程其实就是建立一条单链表，将Add_task_to_threadpool()对应的task_process函数传递进去给单链表的一个节点。在这里，我使用尾插。完成后更新对应的线程池信息，最重要的是最后的pthread_cond_signal()，唤醒当前pthread_cond_wait阻塞的线程。

Void  Add_task_to_threadpool(void *(*task_process)(void *arg),void *arg){
	Work_task *new_work=(Work_task*)malloc(sizeof(Work_task));
	new_work->next=NULL;
	new_work->task_process=task_process;
	new_work->arg=arg;
	pthread_mutex_lock(&(thread_pool->task_queue_lock));
	Work_task *work=thread_pool->task_head;
	if(work!=NULL){
		while(work->next!=NULL)
		      work=work->next;
		work->next=new_work;
	}else
	      thread_pool->task_head=new_work;
	assert(thread_pool->task_head!=NULL);
	thread_pool->current_queue_size++;
	pthread_mutex_unlock(&(thread_pool->task_queue_lock));
	pthread_cond_signal(&(thread_pool->ready_event));
}

   到这里，我们完成了对线程池的任务投放，所需要的只是上面Add_task_to_threadpool()所需要的任务了，即task_process,这只是一个小小的任务测试函数，我们只去执行简单的打印操作（对应的线程id和任务id）便可。

void *work_process(void *arg){
	printf("[ threadid is %lu , working with taskid %d ]\n",pthread_self(),*(int *) arg);
	sleep(2);
	printf("[ workid %d over here with threadid %lu ]\n",*(int *) arg,pthread_self());
	return NULL;
}

   在多个线程处理完大于线程数目的任务量时（一个线程一次处理一个任务），我们需要的就是资源的回收和善后操作，阻塞所有的线程，销毁线程池，销毁任务队列，销毁条件变量和互斥锁。

Void destory_thread_pool(){
	int num;
	thread_pool->destory_flag=1;
	pthread_cond_broadcast(&(thread_pool->ready_event));
	for(num=0;num<thread_pool->max_thread_num;num++)
	      pthread_join(thread_pool->threadid_link[num],NULL);
	free(thread_pool->threadid_link);
	Work_task *crew=NULL;
	while(thread_pool->task_head!=NULL){
		crew=thread_pool->task_head;
		thread_pool->task_head=thread_pool->task_head->next;
		free(crew);
	}
	pthread_mutex_destroy(&(thread_pool->task_queue_lock));
	pthread_cond_destroy(&(thread_pool->ready_event));
	free(thread_pool);
	thread_pool=NULL;
	return 0;
}

   在此我们一个线程池从开始创建到最后的销毁以及中间的管理和运行流程就是这样。我们用这样的main函数来测试我们的线程池：

int main(int argc, char *argv[])
{
	int max_pthread_num=4;
	int work_routine_num=10;
	int tip;
	int *worknumber=(int *)malloc(sizeof(int)*work_routine_num);
	Thread_pool_init(max_pthread_num);
	for(tip=0;tip<work_routine_num;tip++){
		worknumber[tip]=tip;
		Add_task_to_threadpool(work_process,&worknumber[tip]);
	}
	sleep(5);
	destory_thread_pool();
	free(worknumber);
	return EXIT_SUCCESS;
}

测试运行结果如下   ：

一个简单的通过条件变量与加锁机制的线程池管理和执行任务操作的过程就是这样。但是其中涉及到的东西其实还是蛮复杂的，我们通过投放任务再以条件变量的形式唤醒等待线程。包括这种线程池的管理结构其实都是挺不错的思想。不多说了，写到这手都酸了。Over ！