线程池-手写线程池C++11版本(生产者-消费者模型)

这篇具有很好参考价值的文章主要介绍了线程池-手写线程池C++11版本(生产者-消费者模型)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

本项目是基于C++11的线程池。使用了许多C++的新特性,包含不限于模板函数泛型编程、std::future、std::packaged_task、std::bind、std::forward完美转发、std::make_shared智能指针、decltype类型推断、std::unique_lock锁等C++11新特性功能。


本项目有一定的上手难度。推荐参考系列文章

C++11实用技术(一)auto与decltype的使用

C++11实用技术(二)std::function和bind绑定器

C++11实用技术(三)std::future、std::promise、std::packaged_task、async

C++ operator关键字的使用(重载运算符、仿函数、类型转换操作符)

右值引用以及move移动语义和forward 完美转发

C++标准库中的锁lock_guard、unique_lock、shared_lock、scoped_lock、recursive_mutex


代码结构

本项目线程池功能分以下几个函数去实现:

threadpool.init(isize_t num);设置线程的数量
threadpool::get(TaskFuncPtr& task);读取任务队列中的任务
threadpool::run();通过get()读取任务并执行
threadpool.start(); 启动线程池,并通过run()执行任务
threadpool.exec();封装任务到任务队列中
threadpool.waitForAllDone();等待所有任务执行完成
threadpool.stop();分离线程,释放内存

threadpool.init

init的功能是初始化线程池,主要是设置线程的数量到类的成员变量中。

bool ZERO_ThreadPool::init(size_t num)
{
	std::unique_lock<std::mutex> lock(mutex_);

	if (!threads_.empty())
	{
		return false;
	}

	threadNum_ = num;
	return true;
}

threadNum_:保存线程的数量,在init函数中被赋值

此处使用unique_lock或lock_guard的加锁方式都能实现自动加锁和解锁。但是unique_lock可以进行临时解锁和再上锁,而lock_guard不行,特殊情况下还是必须使用unique_lock(用到条件变量的情况)。(lock_guard比较简单,相对来说性能要好一点)

threadpool::get

从任务队列中获取获取任务,这里其实就是我们的消费者模块

bool ZERO_ThreadPool::get(TaskFuncPtr& task)
{
	std::unique_lock<std::mutex> lock(mutex_);
	if (tasks_.empty()) //判断任务是否存在
	{
		//要终止线程池   bTerminate_设置为true,任务队列不为空
		condition_.wait(lock, [this] { return bTerminate_ || !tasks_.empty(); });
	}
	if (bTerminate_)
		return false;
	if (!tasks_.empty())
	{
		task = std::move(tasks_.front());  // 使用了移动语义
		tasks_.pop(); //释放资源,释放一个任务
		return true;
	}
	return false;
}
  • 条件变量condition_.wait(lock, [this] { return bTerminate_ || !tasks_.empty(); });是需要一直等待条件完成才退出。即任务终止,或者任务队列不为空时,就会退出条件变量的阻塞状态,继续执行下面的逻辑。

  • task = std::move(tasks_.front()); 使用了移动语义,将 tasks_.front() 的内容移动到了 task 中。可以减少内容拷贝。移动完之后tasks_.front() 的内容会变为未指定的状态,所以直接pop掉就好了。

threadpool::run

这里是运行我们的任务部分。包括调用get在任务队列中获取任务,以及执行任务。

void ZERO_ThreadPool::run()  // 执行任务的线程
{
	//调用处理部分
	while (!isTerminate()) // 判断是不是要停止
	{
		TaskFuncPtr task;
		bool ok = get(task);        // 读取任务
		if (ok)
		{
			++atomic_;
			try
			{
				if (task->_expireTime != 0 && task->_expireTime < TNOWMS)
				{//如果设置了超时,并且超时了,就需要执行本逻辑
				//超时任务,本代码未实现,有需要可实现在此处
				}
				else
				{
					task->_func();  // 执行任务
				}
			}
			catch (...)
			{
			}
			--atomic_;
			}
		}
	}
}

atomic_:运行一个任务,该参数+1;执行完毕,该参数-1。这里是为了待会停止线程池时判断是否还有运行中的任务(未完成的线程)。

threadpool.start

创建线程,并把线程池存入vector中,后面释放线程池时,好一一释放线程。

bool ZERO_ThreadPool::start()
{
	std::unique_lock<std::mutex> lock(mutex_);
	if (!threads_.empty())
	{
		return false;
	}
	for (size_t i = 0; i < threadNum_; i++)
	{
		threads_.push_back(new thread(&ZERO_ThreadPool::run, this));
	}
	return true;
}

threads_.push_back(new thread(&ZERO_ThreadPool::run, this));创建线程,线程的回调函数为run。

threadpool.exec

exec是将我们的任务存入任务队列中,这段代码是本项目最难的,用了很多C++的新特性。

/*
	template <class F, class... Args>
	它是c++里新增的最强大的特性之一,它对参数进行了高度泛化,它能表示0到任意个数、任意类型的参数
	auto exec(F &&f, Args &&... args) -> std::future<decltype(f(args...))>
	std::future<decltype(f(args...))>:返回future,调用者可以通过future获取返回值
	返回值后置
	*/
	template <class F, class... Args>
	auto exec(int64_t timeoutMs, F&& f, Args&&... args) -> std::future<decltype(f(args...))>//接受一个超时时间 `timeoutMs`,一个可调用对象 `f` 和其它参数 `args...`,并返回一个 `std::future` 对象,该对象可以用于获取任务执行的结果。
	{
		int64_t expireTime = (timeoutMs == 0 ? 0 : TNOWMS + timeoutMs);  // 根据超时时间计算任务的过期时间 `expireTime`,如果超时时间为 0,则任务不会过期。
		//定义返回值类型
		using RetType = decltype(f(args...));  // 使用 `decltype` 推导出 `f(args...)` 的返回值类型,并将其定义为 `RetType`(这里的using和typedef功能一样,就是为一个类型起一个别名)。
		// 封装任务 使用 `std::packaged_task` 将可调用对象 `f` 和其它参数 `args...` 封装成一个可执行的函数,并将其存储在一个 `std::shared_ptr` 对象 `task` 中。
		auto task = std::make_shared<std::packaged_task<RetType()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));

		TaskFuncPtr fPtr = std::make_shared<TaskFunc>(expireTime);  // 封装任务指针,设置过期时间 创建一个 `TaskFunc` 对象,并将任务的过期时间 `expireTime` 传递给它。
		fPtr->_func = [task]() {  // 具体执行的函数 将封装好的任务函数存储在 `TaskFunc` 对象的 `_func` 成员中,该函数会在任务执行时被调用。
			(*task)();
		};

		std::unique_lock<std::mutex> lock(mutex_);
		tasks_.push(fPtr);              // 将任务插入任务队列中
		condition_.notify_one();        // 唤醒阻塞的线程,可以考虑只有任务队列为空的情况再去notify

		return task->get_future();; //返回一个 `std::future` 对象,该对象可以用于获取任务执行的结果。
	}

使用了可变参数模板函数。
tasks_:保存任务的队列
condition_.notify_one():保存一个任务唤醒一个条件变量
std::future : 异步指向某个任务,然后通过future特性去获取任务函数的返回结果。
std::bind:将参数列表和函数绑定,生成一个新的可调用对象
std::packaged_task:将任务和feature绑定在一起的模板,是一种封装对任务的封装。

本函数用到了泛型编程模板函数,输入参数有3个:一个超时时间 timeoutMs,一个可调用对象 f 和参数 args...。采用返回值后置的方式返回一个std::future对象。这里采用返回值后置是为了方便使用decltype(f(args…)推导数据类型。

auto task = std::make_shared<std::packaged_task<RetType()>>(std::bind(std::forward(f), std::forward(args)…));是将我们传进来的任务函数和参数bind成一个对象,这个对象可以看作是一整个函数,其返回值就是RetType 类型,并且没有输入参数。所以用std::packaged_task<RetType()>这样的格式来打包封装。封装好的对象用智能指针(std::make_shared)来管理。

同时还要创建一个TaskFunc的对象,同样用智能指针管理,这个对象包括两项内容,一个就是超时时间,一个就是我们封装好的task对象。
通过TaskFuncPtr fPtr = std::make_shared(expireTime);和fPtr->_func = task {(*task)();};两条代码将这两项传进去。

最后会通过task->get_future()返回我们任务函数执行的结果返回值。

threadpool.waitForAllDone

等待所有任务执行完成。

bool ZERO_ThreadPool::waitForAllDone(int millsecond)
{
	std::unique_lock<std::mutex> lock(mutex_);
	if (tasks_.empty() && atomic_ == 0)
		return true;
	if (millsecond < 0)
	{
		condition_.wait(lock, [this] { return tasks_.empty() && atomic_ == 0; });
		return true;
	}
	else
	{
		return condition_.wait_for(lock, std::chrono::milliseconds(millsecond), [this] { return tasks_.empty() && atomic_ == 0; });
	}
}

使用条件变量来等待任务执行完成。支持超时执行功能。

此处unique_lock的使用是必须的: 条件变量condition_在wait时会进行unlock再进入休眠, lock_guard并无该操作接口

threadpool.stop

终止线程池。会调用waitForAllDone等待所有任务执行完成再终止。

void ZERO_ThreadPool::stop()
{
	{
		std::unique_lock<std::mutex> lock(mutex_);
		bTerminate_ = true;
		condition_.notify_all();
	}
	waitForAllDone();
	for (size_t i = 0; i < threads_.size(); i++)
	{
		if (threads_[i]->joinable())
		{
			threads_[i]->join();
		}
		delete threads_[i];
		threads_[i] = NULL;
	}
	std::unique_lock<std::mutex> lock(mutex_);
	threads_.clear();
}

通过join等线程执行完成后才返回。

主函数调用

class Test
{
public:
	int test(int i) {
		cout << _name << ", i = " << i << endl;
		Sleep(1000);
		return i;
	}
	void setName(string name) {
		_name = name;
	}
	string _name;
};
void test3() // 测试类对象函数的绑定
{
	ZERO_ThreadPool threadpool;
	threadpool.init(2);
	threadpool.start(); // 启动线程池
	Test t1;
	Test t2;
	t1.setName("Test1");
	t2.setName("Test2");
	auto f1 = threadpool.exec(std::bind(&Test::test, &t1, std::placeholders::_1), 10);
	auto f2 = threadpool.exec(std::bind(&Test::test, &t2, std::placeholders::_1), 20);
	cout << "t1 " << f1.get() << endl;
	cout << "t2 " << f2.get() << endl;
	threadpool.stop();
}
int main()
{
	test3(); // 测试类对象函数的绑定
	cout << "main finish!" << endl;
	return 0;
}

运行结果:
线程池-手写线程池C++11版本(生产者-消费者模型),C++进阶,c++,java,开发语言

本项目完整代码下载地址基于C++11的线程池文章来源地址https://www.toymoban.com/news/detail-635683.html

到了这里,关于线程池-手写线程池C++11版本(生产者-消费者模型)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 多线程之生产者消费者

    目的是回顾多线程的几个api 多生产者+多消费者+共享池

    2024年02月07日
    浏览(52)
  • 【Linux】线程安全-生产者消费者模型

    1个线程安全的队列:只要保证先进先出特性的数据结构都可以称为队列 这个队列要保证互斥(就是保证当前只有一个线程对队列进行操作,其他线程不可以同时来操作),还要保证同步,当生产者将队列中填充满了之后要通知消费者来进行消费,消费者消费之后通知生产者

    2024年02月10日
    浏览(46)
  • 线程同步--生产者消费者模型--单例模式线程池

    条件变量是 线程间共享的全局变量 ,线程间可以通过条件变量进行同步控制 条件变量的使用必须依赖于互斥锁以确保线程安全,线程申请了互斥锁后,可以调用特定函数 进入条件变量等待队列(同时释放互斥锁) ,其他线程则可以通过条件变量在特定的条件下唤醒该线程( 唤醒后线

    2024年01月20日
    浏览(46)
  • python爬虫,多线程与生产者消费者模式

    使用队列完成生产者消费者模式 使用类创建多线程提高爬虫速度 通过队列可以让线程之间进行通信 创建继承Thread的类创建线程,run()会在线程start时执行 吃cpu性能

    2024年02月09日
    浏览(38)
  • C#多线程学习(三) 生产者和消费者

    线程学习第一篇: C#多线程学习(一) 多线程的相关概念 线程学习第二篇: C#多线程学习(二) 如何操纵一个线程 前面说过,每个线程都有自己的资源,但是代码区是共享的,即每个线程都可以执行相同的函数。这可能带来的问题就是几个线程同时执行一个函数,导致数据的混

    2023年04月21日
    浏览(41)
  • JavaEE 初阶篇-生产者与消费者模型(线程通信)

    🔥博客主页: 【 小扳_-CSDN博客】 ❤感谢大家点赞👍收藏⭐评论✍   文章目录         1.0 生产者与消费者模型概述         2.0 在生产者与消费者模型中涉及的关键概念         2.1 缓冲区         2.2 生产者         2.3 消费者         2.4 同步机制         2.5 线程间通

    2024年04月28日
    浏览(42)
  • 探究:kafka生产者/消费者与多线程安全

    目录 1. 多线程安全 1.1. 生产者是多线程安全的么? 1.1. 消费者是多线程安全的么? 2. 消费者规避多线程安全方案 2.1. 每个线程维护一个kafkaConsumer 2.2. [单/多]kafkaConsumer实例 + 多worker线程 2.3.方案优缺点对比         Kafka生产者是 线程安全 的,可以在多个线程中共享一个

    2023年04月26日
    浏览(93)
  • 多线程(初阶七:阻塞队列和生产者消费者模型)

    目录 一、阻塞队列的简单介绍 二、生产者消费者模型 1、举个栗子: 2、引入生产者消费者模型的意义: (1)解耦合 (2)削峰填谷 三、模拟实现阻塞队列 1、阻塞队列的简单介绍 2、实现阻塞队列 (1)实现普通队列 (2)加上线程安全 (3)加上阻塞功能 3、运用阻塞队列

    2024年02月05日
    浏览(41)
  • 【linux】线程同步+基于BlockingQueue的生产者消费者模型

    喜欢的点赞,收藏,关注一下把! 在线程互斥写了一份抢票的代码,我们发现虽然加锁解决了抢到负数票的问题,但是一直都是一个线程在抢票,它错了吗,它没错但是不合理。那我们应该如何安全合理的抢票呢? 讲个小故事。 假设学校有一个VIP学霸自习室,这个自习室有

    2024年02月03日
    浏览(100)
  • C# 快速写入日志 不卡线程 生产者 消费者模式

    有这样一种场景需求,就是某个方法,对耗时要求很高,但是又要记录日志到数据库便于分析,由于访问数据库基本都要几十毫秒,可在方法里写入BlockingCollection,由另外的线程写入数据库。 可以看到,在我的机子上面,1ms写入了43条日志。

    2024年02月15日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包