整理自C++併發程式設計實戰

使用std::shared_ptr<>的執行緒安全佇列

/*
 * threadsafe_queue_ptr.cpp
 *
 *  Created on: Mar 2, 2018
 *      Author: [email protected]
 *      為了防止在wait_and_pop()中引發異常，
 *      所以將shared_ptr<>的初始化移動到push()
 *      呼叫，並且儲存shared_ptr<>的例項而不是
 *      儲存值。將內部的queue複製到shared_ptr<>
 *      並不會引發異常。
 */

#include <iostream>
#include <queue>
#include <mutex>
#include <memory>
#include <condition_variable>

using namespace std;

template<typename T>
class threadsafe_queue
{
public:
	threadsafe_queue(){}
	void wait_and_pop(T& value)
	{
		//使用了條件變數之後如果佇列中沒有元素會等待佇列push之後再pop，這裡需要操作佇列所以解鎖互斥鎖
		unique_lock<mutex> lk(_mut);
		cout<<"解鎖_mut"<<endl;
		//呼叫等待函式時使用lambda判斷佇列中是否為空
		_data_cond.wait(lk,[this]{return !_data_queue.empty();});
		value = std::move(*_data_queue.front());
		_data_queue.pop();
	}

	bool try_pop(T& value)
	{
		//為了防止競爭先加鎖佇列
		lock_guard<mutex> lk(_mut);
		cout<<"加鎖_mut"<<endl;
		//如果
		if(_data_queue.empty())
			return false;
		value = move(*_data_queue.front());
		cout<<"pop:"<<value<<endl;
		_data_queue.pop();
		return true;
	}

	shared_ptr<T> wait_and_pop()
	{
		unique_lock<mutex> lk(_mut);
		cout<<"等待pop"<<endl;
		//在pop之前確認佇列中是否有元素，否則不解鎖
		_data_cond.wait(lk,[this]{return !_data_queue.empty();});
		cout<<"開始pop"<<endl;
		shared_ptr<T> res = _data_queue.front();
		_data_queue.pop();
		return res;
	}

	shared_ptr<T> try_pop()
	{
		lock_guard<mutex> lk(_mut);
		if(_data_queue.empty())
			return shared_ptr<T>();
		shared_ptr<T> res = _data_queue.front();
		_data_queue.pop();
		return res;
	}

	void push(T new_value)
	{
		shared_ptr<T> data(
				make_shared<T>(move(new_value)));
		lock_guard<mutex> lk(_mut);
		_data_queue.push(data);
		//喚醒一個執行緒供給wait_and_pop使用
		_data_cond.notify_one();
	}

	bool empty() const
	{
		lock_guard<mutex> lk(_mut);
		return _data_queue.empty();
	}
private:
	//互斥鎖變數
	mutable mutex _mut;
	//使用智慧指標儲存的佇列型別
	queue<shared_ptr<T> > _data_queue;
	//條件變數
	condition_variable _data_cond;
};

int main()
{
	threadsafe_queue<int> my_queue;
	my_queue.push(1);
	my_queue.push(2);
	my_queue.push(3);
	my_queue.push(4);
	my_queue.push(5);
	my_queue.push(6);
	my_queue.push(7);
	int a = 1;
	my_queue.try_pop(a);
	shared_ptr<int> i = my_queue.try_pop();
	cout<<"try_pop pop出的數是:"<<*i<<endl;
	i = my_queue.wait_and_pop();
	cout<<"wait_and_pop pop出的數是:"<<*i<<endl;
	return 0;
}
 

執行結果：

[email protected]:~/testcode/併發$ g++ threadsafe_queue_ptr.cpp -pthread -std=c++11
[email protected]:~/testcode/併發$ ./a.out 
加鎖_mut
pop:1
try_pop pop出的數是:2
等待pop
開始pop
wait_and_pop pop出的數是:3