linux下的线程池

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <pthread.h>
//定义条件变量
typedef struct condition{
	pthread_mutex_t pmutex;
	pthread_cond_t pcond;
}condition_t;

//定义任务结构体
typedef struct task{
	void *(*run) (void *arg);
	void *arg;
	struct task *next;
}task_t;
//定义线程池结构体
typedef struct threadpoll
{
	int count;//计数
	int idle;//闲置
	int max_threads;//最大线程
	int quit;//退出
	task_t *first;
	task_t *last;
	condition_t ready;
}threadpoll_t;



//初始化条件变量
int condition_init(condition_t *cond)
{
	int status;
	if(status=pthread_mutex_init(&cond->pmutex,NULL));//返回值是0则代表初始化成功
	return status;//创出状态值
	if(status=pthread_cond_init(&cond->pcond,NULL));
	return status;
	
	return 0;
}

int condition_lock(condition_t *cond)
{
	return pthread_mutex_lock(&cond->pmutex);
}

int condition_unlock(condition_t *cond)
{
	return pthread_mutex_unlock(&cond->pmutex);
}

int condition_wait(condition_t *cond)
{
	return pthread_cond_wait(&cond->pcond,&cond->pmutex);
}

int condition_timewait(condition_t *cond,const struct timespec* abstime)
{
	return pthread_cond_timedwait(&cond->pcond,&cond->pmutex,abstime);
}

int condition_signal(condition_t *cond)
{
	return pthread_cond_signal(&cond->pcond);
}

int condition_broadcast(condition_t *cond)
{
	return pthread_cond_broadcast(&cond->pcond);
}


int condition_destory(condition_t *cond)
{
	int status;
	if(status=pthread_mutex_destroy(&cond->pmutex));//返回值是0则代表初始化成功
	return status;//创出状态值
	if(status=pthread_cond_destory(&cond->pcond));
	return status;
	
	return 0;
	
}
//工作线程
void *thread_rountin(void *arg)
{
	//等待时长
    struct timespec abstime;
    int timeout;
	//打印线程ID
    printf("thread 0x%0x is starting\n",(int)pthread_self());
	//用于接收结构体
    threadpoll_t *poll = (threadpoll_t *)arg;
	//1、直接锁住
	//2、队列已满，阻塞队列不为满上
	//3、清空任务队列
	//4、添加任务到任务队列里，队尾指针往后挪
	//5、唤醒线程池，等待处理任务线程
	while(1)
	{
		timeout=0;
		//上锁
		condition_lock(&poll->ready);
		poll->idle++;//闲置线程++进行线程的闲置操作
        while(poll -> first == NULL && !poll -> quit)
        {
            printf("thread 0x%0x is waiting\n",(int)pthread_self());
            clock_gettime(CLOCK_REALTIME,&abstime);
            abstime.tv_sec += 2;
            int status=condition_timewait(&poll -> ready,&abstime);
            if(status == ETIMEDOUT)
            {
                printf("thread 0x%0x is wait timed out\n",(int)pthread_self());
                timeout = 1;
                break;
            }

        }
		//等待队列有任务到来或者线程池销毁的通知
		poll->idle--;
		//如果有任务
		if(poll->first!=NULL)
		{
			//移动下标
			task_t *t=poll->first;
			poll->first=t->next;
			 //需要先解锁，以便添加新任务。其他消费者线程能够进入等待任务。
            condition_unlock(&poll -> ready);
            t -> run(t->arg);
            free(t);
            condition_lock(&poll -> ready);
		}
		//等待线程销毁的通知
		 if(poll -> quit && poll ->first == NULL)
        {
            poll -> count--;
            if(poll->count == 0)
            {
                condition_signal(&poll -> ready);
            }
            condition_unlock(&poll->ready);
            //跳出循环之前要记得解锁
            break;
        }
		 if(timeout &&poll -> first ==NULL)
        {
            poll -> count--;
            condition_unlock(&poll->ready);
            //跳出循环之前要记得解锁
            break;
        }
        condition_unlock(&poll -> ready);
	}
	 printf("thread 0x%0x is exiting\n",(int)pthread_self());
    return NULL;
}

//初始化线程池
void  poll_init(threadpoll_t *poll,int threads)
{
	condition_init(&poll->ready);
	poll->first=NULL;
	poll->last=NULL;
	poll->count=0;
	poll->idle=0;
	poll->max_threads=threads;
	poll->quit=0;
	
}

//创建任务
void pthrad_add_task(threadpoll_t *poll, void *(*run)(void *arg),void *arg){
	//创建任务 添加任务
	task_t *newtask=(task_t *)malloc(sizeof(task_t));
	newtask->run=run;
	newtask->arg=arg;
	newtask->next=NULL;
	
	//上锁
	condition_lock(&poll->ready);
	//添加任务到队列
	if(poll->first==NULL)
	{
		poll->first=newtask;
	}
	else
		poll->last->next=newtask;
	poll->last=newtask;
	 //如果有等待线程，则唤醒其中一个
    if(poll -> idle > 0)
    {
		//如果没有闲置的的,则唤醒其中一个
        condition_signal(&poll -> ready);
    }
	else if(poll -> count< poll -> max_threads)
    {
		//如果没有则创建工作线程
        pthread_t tid;
		//执行回调
        pthread_create(&tid,NULL,thread_rountin,poll);
		//忙线程加一
        poll -> count++;
    }
    condition_unlock(&poll -> ready);
}

//销毁线程池
void pthread_task_destory(threadpoll_t *poll)
{
	if(poll->quit)
	{
		return;
	}
	condition_lock(&poll->ready);
	 poll->quit = 1;
    if(poll -> count > 0)
    {
        if(poll -> idle > 0)
            condition_broadcast(&poll->ready);

        while(poll -> count > 0)
        {
            condition_wait(&poll->ready);
        }
    }
    condition_unlock(&poll->ready);
    condition_destory(&poll -> ready);
}


//回调函数
void *mytask(void *arg)
{
    printf("thread 0x%0x is working on task %d\n",(int)pthread_self(),*(int*)arg);
    sleep(1);
    free(arg);
    return NULL;
}
 int main(){
	//1、没有写socket 不然在300行也写不完，用简单的for循环模拟客户端生产的数据每个线程处理数据
	//gcc main.c -o main -lpthread -lrt
    threadpoll_t pool;
	//创建线程池
    poll_init(&pool,100);
    
	//模拟客户端领数据
    int i ;
    for(i = 0; i < 100; i++)
    {
        int *arg = (int *)malloc(sizeof(int));
        *arg = i;
		//向线程池添加任务	
        pthrad_add_task(&pool,mytask,arg);
    }
    //为了等待其他线程结束 当然也可以通过pthread_join来做
    sleep(15);
	//线程销毁
    pthread_task_destory(&pool);
    return 0;
 }