使用 C++11 编写可复用多线程任务池

类的功能

Task （任务基类）

该类主要实现一个任务类

virtual int doWork() = 0;
TaskQueue （任务队列）

该类主要针对任务的存储、删除、撤回等状态做管理
ThreadPool （线程池）

整个线程池的核心业务处理类

代码

Task.h

//任务的基类

#pragma once

#include <time.h>

#include <atomic>

//任务的基类

class Task

{

public:

	//构造、析构函数

	Task():_id(_nRequestID++),_isCancelRequired(false),_createTime(clock()){}

	~Task(){};

	// 任务类虚接口，继承这个类的必须要实现这个接口

	virtual int doWork(void) = 0;

	// 任务已取消回调

	virtual int onCanceled(void)

	{

		return  1;

	}

	// 任务已完成

	virtual int onCompleted(int)

	{

		return 1;

	}

	// 任务是否超时

	virtual bool isTimeout(const clock_t& now)

	{

		return ((now - _createTime) > 5000);

	}

	// 获取任务ID

	size_t getID(void)

	{

		return _id;

	}

	//获取任务取消状态

	bool isCancelRequired(void)

	{

		return _isCancelRequired;

	}

	//设置任务取消状态

	void setCancelRequired(void)

	{

		_isCancelRequired = true;

	}

protected:

	size_t _id;	//任务的唯一标识

	clock_t _createTime;	//任务创建时间，非Unix时间戳

private:

	static std::atomic<size_t> _nRequestID;

	std::atomic<bool> _isCancelRequired;	//任务取消状态

};

//selectany可以让我们在.h文件中初始化一个全局变量而不是只能放在.cpp中。

//这样的代码来初始化这个全局变量。既是该.h被多次include，链接器也会为我们剔除多重定义的错误。

__declspec(selectany) std::atomic<size_t> Task::_nRequestID = 100000;

TaskQueue.h

#pragma once

#include <deque>

#include <mutex>

#include <condition_variable>

#include <unordered_map>

#include <memory>

#include <thread>

//任务队列

template<typename T>

class TaskQueue

{

public:

	//向队列的末尾插入任务,task是任务类

	void put_back(std::shared_ptr<T>& task)

	{

		std::unique_lock<std::mutex> lock(_mutexQueue);

		_queue.push_back(task);

		_conditPut.notify_one();

	}

	//向队列的头部插入任务

	void put_front(std::shared_ptr<T>& task)

	{

		std::unique_lock<std::mutex> lock(_mutexQueue);

		_queue.push_front(task);

		_conditPut.notify_one();

	}

	//获取队首（并将任务加到运行任务列表中），返回tase是任务类

	std::shared_ptr<T> get(void) {

		std::unique_lock<std::mutex> lock(_mutexQueue);

		if (_queue.empty())

			return nullptr;

		//lock_guard取代了mutex的lock()和unlock();

		std::lock_guard<std::mutex> lock_doing_task(_mutexDoingTask);

		std::shared_ptr<T>& task = _queue.front();

		_mapDoingTask.insert(std::make_pair(task->getID(), task));

		_queue.pop_front();

		return task;

	}

	//获取双向链表queue的大小

	size_t size(void)

	{

		std::unique_lock<std::mutex> lock(_mutexQueue);

		return _queue.size();

	}

	//释放队列

	void release(void)

	{

		deleteAllTasks();

		_conditPut.notify_all();

	}

	//删除任务（从就绪队列删除，如果就绪队列没有，则看执行队列有没有，有的话置下取消状态位）

	int deleteTask(size_t nID)

	{

		std::unique_lock<std::mutex> lock(_mutexQueue, std::defer_lock);

		lock.lock();

		auto it = _queue.begin();

		for (; it != _queue.end(); ++it)

		{

			if ((*it)->getID() == nID)

			{

				_queue.erase(it);

				lock.unlock();

				return 0;

			}

		}

		//下面的逻辑可能会造成死锁，这里要及时释放

		lock.unlock();

		// 试图取消正在执行的任务

		{

			std::lock_guard<std::mutex> lock_doing_task(_mutexDoingTask);

			auto it_map = _mapDoingTask.find(nID);

			if (it_map != _mapDoingTask.end())

				it_map->second->setCancelRequired();

		}

		//任务执行完后再返回

		while (_mapDoingTask.count(nID))

			std::this_thread::sleep_for(std::chrono::milliseconds(20));

		return 0;

	}

	//删除所有任务

	int deleteAllTasks(void)

	{

		std::unique_lock<std::mutex> lock(_mutexQueue, std::defer_lock);

		lock.lock();

		if (!_queue.empty())

			_queue.clear();//清空

		{

			std::lock_guard<std::mutex> lock_doing_task(_mutexDoingTask);

			if (!_mapDoingTask.empty())

			{

				auto it_map = _mapDoingTask.begin();

				for (; it_map != _mapDoingTask.end(); ++it_map)

					it_map->second->setCancelRequired();

			}

		}

		lock.unlock();

		//任务执行完后再返回

		while (!_mapDoingTask.empty())

			std::this_thread::sleep_for(std::chrono::milliseconds(50));

		return 0;

	}

	//任务完成回调（从运行列表中删除指定任务）

	int onTaskFinished(size_t nID)

	{

		std::lock_guard<std::mutex> lock_doing_task(_mutexDoingTask);

		auto it_map = _mapDoingTask.find(nID);

		if (it_map != _mapDoingTask.end())

			_mapDoingTask.erase(it_map);

		return 0;

	}

	//判断任务是否执行完毕

	std::shared_ptr<T> isTaskProcessed(size_t nId)

	{

		std::lock_guard<std::mutex> lock_queue(_mutexQueue);

		auto it = _queue.begin();

		for (; it != _queue.end(); ++it) {

			if ((*it)->getID() == nId)

				return *it;

		}

		std::lock_guard<std::mutex> lock_doing_task(_mutexDoingTask);

		auto it_map = _mapDoingTask.find(nId);

		if (it_map != _mapDoingTask.end())

			return it_map->second;

		return nullptr;

	}

	//等待有任务到达（带超时：超时自动唤醒）

	bool wait(std::chrono::milliseconds millsec)

	{

		std::unique_lock<std::mutex> lock(_mutexConditPut);

		_conditPut.wait_for(lock, millsec);

		return true;

	}

private:

	//就绪的任务

	std::mutex _mutexQueue;

	std::deque<std::shared_ptr<T>> _queue;

	//条件变量

	std::mutex _mutexConditPut;

	std::condition_variable _conditPut;

	//运行的任务

	std::mutex _mutexDoingTask;

	std::unordered_map<size_t, std::shared_ptr<T> > _mapDoingTask;

};

ThreadPool.h

#pragma once

#include <atomic>

#include <memory>

#include <mutex>

#include <iostream>

#include <thread>

#include "Task.h"

#include "TaskQueue.h"

class ThreadPool

{

public:

    // 线程池配置参数

    typedef struct tagThreadPoolConfig {

        int nMaxThreadsNum; // 最大线程数量

        int nMinThreadsNum; // 最小线程数量

        double dbTaskAddThreadRate; // 增 最大线程任务比 (任务数量与线程数量，什么比例的时候才加)

        double dbTaskSubThreadRate; // 减 最小线程任务比 (任务数量与线程数量，什么比例的时候才减)

    } ThreadPoolConfig;

public:

	//构造函数

    ThreadPool(void):_taskQueue(new TaskQueue<Task>()), _atcCurTotalThrNum(0), _atcWorking(true){}

	//析构函数

	~ThreadPool(void)

    {

        release();

    }

    //初始化资源

    int init(const ThreadPoolConfig& threadPoolConfig) {

        // 错误的设置

        if (threadPoolConfig.dbTaskAddThreadRate < threadPoolConfig.dbTaskSubThreadRate)

            return 87;

        _threadPoolConfig.nMaxThreadsNum = threadPoolConfig.nMaxThreadsNum;

        _threadPoolConfig.nMinThreadsNum = threadPoolConfig.nMinThreadsNum;

        _threadPoolConfig.dbTaskAddThreadRate = threadPoolConfig.dbTaskAddThreadRate;

        _threadPoolConfig.dbTaskSubThreadRate = threadPoolConfig.dbTaskSubThreadRate;

        int ret = 0;

        // 创建线程池

        if (_threadPoolConfig.nMinThreadsNum > 0)

            ret = addProThreads(_threadPoolConfig.nMinThreadsNum);

        return ret;

    }

    // 添加任务

    int addTask(std::shared_ptr<Task> taskptr, bool priority=false)

	{

        const double& rate = getThreadTaskRate();

        int ret = 0;

        if (priority)

        {

            if (rate > 1000)

                std::this_thread::sleep_for(std::chrono::milliseconds(1));

            _taskQueue->put_front(taskptr);

        }

        else

        {

            // 检测任务数量

            if (rate > 100) {

                taskptr->onCanceled();

                return 298;

            }

            // 将任务推入队列

            _taskQueue->put_back(taskptr);

        }

        // 检查是否要扩展线程

        if (_atcCurTotalThrNum < _threadPoolConfig.nMaxThreadsNum

            && rate > _threadPoolConfig.dbTaskAddThreadRate)

            ret = addProThreads(1);

        return ret;

    }

    // 删除任务（从就绪队列删除，如果就绪队列没有，则看执行队列有没有，有的话置下取消状态位）

    int deleteTask(size_t nID)

	{

        return _taskQueue->deleteTask(nID);

    }

    // 删除所有任务

    int deleteAllTasks(void)

	{

        return _taskQueue->deleteAllTasks();

    }

    std::shared_ptr<Task> isTaskProcessed(size_t nId)

	{

        return _taskQueue->isTaskProcessed(nId);

    }

    // 释放资源（释放线程池、释放任务队列）

    bool release(void)

	{

        // 1、停止线程池。

    	// 2、清楚就绪队列。

    	// 3、等待执行队列为0

        releaseThreadPool();

        _taskQueue->release();

        int i = 0;

        while (_atcCurTotalThrNum != 0)

        {

            std::this_thread::sleep_for(std::chrono::milliseconds(500));

            // 异常等待

            if (i++ == 10)

                exit(23);

        }

        _atcCurTotalThrNum = 0;

        return true;

    }

    // 获取当前线程任务比

    double getThreadTaskRate(void)

	{

        if (_atcCurTotalThrNum != 0)

            return _taskQueue->size() * 1.0 / _atcCurTotalThrNum;

        return 0;

    }

    // 当前线程是否需要结束

    bool shouldEnd(void)

	{

        bool bFlag = false;

        double dbThreadTaskRate = getThreadTaskRate();

        // 检查线程与任务比率

        if (!_atcWorking || _atcCurTotalThrNum > _threadPoolConfig.nMinThreadsNum

            && dbThreadTaskRate < _threadPoolConfig.dbTaskSubThreadRate)

            bFlag = true;

        return bFlag;

    }

    // 释放线程池

    bool releaseThreadPool(void)

	{

        _threadPoolConfig.nMinThreadsNum = 0;

        _threadPoolConfig.dbTaskSubThreadRate = 0;

        _atcWorking = false;

        return true;

    }

    // 添加指定数量的处理线程

    int addProThreads(int nThreadsNum)

	{

    	try {

            for (; nThreadsNum > 0; --nThreadsNum)

                std::thread(&ThreadPool::taskProcessThread, this).detach();

        }

        catch (...){

            return 155;

        }

        return 0;

    }

    // 任务处理线程函数

    void taskProcessThread(void)

	{

        int nTaskProcRet = 0;

        // 线程增加

        _atcCurTotalThrNum.fetch_add(1);

        std::chrono::milliseconds mills_sleep(500);

        std::shared_ptr<Task> pTask;

        while (_atcWorking)

        {

            // 从任务队列中获取任务

            pTask = _taskQueue->get();  //get会将任务添加到运行任务的map中去

            if (pTask == nullptr)

            {

                if (shouldEnd())

                    break;

                // 进入睡眠池

                _taskQueue->wait(mills_sleep);

                continue;

            }

            // 检测任务取消状态

            if (pTask->isCancelRequired())

                pTask->onCanceled();

            else

                // 处理任务

                pTask->onCompleted(pTask->doWork());

            // 从运行任务队列中移除任务

            _taskQueue->onTaskFinished(pTask->getID());

            // 判断线程是否需要结束

            if (shouldEnd())

                break;

        }

        // 线程个数减一

        _atcCurTotalThrNum.fetch_sub(1);

    }

private:

	std::shared_ptr<TaskQueue<Task> > _taskQueue;	//任务队列

    ThreadPoolConfig _threadPoolConfig; //线程池配置

    std::atomic<bool> _atcWorking;  //线程池是否被要求结束

    std::atomic<int> _atcCurTotalThrNum;    //当前线程个数

};

FunTask.h

#pragma once

#include <functional>

#include "Task.h"

class FuncTask:public Task

{

public:

	FuncTask(std::function<int(void)> f) : _pf(f) {}

	FuncTask(void) : _pf(nullptr){}

	virtual ~FuncTask(){}

	template <typename F,typename... Args>

	void asynBind(F(*f)(Args...), Args... args)

	{

		_pf = std::bind(f, args...);

	}

	virtual int doWork()

	{

		if (_pf == nullptr)

			return 86;

		return _pf();

	}

private:

	typedef std::function<int(void)> pvFunc;

	pvFunc _pf;

};

main.cpp

#pragma once

#include <time.h>

#include <iostream>

#include <memory>

#include <string>

#include "ThreadPool.h"

#include "FuncTask.h"

using namespace std;

int vFunction(void)

{

	std::cout << __FUNCTION__ << std::endl;

	return 0;

}

int counter(int a,int b)

{

	std::cout << a << ":" << b << std::endl;

	return 0;

}

int main()

{

	ThreadPool::ThreadPoolConfig threadPoolConfig;

	threadPoolConfig.nMaxThreadsNum = 100;

	threadPoolConfig.nMinThreadsNum = 5;

	threadPoolConfig.dbTaskAddThreadRate = 3;

	threadPoolConfig.dbTaskSubThreadRate = 0.5;

	clock_t start = clock();

	{

		std::shared_ptr<ThreadPool> threadPool(new ThreadPool);

		threadPool->init(threadPoolConfig);

		int i = 1;

		while (true)

		{

			/*std::shared_ptr<FuncTask> request(new FuncTask(vFunction));

			threadPool->addTask(request);*/

			std::shared_ptr<FuncTask> request(new FuncTask);

			request->asynBind(counter, i++, 1);

			threadPool->addTask(request);

            if (request->getID() == 110000) {

				break;

            }

		}

		threadPool->release();

	}

	clock_t finish = clock();

	std::cout << "duration:" << finish - start << "ms" << std::endl;

	cout << "main:thread" << endl;

	return 0;

}

巴特西

使用 C++11 编写可复用多线程任务池

类的功能

代码

参考

最新文章

热门文章