C++多线程入门：简单的线程池

2024-04-07

字数统计: 1.3k字 | 阅读时长≈ 5分

很多时候我们并不会频繁的申请线程，而是采用线程池的方式，将任务加入到线程池中等待线程池中的空闲线程去处理该任务。

声明线程池基类

一个线程池应包括如下几个基本功能：

线程池初始化
启动所有线程
线程池入口

稍后为其增加添加任务的功能。

对于上述3个功能，提供如下实现：
直接将文件声明为hpp

#include <thread>
#include <vector>

class BaseThreadPool
{
public:
    void Init(int Num)；

    void Start();

private:
    void Run();

    int ThreadNum;
    std::mutex mtx;
    std::vector<Thread*> ThreadPool;
};

void BaseThreadPool::Init(int Num)
{
    std::unique_lock<std::mutex> Lock(mtx);
    ThreadNum=Num;
}

void BaseThreadPool::Start()
{
    std::unique_lock<std::mutex> Lock(mtx);
    if(ThreadNum<=0){
        return;
    }else if(!ThreadPool.empty()){
        return;
    }

    for(int i=0;i<ThreadNum;i++){
        auto th = new std::thread(&BaseThreadPool::Run,this);
        ThreadPool.push_back(th);
        th->detach();
    }
}

void BaseThreadPool::Run()
{
    using namespace std;
    cout << "Thread Run" << endl;

    cout << "Thread Exit" << endl;
}

至此完成了线程池的基本框架。
在Start函数中对线程数为0和线程池为空两种情况进行了处理，为了方便这里直接返回。

提供任务基类

提供一个任务基类，所有需要多线程调用的任务都需要继承自该类，由于该类并不需要实例，所以将其声明为抽象类。

class BaseTask
{
public:
    virtual void TaskRun() = 0;
    virtual ~BaseTask()
    {

    }
};

在需要继承的类中重写TaskRun以完成相应逻辑。

对线程池基类稍加改进

有了相应的任务类后为线程池类增加AddTask和GetTask方法：
使用list来存储所有任务，std::list<BaseTask*> TaskList
同时希望在增加一个任务后线程能立刻做出回应，所以再使用一个条件变量

#include "BaseTask.h"
#include <list>
#include <condition_vatiable>

void AddTask(BaseTask* Task)
{
	std::unique_lock<std::mutex> Lock(mtx);
	TaskList.push_back(Task);
	Lock.unlock();
	cv.notify_one();
}

BaseTask* GetTask()
{
	std::unique_lock<std::mutex> Lock(mtx);
	if (TaskList.empty()) {
		cv.wait(Lock);;
	}
	// 防止多次通知导致为空
	if (TaskList.empty()) {
		return nullptr;
	}
	auto task = TaskList.front();
	TaskList.pop_front();
	return task;
}

在完成对任务的添加和获取后需要对Run进行一些修改以便能够执行相应任务：

void BaseThreadPool::Run()
{
	using namespace std;
	cout << "Thread Run" << endl;
	while (true) {
		auto Task=GetTask();
		if (!Task) {
			continue;
		}
		try
		{
			Task->TaskRun();
		}
		catch (...)
		{
			// 处理异常，防止一个线程运行错误导致整个线程退出
		}
	}
	cout << "Thread Exit" << endl;
}

简单测试

#include <iostream>
#include <string>
#include "BaseThreadPool.h"

class TestTask:public BaseTask
{
public:
	virtual int TaskRun() override
	{
		std::cout << name << " Task Run" << std::endl;
		return 0;
	}

	std::string name;
};

int main()
{
	BaseThreadPool pool;
	pool.Init(4);
	pool.Start();

	TestTask t;
	t.name="Name01";
	pool.AddTask(&t);

	getchar();
	return 0;
}

更进一步完善线程池的功能

线程池不会一直运行下去，有时候需要结束并清空线程池，这时就需要未线程池增加一个退出功能。
未线程池增加bool变量bExit和设置bExit的函数。

void Exit()
{
	bExit=true;
	std::cout << "Exit" << std::endl;
	cv.notify_all();
	for(auto& th:ThreadPool){
		if(th.joinable()){
			th->join();
		}
	}
	std::unique_lock<std::mutex> Lock(mtx);
	for (auto& th : ThreadPool) {
		delete th;
		th=nullptr;
	}
}

bool bExit;

先设置bExit为true并通知所有线程退出，让线程去处理相应逻辑。
遍历所有线程池中的线程处理待办的任务
随后清空线程池

主要退出逻辑完成后还需要完善其它逻辑

首先当设置线程池退出后就不应该再有新的任务增加进来，所以我们对AddTask做出处理：

void AddTask()
{
	// 加锁逻辑
	if(bExit){
		// 额外处理，比如直接抛弃任务
		return;
	}
	//...
}

修改GetTask和Run

GetTask修改与AddTask类似：

void GetTask()
{
	// 这里是wait逻辑，Exit后进行了一次通知
	if(TaskList.empty()){
		return nullptr;
	}
	// 这里有一个相同的逻辑，可以公用一个
	//...
}

之前Run是一个死循环，这里将循环条件改为bExit即可

void Run()
{
	//...
	while(bExit){
		//...
	}
	//...
}

获取正在运行的线程数量

使用原子变量来存储正在运行的线程数量，原子变量本身就是线程安全的，所以可以直接改变值。
在使用时需要使用atomic头文件。
提供一个int类型原子变量和一个Get函数获取正在运行线程数：

#include <atomic>

std::atomic<int> RunningTaskNum{0};

void GetRunningTaskNum() { RunningTaskNum; }

在线程Run函数中，在线程执行前令RunningTaskNum自增（前置自增），并在任务结束后自减。

总结

通过提供任务基类和线程池基类实现了一个简单的线程池，后面我们会使用智能指针等技术对已有的线程池进行优化。

本文作者： KongXinQing
本文链接： https://13114987559.github.io/2024/04/07/note/C++多线程入门：简单的线程池/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！