快捷導(dǎo)航

C++線程間的互斥和通信場景分析

更新時(shí)間：2021年05月21日 10:31:41 作者：Redamanc

很多朋友對C++線程間的互斥和通信知識(shí)掌握不是多牢靠，今天小編通過模擬車站賣票應(yīng)用場景給大家詳細(xì)解析C++線程間的互斥和通信知識(shí)，感興趣的朋友跟隨小編一起看看吧

互斥鎖（mutex）

為了更好地理解，互斥鎖，我們可以首先來看這么一個(gè)應(yīng)用場景：模擬車站賣票。

模擬車站賣票

場景說明：
Yang車站售賣從亞特蘭蒂斯到古巴比倫的時(shí)光飛船票；因?yàn)闄C(jī)會(huì)難得，所以票數(shù)有限，一經(jīng)發(fā)售，謝絕補(bǔ)票。
飛船票總數(shù)：100張；
售賣窗口：3個(gè)。
對于珍貴的飛船票來說，這個(gè)資源是互斥的，比如第100張票，只能賣給一個(gè)人，不可能同時(shí)賣給兩個(gè)人。3個(gè)窗口都有權(quán)限去售賣飛船票（唯一合法途徑）。

不加鎖的結(jié)果

根據(jù)場景說明，我們可以很快地分析如下：
可以使用三個(gè)線程來模擬三個(gè)獨(dú)立的窗口同時(shí)進(jìn)行賣票；
定義一個(gè)全局變量，每當(dāng)一個(gè)窗口賣出一張票，就對這個(gè)變量進(jìn)行減減操作。
故寫出如下代碼：

#include <iostream>
#include <thread>
#include <list>
using namespace std;

int tickets = 100; // 車站剩余票數(shù)總數(shù)

void sellTickets(int win)
{
	while (tickets > 0)
	{
		{
			if (tickets > 0)
			{
				cout << "窗口：" << win << " 賣出了第：" << tickets << "張票!" << endl;
				tickets--;
			}
			std::this_thread::sleep_for(std::chrono::microseconds(400));
		}
	}
}

int main()
{
	list<std::thread> tlist;
	for (int i = 1; i <= 3; ++i)
	{
		tlist.push_back(std::thread(sellTickets, i));
	}
	for (std::thread& t : tlist)
	{
		t.join();
	}
	cout << "所有窗口賣票結(jié)束！" << endl;

	return 0;
}

運(yùn)行結(jié)果如下：

在這里插入圖片描述

通過運(yùn)行，我們可以發(fā)現(xiàn)問題：
對于一張票來說，賣出去了多次！
這不白嫖嗎？？？這合適嗎？
原因也很簡單，對于線程來說，誰先執(zhí)行，誰后執(zhí)行，完全是根據(jù)CPU的調(diào)度，根本不可能掌握清楚。
所以，這個(gè)代碼是線程不安全的！
那，怎么解決呢？
當(dāng)然是：互斥鎖了！

加鎖后的結(jié)果

我們對上述代碼做出如下修改：

#include <iostream>
#include <thread>
#include <list>
#include <mutex>
using namespace std;

int tickets = 100; 
std::mutex mtx;

void sellTickets(int win)
{
	while (tickets > 0)
	{
		{
			lock_guard<std::mutex> lock(mtx);
			if (tickets > 0)
			{
				cout << "窗口：" << win << " 賣出了第：" << tickets << "張票!" << endl;
				tickets--;
			}
			std::this_thread::sleep_for(std::chrono::microseconds(400));
		}
	}
}

int main()
{
	list<std::thread> tlist;
	for (int i = 1; i <= 3; ++i)
	{
		tlist.push_back(std::thread(sellTickets, i));
	}
	for (std::thread& t : tlist)
	{
		t.join();
	}
	cout << "所有窗口賣票結(jié)束！" << endl;

	return 0;
}

首先定義了一個(gè)全局的互斥鎖std::mutex mtx；接著在對票數(shù)tickets進(jìn)行減減操作時(shí)，定義了lock_guard，這個(gè)就相當(dāng)于智能指針scoped_ptr一樣，可以出了作用域自動(dòng)釋放鎖資源。
運(yùn)行結(jié)果如下：

在這里插入圖片描述

我們可以看到這一次，就沒問題了。

簡單總結(jié)

互斥鎖的使用可以有三種：
（首先都需要在全局定義互斥鎖std::mutex mtx）

首先可以直接在需要加鎖和解鎖的地方，手動(dòng)進(jìn)行：加鎖mtx.lock()、解鎖mtx.unlock()；
可以在需要加鎖的地方定義保護(hù)鎖：lock_guard<std::mutex> lock(mtx)，這個(gè)鎖在定義的時(shí)候自動(dòng)上鎖，出了作用域自動(dòng)解鎖。（其實(shí)就是借助了智能指針的思想，定義對象出調(diào)用構(gòu)造函數(shù)底層調(diào)用lock()，出了作用域調(diào)用析構(gòu)函數(shù)底層調(diào)用unlock()）；
可以在需要加鎖的地方定義唯一鎖：unique_lock<std::mutex> lock(mtx)，這個(gè)鎖和保護(hù)鎖類似，但是比保護(hù)鎖更加好用。（可以類比智能指針中的scoped_ptr和unique_ptr的區(qū)別，二者都是將拷貝構(gòu)造和賦值重載函數(shù)刪除了，但是unique_ptr和unique_lock都定義了帶有右值引用的拷貝構(gòu)造和賦值）

條件變量（conditon_variable）

如果說，互斥鎖是為了解決線程間互斥的問題，那么，條件變量就是為了解決線程間通信的問題。
同樣的，我們可以首先來看一個(gè)問題（模型）：

生產(chǎn)者消費(fèi)者線程模型

生產(chǎn)者消費(fèi)者線程模型是一個(gè)很經(jīng)典的線程模型；
首先會(huì)有兩個(gè)線程，一個(gè)是生產(chǎn)者，一個(gè)是消費(fèi)者，生產(chǎn)者只負(fù)責(zé)生產(chǎn)資源，消費(fèi)者只負(fù)責(zé)消費(fèi)資源。

產(chǎn)生問題

根據(jù)上述互斥鎖的理解，我們可以寫出如下代碼：

#include <iostream>
#include <thread>
#include <mutex>
#include <queue>
using namespace std;

std::mutex mtx; 

class Queue
{
public:
	void put(int num)
	{
		lock_guard<std::mutex> lock(mtx);
		que.push(num);
		cout << "生產(chǎn)者，生產(chǎn)了：" << num << "號(hào)產(chǎn)品" << endl;
	}
	void get()
	{
		lock_guard<std::mutex> lock(mtx);
		int val = que.front();
		que.pop();
		cout << "消費(fèi)者，消費(fèi)了：" << val << "號(hào)產(chǎn)品" << endl;
	}
private:
	queue<int> que;
};

void producer(Queue* que)
{
	for (int i = 0; i < 10; ++i)
	{
		que->put(i);
		std::this_thread::sleep_for(std::chrono::milliseconds(200));
	}
}

void consumer(Queue* que)
{
	for (int i = 0; i < 10; ++i)
	{
		que->get();
		std::this_thread::sleep_for(std::chrono::milliseconds(200));
	}
}
int main()
{
	Queue que;

	std::thread t1(producer, &que);
	std::thread t2(consumer, &que);

	t1.join();
	t2.join();

	return 0;
}

同樣的，我們定義了兩個(gè)線程：t1、t2分別作為生產(chǎn)者和消費(fèi)者，并且定義了兩個(gè)線程函數(shù)：producer和consumer，這兩個(gè)函數(shù)接受一個(gè)Queue*的參數(shù)，并且通過這個(gè)指針調(diào)用put和get方法，這兩個(gè)方法就是往資源隊(duì)列里面執(zhí)行入隊(duì)和出隊(duì)操作。
運(yùn)行結(jié)果如下：

在這里插入圖片描述

我們會(huì)發(fā)現(xiàn)，出錯(cuò)了。
多運(yùn)行幾次試試：

在這里插入圖片描述

我們發(fā)現(xiàn)，每次運(yùn)行的結(jié)果還都不一樣，但是都會(huì)出現(xiàn)系統(tǒng)崩潰的問題。
仔細(xì)來看這個(gè)錯(cuò)誤原因：

在這里插入圖片描述

我們再想想這個(gè)代碼的邏輯：
一個(gè)生產(chǎn)者只負(fù)責(zé)生產(chǎn)；
一個(gè)消費(fèi)者只負(fù)責(zé)消費(fèi)；
他們共同在隊(duì)列里面存取資源；
存取資源操作本身是互斥的。
發(fā)現(xiàn)問題了嗎？
這兩個(gè)線程之間彼此的操作獨(dú)立，換句話說，
沒有通信！
生產(chǎn)者生產(chǎn)的時(shí)候，消費(fèi)者不知道；
消費(fèi)者消費(fèi)的時(shí)候，生產(chǎn)者也不知道；
但是消費(fèi)者是要從隊(duì)列里面取資源的，如果某一個(gè)時(shí)刻，隊(duì)列里為空了，它就不能取了！

解決問題

分析完問題之后，我們知道了：
問題出在：沒有通信上面。
那么如何解決通信問題呢？
當(dāng)然就是：條件變量了！
我們做出如下代碼的修改：

#include <iostream>
#include <thread>
#include <mutex>
#include <queue>
#include <condition_variable>
using namespace std;

std::mutex mtx; // 互斥鎖，用于線程間互斥
std::condition_variable cv;// 條件變量，用于線程間通信

class Queue
{
public:
	void put(int num)
	{
		unique_lock<std::mutex> lck(mtx);
		while (!que.empty())
		{
			cv.wait(lck);
		}
		que.push(num);
		cv.notify_all();
		cout << "生產(chǎn)者，生產(chǎn)了：" << num << "號(hào)產(chǎn)品" << endl;
	}
	void get()
	{
		unique_lock<std::mutex> lck(mtx);
		while (que.empty())
		{
			cv.wait(lck);
		}
		int val = que.front();
		que.pop();
		cv.notify_all();
		cout << "消費(fèi)者，消費(fèi)了：" << val << "號(hào)產(chǎn)品" << endl;
	}
private:
	queue<int> que;
};

void producer(Queue* que)
{
	for (int i = 0; i < 10; ++i)
	{
		que->put(i);
		std::this_thread::sleep_for(std::chrono::milliseconds(200));
	}
}

void consumer(Queue* que)
{
	for (int i = 0; i < 10; ++i)
	{
		que->get();
		std::this_thread::sleep_for(std::chrono::milliseconds(200));
	}
}
int main()
{
	Queue que;

	std::thread t1(producer, &que);
	std::thread t2(consumer, &que);

	t1.join();
	t2.join();

	return 0;
}

這個(gè)時(shí)候我們再來看運(yùn)行結(jié)果：

在這里插入圖片描述

這個(gè)時(shí)候就是：
生產(chǎn)一個(gè)、消費(fèi)一個(gè)。

原子類型（atomic）

我們前面遇到線程不安全的問題，主要是因?yàn)樯婕?code>++、--操作的時(shí)候，有可能被其他的線程干擾，所以使用了互斥鎖。
只允許得到鎖的線程進(jìn)行操作；
其他沒有得到鎖的線程只能眼巴巴的干看著。
但是，對于互斥鎖來說，它是比較重的，它對于臨界區(qū)代碼做的事情比較復(fù)雜。
簡單來說，如果只是為了++、--這樣的簡單操作互斥的話，使用互斥鎖，就有點(diǎn)殺雞用牛刀的意味了。
那么有沒有比互斥鎖更加輕量的，并且能夠解決問題的呢？
當(dāng)然有，就是我們要說的原子類型。

簡單使用

我們可以簡單設(shè)置一個(gè)場景：
定義十個(gè)線程，對一個(gè)公有的變量myCount進(jìn)行task的操作，該操作是對變量進(jìn)行100次的++。
所以，如果順利，我們會(huì)最終得到myCount = 1000。
代碼如下：

#include <iostream>
#include <thread>
#include <atomic>
#include <list>

volatile std::atomic_bool isReady = false;
volatile std::atomic_int myCount = 0;

void task()
{
	while (!isReady)
	{
		// 線程讓出當(dāng)前的CPU時(shí)間片，等待下一次調(diào)度
		std::this_thread::yield();
	}

	for (int i = 0; i < 100; ++i)
	{
		myCount++;
	}
}

int main()
{
	std::list<std::thread> tlist;
	for (int i = 0; i < 10; ++i)
	{
		tlist.push_back(std::thread(task));
	}

	std::this_thread::sleep_for(std::chrono::milliseconds(200));
	isReady = true;

	for (std::thread& it : tlist)
	{
		it.join();
	}
	std::cout << "myCount:" << myCount << std::endl;

	return 0;
}

運(yùn)行結(jié)果如下：

在這里插入圖片描述

改良車站賣票

對于原子類型來說，使用方法非常簡單：
首先包含頭文件：#include <atomic>；
接著把需要原子操作的變量定義為對應(yīng)的原子類型就好：
bool -> atomic_bool;
int -> atomic_int;
其他同理。
理解了這個(gè)以后，我們可以使用原子類型對我們的車站賣票進(jìn)行改良：

#include <iostream>
#include <thread>
#include <list>
#include <mutex>
#include <atomic>
using namespace std;

std::atomic_int tickets = 100; // 車站剩余票數(shù)總數(shù)

void sellTickets(int win)
{
	while (tickets > 0)
	{
		tickets--;
		cout << "窗口：" << win << " 賣出了第：" << tickets << "張票!" << endl;
	}
}

int main()
{
	list<std::thread> tlist;
	for (int i = 1; i <= 3; ++i)
	{
		tlist.push_back(std::thread(sellTickets, i));
	}
	for (std::thread& t : tlist)
	{
		t.join();
	}
	cout << "所有窗口賣票結(jié)束！" << endl;

	return 0;
}

可以看到，從代碼長度來說就輕量了很多！
運(yùn)行結(jié)果如下：

在這里插入圖片描述