【Linux】实现一个简单的线程池

清疚 2021-12-03 17:53 300阅读 0赞

#### 线程池的概念 ####

“池”的概念大家应该不陌生，线程池顾名思义就是存放线程的“池子”。  
它是一种线程的使用模式。我们发现如果**线程过多会带来调度的开销**，进而影响缓存的局部性和整体性能。  
而线程池维护着一定数量的线程，等待监督管理者分配可并发执行的任务，从而避免了在处理短时间任务时创建和销毁线程的代价。所以**线程池不仅能够保证内核的充分利用，还能防止线程的过分调度。**

> 总结：线程池很好的解决了在处理任务时的效率问题，因为当接到任务时既可以立马处理，并且对于短处理时间任务不用付出创建和销毁的代价。  
> 线程池的另一个好处就是保证了调度的合理性，如果我们没有上限的创建线程，有可能导致调度周期变长，甚至创建线程数量过度时程序可能直接挂掉。

确定可用线程数量的考虑因素：

*  CPC的个数 （可用的并发处理器）
 *  CPC的核数 （或者内存）
 *  任务的类别 （IO密集型/计算密集型）

#### 线程池的应用场景 ####

1.  需要大量的线程来完成任务，且完成任务的时间比较短。  
    例如WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。  
    但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了，因为Telnet会话时间比线程的创建时间大多了。
2.  对性能要求苛刻的应用，比如要求服务器迅速响应客户请求
3.  接受突发性的大量请求，但请求数量小于线程池中线程的数量。  
    突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上操作系统中线程数目最大值不是问题，但短时间内产生大量线程可能使内存到达极限，从而出现错误

#### 线程池的实现 ####

要想自己学习实现一个简易的线程池，那么必须具备以下知识：

>  *  熟悉队列操作
>  *  熟悉互斥锁
>  *  熟悉条件变量
>  *  类的封装与使用

接下来我们通过线程池实现一个计算器：

1.  我们实现线程池的主要目的就是执行任务，所以我们应该先定义一个任务的类描述我们的任务（计算器），类中包括处理数据和处理方法，以及最后的处理过程

> 这里 x,y为待处理数据，op是运算方式（0 1 2 3 分别表示 + - \* /），handler是处理方式（cal）

int cal(int x, int y, int op);
    typedef int(*HandlerTask_t) (int x, int y, int op);
    class Task{
    private:         //定义用户要处理的数据和方式
    	int x，y，op;               
    	HandlerTask_t handler;
    public:
    	Task(int _x = -1, int _y = -1, int _op = -1) :x(_x), y(_y), op(_op){}   //构造函数
    	
    	void Register(HandlerTask_t _handler)   //从外部注册处理方式
    	{		handler = _handler;   }   //通过t.Register(cal);实现注册handler为cal运算
    	
    	void Run()        //进行处理及结果
    	{ }
    	~Task(){}
    };

1.  创建线程池

class ThreadPool{
    private:
    	int thread_nums;//线程池中的线程总数
    	int idle_nums;//当前闲置线程的数量
    	queue<Task> task_queue;//线程所要处理的任务队列
    	pthread_mutex_t lock;
    	pthread_cond_t cond;
    public:
    	ThreadPool() {}
    	~ThreadPool() {}
    };

1.  添加任务，唤醒线程

void WakeupThread()
    	{
    		pthread_cond_signal(&cond);
    	}
    	
    void PushTask(const Task& t)       //添加线程需要执行的任务
    	{
    		LockQueue();
    		task_queue.push(t);    //添加任务
    		WakeupThread();      //唤醒线程
    		UnlockQueue();
    	}

1.  创建线程，执行任务

void ThreadIdle()       //线程等待
    	{
    		idle_nums++;   //一旦进入等待状态闲置线程加一
    		pthread_cond_wait(&cond, &lock);    //线程等待
    		idle_nums--;     //等待条件成熟，被唤醒闲置线程减一
    	}
    	
    static void* ThreadRotinue(void* arg)      //线程执行任务过程
    	{
    		pthread_detach(pthread_self());      //线程分离 
    		ThreadPool *tp = (ThreadPool*)arg;
    		for (;;){
    			tp->LockQueue();
    			while (tp->QueueIsEmpty()){
    				tp->ThreadIdle();         //任务队列为空时，线程等待
    			}
    			Task t;
    			tp->PopTask(t);   
    			tp->UnlockQueue();
    			t.Run();
    		}
    	}
    	void InitThreadPool()     //创建所需要的线程
    	{
    		pthread_t t;
    		for (auto i = 0; i < thread_nums; i++)
    		{
    			pthread_create(&t, NULL, ThreadRotinue, this);   //线程执行ThreadRotinue
    		}
    	}

#### 思考 ####

*  问题1：关于`static void* ThreadRotinue(void* arg)` 函数头为什么要加static？

大家都知道这里pthread\_create函数执行的回调函数必须是void\* fun(void\*)类型的！！可是因为现在这个函数我们最后要写在类中，但是成员函数的特性隐藏了一个this指针，我们为了满足pthread\_create函数要求，我们不得不将函数改成静态的，所以这也就是为什么创建线程时需要将this指针传给此函数。正如你所看到的，后面定义的`ThreadPool *tp = (ThreadPool*)arg;`就是this指针，并且之后调用线程池中的其他函数都需要使用tp指针。

*  问题2：等待生产者生产为什么需要使用while循环？if不就可以了么？

其实不然，如果等待函数被误唤醒或者等待失败，那么其实队列中现在并没有任务可以处理的任务，然而if的话会继续执行下去，从而程序就会崩溃。

*  问题3：为什么还要封装一个PopTask()函数？直接调用不就行了么？

这里就回到问题1，我们这个函数已经被转换为静态成员了，我们此时无法访问类的私有成员，所以必须另外封装一个接口

*  问题4：t.Run()是让任务执行的调用，为什么不放在锁里执行，这样不会有线程安全问题么？

这里真的非常十分爆炸重要，你可千万不要小看这一句代码，让我们思考一下，如果你已经得到了任务，还需要锁保护么？意思就是说，这个任务被你拿到就一定属于你了，以后不会再有任何人和你竞争了，这样放在锁外就没有任何问题了，但是这不是放在锁外的真正目的。如果你将这句代码放在锁中执行真是让人毛骨悚然，因为你在任务执行时拿着锁不释放，其他线程就会被阻塞住，从而出现线程的饥饿问题，而且你的程序居然变成了串行的了，天哪，本来我们就是为了效率而写的线程池，现在居然效率还不如之前，这不是笑话么，所以一定要擦亮眼睛看清这句代码的位置。

#### 完整代码 ####

#include<iostream>
    #include<pthread.h>
    #include<time.h>
    #include<unistd.h>
    #include<stdlib.h>
    #include<queue>
    using namespace std;
    
    int cal(int x, int y, int op);
    typedef int(*HandlerTask_t) (int x, int y, int op);
    class Task{
    private:
    	int x;
    	int y;
    	int op;      //0+ 1- 2* 3/
    	HandlerTask_t handler;
    public:
    	Task(int _x = -1, int _y = -1, int _op = -1) :x(_x), y(_y), op(_op){}
    	void Register(HandlerTask_t _handler)     //由外界提供处理方式
    	{
    		handler = _handler;
    	}
    	void Run()        //执行处理并展示处理结果         
    	{
    		int ret = handler(x, y, op);
    		const char* arr = "+-*/";
    		cout << pthread_self() << " : " << x << arr[op] << y << "=" << ret << endl;
    	}
    	~Task(){}
    };
    
    class ThreadPool{
    private:
    	int thread_nums;//线程池中的线程总数
    	int idle_nums;//当前闲置线程的数量
    	queue<Task> task_queue;//线程所要处理的任务队列
    	pthread_mutex_t lock;
    	pthread_cond_t cond;
    public:
    	void LockQueue()
    	{
    		pthread_mutex_lock(&lock);
    	}
    	void UnlockQueue()
    	{
    		pthread_mutex_unlock(&lock);
    	}
    	bool QueueIsEmpty()
    	{
    		return task_queue.size() == 0 ? true : false;
    	}
    	void ThreadIdle()       //线程等待
    	{
    		idle_nums++;   //一旦进入等待状态闲置线程加一
    		pthread_cond_wait(&cond, &lock);    //线程等待
    		idle_nums--;     //等待条件成熟，被唤醒闲置线程减一
    	}
    	void WakeupThread()
    	{
    		pthread_cond_signal(&cond);
    	}
    	void PopTask(Task& t)     //任务被执行
    	{
    		t = task_queue.front();
    		task_queue.pop();
    	}
    public:
    	ThreadPool(int _num = 5) :thread_nums(_num), idle_nums(0)
    	{
    		pthread_mutex_init(&lock, NULL);
    		pthread_cond_init(&cond, NULL);
    	}
    	static void* ThreadRotinue(void* arg)      //线程执行任务过程
    	{
    		pthread_detach(pthread_self());      //分离线程
    		ThreadPool *tp = (ThreadPool*)arg;
    		for (;;){
    			tp->LockQueue();
    			while (tp->QueueIsEmpty()){
    				tp->ThreadIdle();         //任务队列为空时，线程等待
    			}
    			Task t;
    			tp->PopTask(t);    //否则执行任务
    			tp->UnlockQueue();
    			t.Run();
    		}
    	}
    	void InitThreadPool()     //创建所需要的线程
    	{
    		pthread_t t;
    		for (auto i = 0; i < thread_nums; i++)
    		{
    			pthread_create(&t, NULL, ThreadRotinue, this);   //线程需要完成的任务
    		}
    	}
    	void PushTask(const Task& t)       //添加线程需要执行的任务
    	{
    		LockQueue();
    		task_queue.push(t);    //添加任务
    		WakeupThread();      //唤醒线程
    		UnlockQueue();
    	}
    	~ThreadPool()
    	{
    		pthread_mutex_destroy(&lock);
    		pthread_cond_destroy(&cond);
    	}
    };
    
    int cal(int x, int y, int op)      //自由定义数据处理方法
    {
    	int ret = -1;
    	switch (op){
    	case 0:
    		ret = x + y;
    		break;
    	case 1:
    		ret = x - y;
    		break;
    	case 2:
    		ret = x * y;
    		break;
    	case 3:
    		ret = x / y;
    		break;
    	default:
    		std::cout << "cal error!" << std::endl;
    	}
    	return ret;
    }
    int main()
    {
    	ThreadPool tp;
    	tp.InitThreadPool();    //创建线程池中
    	srand((unsigned long)time(NULL));
    	for (;;){
    		int x = rand() % 100 + 1;
    		int y = rand() % 100 + 1;
    		int op = rand() % 4;
    		Task t(x, y, op);
    		t.Register(cal);     //注册方法
    		tp.PushTask(t);     //添加任务
    		sleep(1);
    	}
    	return 0;
    }

喜欢博主文章的记得关注哦~