线程池精讲（pthread pool）

青旅半醒 2022-12-10 10:29 191阅读 0赞

线程池也是多线程的处理方式。是将“生产者”线程提出任务添加到“任务
    队列”，然后一些线程自动完成“任务队列”上的任务。
    
    
    多线程编程，创建一个线程，指定去完成某一个任务，等待线程的退出。
    虽然能够满足编程需求，但是当我们需要创建大量的线程的时候，在创建
    过程以及销毁线程的过程中可能会消耗大量的CPU.增加很大开销。
    如： 
    	文件夹的copy
    	WEB服务器的响应。
    	。。。。
    	。。。。
    	
    线程池就是用来解决类似于这样的一个问题的，可以降低频繁地创建和销毁
    线程所带来地开销。
    
    线程池技术思路：
    	一般采用预创建线程技术，也就是提前把需要用线程先创建一定数目。
    	这些线程提前创建好了之后，“任务队列”里面假设没有任务，那么就
    	让这些线程休眠，一旦有任务，就唤醒线程去执行任务，任务执行完了，
    	也不需要去销毁线程，直到当你想退出或者是关机时，这个时候，那么你
    	调用销毁线程池地函数去销毁线程。
    	
    	线程完成任务之后不会销毁，而是自动地执行下一个任务。。。。
    	而且，当任务有很多，你可以有函数接口去增加线程数量，
    	当任务较少时，你可以有函数接口去销毁部分线程。。。。。
    	
    	如果，创建和销毁线程的时间对比执行任务的时间可以忽略不计，
    	那么我们在这种情况下面也就没有必要用线程池。
    	
    	
    	“任务队列”是一个共享资源“互斥访问”
    	
    	
    	线程池本质上也是一个数据结构，需要一个结构体去描述它：
    	
    	
    struct pthread_pool //线程池的实现
    {
    	//一般会有如下成员
    
    	
    	
    	//互斥锁，用来保护这个“任务队列”
    	pthread_mutex_t lock; //互斥锁 
    	
    	//线程条件变量 表示“任务队列”是否有任务
    	pthread_cond_t cond; //条件变量
    	
    	bool shutdown; //表示是否退出程序 bool：类型 false / true
    	
    
    	//任务队列(链表)，指向第一个需要指向的任务
    	//所有的线程都从任务链表中获取任务 "共享资源"
    	struct task * task_list;
    	
    	//线程池中有多个线程，每一个线程都有tid, 需要一个数组去保存tid
    	pthread_t * tids; //malloc() 
    	
    	//线程池中正在服役的线程数，当前线程个数
    	unsigned int active_threads;
    	
    	
    	//线程池任务队列最大的任务数量
    	unsigned int max_waiting_tasks;
    	
    	
    	//线程池任务队列上当前有多少个任务
    	unsigned int cur_waiting_tasks;
    	
    	//......
    
    };
    
    
    //任务队列(链表)上面的任务结点，只要能够描述好一个任务就可以了，
    //线程会不断地任务队列取任务
    struct task  //任务结点 
    {
    	// 1. 任务结点表示的任务，“函数指针”指向任务要执行的函数(cp_file)
    	void*(* do_task)(void * arg);
    	
    	//2. 指针，指向任务指向函数的参数(文件描述符)
    	void * arg;
    	
    	//3. 任务结点类型的指针，指向下一个任务
    	struct task * next;
    };
    
    
    操作线程池所需要的函数接口:
    	pthread_pool.c 
    	pthread_pool.h
    	
    	把“线程池”想象成一个外包公司
    	你需要去完成的就是操作线程池所提供的函数接口。
    	
    	
    	
    
    /*
    	init_pool: 线程池初始化函数，初始化指定的线程池中有thread_num
    		个初始线程
    	@pool：指针，指向您要初始化的那个线程池
    	@threa_num: 您要初始化的线程池中开始的线程数量
    	返回值： 
    		成功 0
    		失败 -1
    */
    
    int init_pool(pthread_pool * pool , unsigned int threa_num)
    {
    	//初始化线程池的结构体
    	
    	//初始化线程互斥锁
    	
    	//初始化线程条件变量
    	
    	//创建thread_num个线程，并且让线程去执行任务调配函数，
    	//记录所有线程的tid
    	
    }
    
    
    /*
    	routine: 任务调配函数。
    		所有线程开始都执行此函数，此函数会不断的从线程池的任务队列
    		中取下任务结点，去执行。
    		
    		任务结点中包含“函数指针” h "函数参数"
    */
    
    void * routine(void * arg)
    {
    	//arg表示你的线程池的指针
    	
    	while()
    	{
    		//获取线程互斥锁，lock 
    		
    		//当线程池没有结束的时候，不断地从线程池的任务队列取下结点
    		//去执行。
    		
    		
    		//释放线程互斥锁，unlock
    		
    		//释放任务结点
    	}
    }
    
    
    
    /*
    	destroy_pool: 销毁线程池，销毁前要保证所有的任务已经完成
    */
    
    int destroy_pool(pthread_pool * pool)
    {
    	//释放所有空间 等待任务执行完毕(join)。
    	//唤醒所有线程
    	//利用join函数回收每一个线程资源。
    }
    
    
    
    /*
    	add_task：给任务队列增加任务， 把do_task指向的任务(函数指针)和
    		arg指向的参数保存到一个任务结点，添加到pool任务队列中。
    		
    	@pool : 您要添加任务的线程池
    	@do_task ： 您需要添加的任务(cp_file)
    	@arg： 您要执行的任务的参数(文件描述符)
    */
    
    
    int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg)
    {
    	//把第二个参数和第三个参数封装成struct task 
    	
    	
    	//再把它添加到 pool->task 任务队列中去
    	
    	//注意任务队列是一个共享资源
    	
    	//假如任务后要唤醒等待的线程。
    }
    
    
    
    //如果任务多的时候，往线程池中添加线程  pthread_create
    int add_threads(pthread_pool * pool, unsigned int num);
    {
    	//新创建num个线程，让每一个线程去执行线程调配函数
    	
    	//将每一个新创建的线程tid，添加到pool-> tids 
    }
    
    //如果任务少的时候，减少线程池中线程的数量 pthread_cancel join
    int remove_threads(pthread_pool * pool, unsigned int num)
    {
    	//用pthread_cancel取消num个线程 
    	//利用pthread_join函数去回收资源。
    }

## pthread\_pool.c ##

#include "pthread_pool.h"
    
    /* init_pool: 线程池初始化函数，初始化指定的线程池中有thread_num 个初始线程 @pool：指针，指向您要初始化的那个线程池 @threa_num: 您要初始化的线程池中开始的线程数量 返回值： 成功 0 失败 -1 */
    
    int init_pool(pthread_pool * pool , unsigned int threa_num)
    { 
    	//初始化线程池的结构体
    	
    	//初始化线程互斥锁
    	pthread_mutex_init(&pool->lock, NULL);
    	
    	//初始化线程条件变量
    	pthread_cond_init(&pool->cond, NULL);
    
    	pool->shutdown = false ;// 不退出
    
    	pool->task_list = (struct task*)malloc(sizeof(struct task));
    
    	pool->tids = (pthread_t *)malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);
    	if(pool->task_list == NULL || pool->tids == NULL)
    	{ 
    		perror("malloc memery error");
    		return -1;
    	}
    
    	pool->task_list->next = NULL;
    
    	//线程池中一开始初始化多少个线程来服役
    	pool->active_threads = threa_num;
    
    	//表示线程池中最多有多少个任务
    	pool->max_waiting_tasks = MAX_WAITING_TASKS;
    
    	//线程池中任务队列当前的任务数量
    	pool->cur_waiting_tasks = 0;
    
    	
    	//创建thread_num个线程，并且让线程去执行任务调配函数，
    	//记录所有线程的tid
    	int i = 0;
    	for(i = 0; i < threa_num; i++)
    	{ 
    		int ret = pthread_create(&(pool->tids)[i], NULL, routine, (void*)pool);
    		if(ret != 0)
    		{ 
    			perror("create thread error");
    			return -1;
    		}
    
    		printf("[%lu]:[%s] ===> tids[%d]:[%lu]",pthread_self(),
    			__FUNCTION__, i , pool->tids[i]);
    	}
    
    	return 0;
    	
    	
    }
    
    /* routine: 任务调配函数。 所有线程开始都执行此函数，此函数会不断的从线程池的任务队列 中取下任务结点，去执行。 任务结点中包含“函数指针” h "函数参数" */
    
    void * routine(void * arg)
    { 
    	//arg表示你的线程池的指针
    	
    	while()
    	{ 
    		//获取线程互斥锁，lock 
    		
    		//当线程池没有结束的时候，不断地从线程池的任务队列取下结点
    		//去执行。
    		
    		
    		//释放线程互斥锁，unlock
    		
    		//释放任务结点
    	}
    }
    
    /* destroy_pool: 销毁线程池，销毁前要保证所有的任务已经完成 */
    
    int destroy_pool(pthread_pool * pool)
    { 
    	//释放所有空间 等待任务执行完毕(join)。
    	//唤醒所有线程
    	//利用join函数回收每一个线程资源。
    }
    
    /* add_task：给任务队列增加任务， 把do_task指向的任务(函数指针)和 arg指向的参数保存到一个任务结点，添加到pool任务队列中。 @pool : 您要添加任务的线程池 @do_task ： 您需要添加的任务(cp_file) @arg： 您要执行的任务的参数(文件描述符) */
    
    int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg)
    { 
    	//把第二个参数和第三个参数封装成struct task 
    	
    	
    	//再把它添加到 pool->task 任务队列中去
    	
    	//注意任务队列是一个共享资源
    	
    	//假如任务后要唤醒等待的线程。
    }
    
    //如果任务多的时候，往线程池中添加线程 pthread_create
    int add_threads(pthread_pool * pool, unsigned int num);
    { 
    	//新创建num个线程，让每一个线程去执行线程调配函数
    	
    	//将每一个新创建的线程tid，添加到pool-> tids 
    }
    
    //如果任务少的时候，减少线程池中线程的数量 pthread_cancel join
    int remove_threads(pthread_pool * pool, unsigned int num)
    { 
    	//用pthread_cancel取消num个线程 
    	//利用pthread_join函数去回收资源。
    }

## pthread\_pool.h ##

#ifndef __PTHREAD_POOL_H__
    #define __PTHREAD_POOL_H__
    
    //表示线程池中最多有多少个线程
    #define MAX_ACTIVE_THREADS 20
    
    //表示线程池中最多有多少个任务
    #define MAX_WAITING_TASKS 1024
    
    //任务队列(链表)上面的任务结点，只要能够描述好一个任务就可以了，
    //线程会不断地任务队列取任务
    struct task  //任务结点 
    { 
    	// 1. 任务结点表示的任务，“函数指针”指向任务要执行的函数(cp_file)
    	void*(* do_task)(void * arg);
    	
    	//2. 指针，指向任务指向函数的参数(文件描述符)
    	void * arg;
    	
    	//3. 任务结点类型的指针，指向下一个任务
    	struct task * next;
    };
    
    struct pthread_pool //线程池的实现
    { 
    	//一般会有如下成员
    
    	//互斥锁，用来保护这个“任务队列”
    	pthread_mutex_t lock; //互斥锁 
    	
    	//线程条件变量 表示“任务队列”是否有任务
    	pthread_cond_t cond; //条件变量
    	
    	bool shutdown; //表示是否退出程序 bool：类型 false / true
    	
    	//任务队列(链表)，指向第一个需要指向的任务
    	//所有的线程都从任务链表中获取任务 "共享资源"
    	struct task * task_list;
    	
    	//线程池中有多个线程，每一个线程都有tid, 需要一个数组去保存tid
    	pthread_t * tids; //malloc() 
    	
    	//线程池中正在服役的线程数，当前线程个数
    	unsigned int active_threads;
    	
    	//线程池任务队列最大的任务数量
    	unsigned int max_waiting_tasks;
    	
    	//线程池任务队列上当前有多少个任务
    	unsigned int cur_waiting_tasks;
    	
    	//......
    
    };
    
    /* init_pool: 线程池初始化函数，初始化指定的线程池中有thread_num 个初始线程 @pool：指针，指向您要初始化的那个线程池 @threa_num: 您要初始化的线程池中开始的线程数量 返回值： 成功 0 失败 -1 */
    
    int init_pool(pthread_pool * pool , unsigned int threa_num);
    
    /* routine: 任务调配函数。 所有线程开始都执行此函数，此函数会不断的从线程池的任务队列 中取下任务结点，去执行。 任务结点中包含“函数指针” h "函数参数" */
    
    void * routine(void * arg);
    
    /* destroy_pool: 销毁线程池，销毁前要保证所有的任务已经完成 */
    
    int destroy_pool(pthread_pool * pool);
    
    /* add_task：给任务队列增加任务， 把do_task指向的任务(函数指针)和 arg指向的参数保存到一个任务结点，添加到pool任务队列中。 @pool : 您要添加任务的线程池 @do_task ： 您需要添加的任务(cp_file) @arg： 您要执行的任务的参数(文件描述符) */
    
    
    int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg);
    
    //如果任务多的时候，往线程池中添加线程 pthread_create
    int add_threads(pthread_pool * pool, unsigned int num);
    
    //如果任务少的时候，减少线程池中线程的数量 pthread_cancel join
    int remove_threads(pthread_pool * pool, unsigned int num);
    
    #endif