利用C++如何实现一个阻塞队列详解

2023-07-14 09:05:03 75

阻塞队列是多线程中常用的数据结构，对于实现多线程之间的数据交换、同步等有很大作用。

阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。简而言之，阻塞队列是生产者用来存放元素、消费者获取元素的容器。

考虑下，这样一个多线程模型，程序有一个主线程master和一些worker线程，master线程负责接收到数据，给worker线程分配数据，worker线程取得一个任务后便可以开始工作，如果没有任务便阻塞住，节约cpu资源。

master线程（生产者）：负责往阻塞队列中塞入数据，并唤醒正在阻塞的worker线程。
workder线程（消费者）：负责从阻塞队列中取数据，如果没有数据便阻塞，直到被master线程唤醒。

那么怎样的数据结构比较适合做这样的唤醒呢？显而易见，是条件变量，在c++11中，stl已经引入了线程支持库。

C++11中条件变量

条件变量一般与一个互斥量同时使用，使用时需要先给互斥量上锁，然后条件变量会检测是否满足条件，如果不满足条件便会暂时释放锁，然后阻塞线程。

c++11使用方法主要如下：

#include
#include
//互斥量与条件变量
std::mutexm_mutex;
std::condition_valuem_condition;

//请求信号的一方
std::unique_locklock(mutex);
while(xxx)
{
//这里会先释放lock，
//如果有信号唤醒的话，会重新加锁。
m_condition.wait(lock);
}

//发送消息进行同步的一方
{
std::unique_locklock(mutex);
//唤醒其他正在wait的线程
m_condition.notify_all();
}

用C++11实现阻塞队列

我们使用条件变量包装STL中的queue就可以实现阻塞队列功能，如果有兴趣，甚至可以自己实现一个效率更高的队列数据结构。

我们先假设一下阻塞队列需要如下接口：

push将一个变量塞入队列；
take从队列中取出一个元素；
size查看队列有多少个元素；

template
classBlockingQueue
{
public:
BlockingQueue();
voidpush(T&&value);
Ttake();
size_tsize()const;

private:
//实际使用的数据结构队列
std::queuem_data;

//条件变量
std::mutexm_mutex;
std::condition_variablem_condition;
};

push一个变量时，我们需要先加锁，加锁成功后才可以压入变量，这是为了线程安全。压入变量后，就可以发送信号通知正在阻塞的条件变量。

voidpush(T&&value)
{
//往队列中塞数据前要先加锁
std::unique_locklock(m_mutex);
m_data.push(value);
//唤醒正在阻塞的条件变量
m_condition.notify_all();
}

take一个变量时，就要有些不一样：

先加锁，加锁成功后，如果队列不为空，可以直接取数据，不需要wait；
如果队列为空呢，则wait等待，直到被唤醒；
考虑特殊情况，唤醒后队列依然是空的……

Ttake()
{
std::unique_locklock(m_mutex);
while(m_data.empty())
{
//等待
m_condition.wait(lock);
}
assert(!m_data.empty());
Tvalue(std::move(m_data.front()));
m_data.pop();

returnvalue;
}

总结下，代码如下：

#ifndefBLOCKINGQUEUE_H
#defineBLOCKINGQUEUE_H

#include
#include
#include
#include

template
classBlockingQueue
{
public:
BlockingQueue()
:m_mutex(),
m_condition(),
m_data()
{
}

//禁止拷贝构造
BlockingQueue(BlockingQueue&)=delete;

~BlockingQueue()
{
}

voidpush(T&&value)
{
//往队列中塞数据前要先加锁
std::unique_locklock(m_mutex);
m_data.push(value);
m_condition.notify_all();
}

voidpush(constT&value)
{
std::unique_locklock(m_mutex);
m_data.push(value);
m_condition.notify_all();
}

Ttake()
{
std::unique_locklock(m_mutex);
while(m_data.empty())
{
m_condition.wait(lock);
}
assert(!m_data.empty());
Tvalue(std::move(m_data.front()));
m_data.pop();

returnvalue;
}

size_tsize()const
{
std::unique_locklock(m_mutex);
returnm_data.size();
}
private:
//实际使用的数据结构队列
std::queuem_data;

//条件变量的锁
std::mutexm_mutex;
std::condition_variablem_condition;
};
#endif//BLOCKINGQUEUE_H

实验代码

我们写个简单的程序实验一下，下面程序有一个master线程，5个worker线程，master线程生成一个随机数，求0-随机数的和。

#include
#include
#include
#include

#include

#include
usingnamespacestd;

inttask=100;
BlockingQueueblockingqueue;
std::mutexmutex_cout;

voidworker()
{
intvalue;
thread::idthis_id=this_thread::get_id();
while(true)
{
value=blockingqueue.take();
uint64_tsum=0;
for(inti=0;ilock(mutex_cout);
std::cout<<"workder:"<th_workers;
for(inti=0;i<5;i++)
{
th_workers.emplace_back(thread(worker));
}

th_master.join();
return0;
}

从输出结果可以看出，master线程将任务分配给了正在空闲的worker线程，具体是哪个线程就看操作系统的随机调度了。

master465
worker:3workerline:34sum:20998440
master466
worker:7workerline:34sum:3308878
master467
worker:4workerline:34sum:34598721
master468
worker:6workerline:34sum:1563796
master469
worker:5workerline:34sum:27978940

Reference

条件变量

总结

到此这篇关于利用C++如何实现一个阻塞队列的文章就介绍到这了,更多相关C++实现阻塞队列内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！

利用C++如何实现一个阻塞队列详解

热门推荐

随机推荐