时间:2021-05-19
阻塞队列是多线程中常用的数据结构,对于实现多线程之间的数据交换、同步等有很大作用。
阻塞队列常用于生产者和消费者的场景,生产者是向队列里添加元素的线程,消费者是从队列里取元素的线程。简而言之,阻塞队列是生产者用来存放元素、消费者获取元素的容器。
考虑下,这样一个多线程模型,程序有一个主线程 master 和一些 worker 线程,master 线程负责接收到数据,给 worker 线程分配数据,worker 线程取得一个任务后便可以开始工作,如果没有任务便阻塞住,节约 cpu 资源。
那么怎样的数据结构比较适合做这样的唤醒呢?显而易见,是条件变量,在 c++ 11 中,stl 已经引入了线程支持库。
C++11 中条件变量
条件变量一般与一个 互斥量 同时使用,使用时需要先给互斥量上锁,然后条件变量会检测是否满足条件,如果不满足条件便会暂时释放锁,然后阻塞线程。
c++ 11使用方法主要如下:
#include <mutex>#include <condition_value>// 互斥量与条件变量std::mutex m_mutex;std::condition_value m_condition;// 请求信号的一方std::unique_lock<std::mutex> lock(mutex);while(xxx){ // 这里会先释放 lock, // 如果有信号唤醒的话,会重新加锁。 m_condition.wait(lock);}// 发送消息进行同步的一方{ std::unique_lock<std::mutex> lock(mutex); // 唤醒其他正在 wait 的线程 m_condition.notify_all();}用 C++11 实现阻塞队列
我们使用条件变量包装 STL 中的 queue 就可以实现阻塞队列功能,如果有兴趣,甚至可以自己实现一个效率更高的队列数据结构。
我们先假设一下阻塞队列需要如下接口:
push 一个变量时,我们需要先加锁,加锁成功后才可以压入变量,这是为了线程安全。压入变量后,就可以发送信号通知正在阻塞的条件变量。
void push(T&& value) { // 往队列中塞数据前要先加锁 std::unique_lock<std::mutex> lock(m_mutex); m_data.push(value); // 唤醒正在阻塞的条件变量 m_condition.notify_all(); }take 一个变量时,就要有些不一样:
总结下,代码如下:
#ifndef BLOCKINGQUEUE_H#define BLOCKINGQUEUE_H#include <queue>#include <mutex>#include <condition_variable>#include <assert.h>template <typename T>class BlockingQueue{public: BlockingQueue() :m_mutex(), m_condition(), m_data() { } // 禁止拷贝构造 BlockingQueue(BlockingQueue&) = delete; ~BlockingQueue() { } void push(T&& value) { // 往队列中塞数据前要先加锁 std::unique_lock<std::mutex> lock(m_mutex); m_data.push(value); m_condition.notify_all(); } void push(const T& value) { std::unique_lock<std::mutex> lock(m_mutex); m_data.push(value); m_condition.notify_all(); } T take() { std::unique_lock<std::mutex> lock(m_mutex); while(m_data.empty()) { m_condition.wait(lock); } assert(!m_data.empty()); T value(std::move(m_data.front())); m_data.pop(); return value; } size_t size() const { std::unique_lock<std::mutex> lock(m_mutex); return m_data.size(); }private: // 实际使用的数据结构队列 std::queue<T> m_data; // 条件变量的锁 std::mutex m_mutex; std::condition_variable m_condition;};#endif // BLOCKINGQUEUE_H实验代码
我们写个简单的程序实验一下,下面程序有 一个 master 线程,5个 worker 线程,master线程生成一个随机数,求 0-随机数 的和。
#include <iostream>#include <thread>#include <mutex>#include <random>#include <windows.h>#include <blockingqueue.h>using namespace std;int task=100;BlockingQueue<int> blockingqueue;std::mutex mutex_cout;void worker(){ int value; thread::id this_id = this_thread::get_id(); while(true) { value = blockingqueue.take(); uint64_t sum = 0; for(int i = 0; i < value; i++) { sum += i; } // 模拟耗时操作 Sleep(100); std::lock_guard<mutex> lock(mutex_cout); std::cout << "workder: " << this_id << " " << __FUNCTION__ << " line: " << __LINE__ << " sum: " << sum << std::endl; }}void master(){ srand(time(nullptr)); for(int i = 0; i < task; i++) { blockingqueue.push(rand()%10000); printf("%s %d %i\n",__FUNCTION__, __LINE__, i); Sleep(20); }}int main(){ thread th_master(master); std::vector<thread> th_workers; for(int i =0; i < 5; i++) { th_workers.emplace_back(thread(worker)); } th_master.join(); return 0;}从输出结果可以看出,master 线程将任务分配给了正在空闲的 worker 线程,具体是哪个线程就看操作系统的随机调度了。
master 46 5
worker: 3 worker line: 34 sum: 20998440
master 46 6
worker: 7 worker line: 34 sum: 3308878
master 46 7
worker: 4 worker line: 34 sum: 34598721
master 46 8
worker: 6 worker line: 34 sum: 1563796
master 46 9
worker: 5 worker line: 34 sum: 27978940
Reference
条件变量
总结
到此这篇关于利用C++如何实现一个阻塞队列的文章就介绍到这了,更多相关C++实现阻塞队列内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
c++链表实现的阻塞队列最近从java源码里发现了阻塞队列的实现,觉得非常有趣。首先,介绍下什么是阻塞队列。阻塞队列代表着一个队列可以线程安全的往该队列中写数据
阻塞队列简介阻塞队列(BlockingQueue)首先是一个支持先进先出的队列,与普通的队列完全相同;其次是一个支持阻塞操作的队列,即:当队列满时,会阻塞执行插
C++中"priority_queue"优先级队列实例详解1.简介标准库队列使用了先进先出(FIFO)的存储和检索策略.进入队列的对象被放置在尾部,下一个被取出
今天重温了下java多线程中的notify()方法以及wait()方法,一时兴起,决定通过这俩个方法,实现一个简易的自定义阻塞队列。阻塞队列是什么,与普通队列的
C++数据结构实现两个栈实现一个队列栈为后进先出,队列为先进先出用两个栈实现一个队列。是一个比较经典的问题。看到这个问题,我的第一个解题思路为:定义两个栈,s1