线程池的核心实现

一句话总结：用队列实现，生产者添加任务，消费者执行任务；注意原子性操作。

#include <iostream>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <pthread.h>
#include <queue>
using namespace std;
#define THREAD_NUM 3
#define QUEUE_SIZE 3
typedef boost::function<void ()> Task;
queue<Task> queueTask;
pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t g_cond = PTHREAD_COND_INITIALIZER;

Task take()
{
	pthread_mutex_lock(&g_mutex);
	if(queueTask.empty())
		cout <<"queue task is empty" <<endl;
	while (queueTask.empty())
	{
		pthread_cond_wait(&g_cond, &g_mutex);
	}
	Task task;

	if(!queueTask.empty())
		cout <<"queue task is not empty" <<endl;
	if (!queueTask.empty())
	{
		task = queueTask.front();
		queueTask.pop();
		pthread_cond_signal(&g_cond);
	}
	pthread_mutex_unlock(&g_mutex);
	return task;
}

void add(Task& task)
{
	pthread_mutex_lock(&g_mutex);
	if(queueTask.size() >= QUEUE_SIZE)
		cout <<"queue task is full" <<endl;
	while(queueTask.size() >= QUEUE_SIZE)
	{
		pthread_cond_wait(&g_cond, &g_mutex);
	}
	queueTask.push(task);
	pthread_cond_signal(&g_cond);
	pthread_mutex_unlock(&g_mutex);
	return;
}

void* func(void* arg)
{
	while (1)
	{
		Task task(take());
		if (task)
		{
			cout << "run task........."<<endl;
			task();
		}
	}
}

void print()
{
	cout<<"execute thread pool task............" <<endl;
}

int main()
{
	cout<<"test thread pool, pool size is 3" <<endl;
	pthread_t thread[THREAD_NUM];
	for (int i = 0; i < THREAD_NUM; ++i)
	{
		int ret = pthread_create(&thread[i], NULL, func, NULL);
		if (ret != 0)
		{
			return 1;
		}
	}
	boost::function<void()> fp = boost::bind(print);

	int cnt = 0;
	cout <<"add 5 tasks " <<endl;
	while(cnt < 5)
	{
		add(fp);
		++cnt;
	}
	sleep(3);	
	cout <<"=================add another task==============="<<endl;
	sleep(3);
	add(fp);

	for(int i=0; i<THREAD_NUM; ++i)
	{
		pthread_join(thread[i], NULL);
	}
	add(fp);
	return 0;
}