快樂在于態(tài)度，成功在于細(xì)節(jié)，命運(yùn)在于習(xí)慣。
Happiness lies in the attitude, success lies in details, fate is a habit.

如果共享數(shù)據(jù)都是只讀數(shù)據(jù)，就不會(huì)有問題；但是，同時(shí)一旦有刪除刪除數(shù)據(jù)就會(huì)出現(xiàn)問題。

在并發(fā)編程中，操作由兩個(gè)或者多個(gè)線程負(fù)責(zé)，它們爭(zhēng)先讓線程執(zhí)行各自的操作，而結(jié)果取決于它們執(zhí)行的相對(duì)次序，所有這種情況都是條件競(jìng)爭(zhēng)。

1、但是如何防止惡性的條件競(jìng)爭(zhēng)呢？

• 有鎖數(shù)據(jù)結(jié)構(gòu)：采取保護(hù)措施包裝數(shù)據(jù)結(jié)構(gòu)，確保不變量被破壞時(shí)，中間狀態(tài)只對(duì)執(zhí)行改動(dòng)線程可見。在其他訪問同一數(shù)據(jù)結(jié)構(gòu)的視角中，這種改動(dòng)要么尚未開始，要么已經(jīng)完成。
• 無鎖數(shù)據(jù)結(jié)構(gòu)：修改數(shù)據(jù)結(jié)構(gòu)的設(shè)計(jì)及其不變量，由一連串不可拆分的改動(dòng)完成數(shù)據(jù)變更，每個(gè)改動(dòng)都維持?jǐn)?shù)據(jù)變量不被破壞。通常這種編程難以正確編寫。

保護(hù)共享數(shù)據(jù)的最基本方式就是互斥。

2、如何用互斥保護(hù)共享數(shù)據(jù)？

訪問一個(gè)數(shù)據(jù)結(jié)構(gòu)前，先鎖住與數(shù)據(jù)相關(guān)的互斥；訪問結(jié)束后，再解鎖互斥。C++線程庫保證了，一旦有線程鎖住了某個(gè)互斥，若其他線程試圖再給它加鎖，則須等待，直至最初成功加鎖的線程把該互斥解鎖。這確保了全部線程所見到的共享數(shù)據(jù)是正確的，不變量沒有沒有被破壞。

2.1 如何使用互斥？

通過 std::mutex的實(shí)例創(chuàng)建互斥，調(diào)用成員函數(shù)lock()對(duì)其加鎖，調(diào)用unlock()解鎖。但是一般不推薦直接調(diào)用成員函數(shù)的做法。原因是若按此處理，那就必須記住，在函數(shù)以外的每條代碼路徑上都要調(diào)用unlock()，包括由于異常導(dǎo)致退出的路徑。

C++標(biāo)準(zhǔn)庫提供了類模版std::lock_guard<>,針對(duì)互斥融合實(shí)現(xiàn)了RAII機(jī)制：在構(gòu)造是給互斥加鎖，在析構(gòu)時(shí)給互斥解鎖，從而保證互斥總被正確的解鎖。

如下，用互斥保護(hù)鏈表：

#include <list>
#include <mutex>
#include <algorithm>
std::list<int> some_list;
void add_to_list(int new_value)
{
	std::lock_guard<std::mutex> guard(some_mutex);
	some_list.push_back(new_value);
}
bool list_contains(int value_to_find)
{
	std::lock_gurd<std::mutex> guard(some_mutex);
	return std::find(some_list.begin(), some_list.end(), value_to_find) != some_list.end();
}

2.2 本意互斥保護(hù)數(shù)據(jù)卻留有余地，如何防止隱患呢？

如果成員函數(shù)返回的指針或者引用，指向受保護(hù)的共享數(shù)據(jù)，那么即便成員函數(shù)全都良好、有序的方式鎖定互斥，仍然無濟(jì)于事，因?yàn)槭鼙Ｗo(hù)已被打破，出現(xiàn)大漏洞。 只要存在任何能訪問該指針和引用的代碼，它就可以訪問受保護(hù)的共享數(shù)據(jù)，則須謹(jǐn)慎設(shè)計(jì)程序接口，從而確?；コ庀刃墟i定，再對(duì)受保護(hù)的共享數(shù)據(jù)進(jìn)行訪問，并保證不留后門。

我們來看看，意外的向外傳遞引用，指向受保護(hù)的共享數(shù)據(jù)：

class some_data
{
	int a;
	std::string b;
public:
	void do_something();
};
class data_wrapper
{
private:
	some_data data;
	std::mutex m;
public:
	template<typename Fubction>
	void process_data(Function func)
	{
		std::lock_guard<std::mutex> l(m);
		//向使用者提供的函數(shù)傳遞受保護(hù)的數(shù)據(jù)
		func(data);
	}
}；
some_data* unprotected;
void malicious_function(some_data& protected_data)
{
	unprotected = &protected_data;
}
data_wrapper x;
void foo()
{
	//傳入惡意函數(shù)
	x.process_data(malicious_function);
	//以無保護(hù)方式訪問本應(yīng)受保護(hù)的共享數(shù)據(jù)
	unprotected->do_something();
}

我們除了要檢查成員函數(shù)，防止向調(diào)用者傳出指針或引用，還必須檢查另一種情況：若成員函數(shù)在自身內(nèi)部調(diào)用了別的函數(shù)，而這些函數(shù)卻不受我們掌控，那么，也不得向他們傳遞這些指針或引用。如果有就很危險(xiǎn)。

2.3 如何解決容器本身接口固有的條件競(jìng)爭(zhēng)？

多線程訪問的情況下STL容器內(nèi)的empty()和size()的結(jié)果不可信。盡管，在某個(gè)線程調(diào)用empty()或size()時(shí)，返回值可能是正確的。然而，一旦函數(shù)返回，其他線程就不再受限，從而能自由地訪問棧容器，可能馬上有新元素入棧，或者，現(xiàn)有的元素會(huì)立刻出棧，令前面的線程得到結(jié)果失效而無法使用。

在空棧上調(diào)用top()會(huì)導(dǎo)致未定義行為。

但是如何解決上述問題？

• 傳入引用；
• 提供不拋出異常的拷貝構(gòu)造函數(shù)或不拋出異常的移動(dòng)構(gòu)造函數(shù)；
• 返回指針指向彈出的元素。

如下代碼，線程安全的棧容器類：

#include <exception>
#include <memory>
#include <mutex>
#include <stack>
struct empty_stack: std::exception
{
	const char* what() const throw();
};
template<typename T>
class threadsafe_stack
{
private:
	std::statck<T> data;
	mutable std::mutex m;
public:
	threadsafe_stack(){}
	threadsafe_stack(const threadsafe_stack& other)
	{
		std::lock_guard<std::metux> lock(other.m);
		//在構(gòu)造函數(shù)的函數(shù)體（constructor body）內(nèi)進(jìn)行復(fù)制操作
		data = other.data;
	}
	//將賦值運(yùn)算符刪除
	threadsafe_stack& operator=(const threadsafe_stack&) = delete;
	void push(T new_value)
	{
		std::lock_guard<std::mutex> lock(m);
		data.push(std::move(new_value));
	}
	//返回std::share_ptr<T>
	std::share_ptr<T> pop()
	{
		std::lock_guard<std::mutex> lock(m);
		//試圖在彈出前檢查是否為空棧
		if(data.empty()) throw empty_stack();
		//改動(dòng)棧容器前設(shè)置返回值
		std::share_ptr<T> const res(std::make_shared<T>(data.top()));
		data.pop();
		return res;
	}
	//接收引用參數(shù)，指向某外部變量的地址，存儲(chǔ)彈出的值
	void pop(T& value)
	{
		std::lock_guard<std::mutex> lock(m);
		if(data.empty()) throw empty_stack();
		value = data.top();
		data.pop();
	}
	bool empty() const
	{
		std::lock_guard<std::mutex> lock(m);
		return data.empty();
	}
};

2.4 如何解決死鎖問題？

線程在互斥上爭(zhēng)奪搶鎖：有兩個(gè)線程，都需同時(shí)鎖住兩個(gè)互斥，都等待著再給另一個(gè)互斥加鎖。于是，雙方毫無進(jìn)展，因?yàn)樗鼈兺瑫r(shí)在苦苦等待對(duì)方解鎖互斥。此時(shí)造成死鎖。
死鎖：兩個(gè)線程互相等待，停滯不前。

防范死鎖的建議通常是始終按相同的順序?qū)蓚€(gè)互斥加鎖。若我們總是先鎖互斥A，在鎖互斥B，則永遠(yuǎn)不會(huì)發(fā)生死鎖。

（1）運(yùn)用std::lock函數(shù)和std::lock_guard類模版，進(jìn)行內(nèi)部數(shù)據(jù)的互換操作：

class some_big_object;
void swap(some_big_object& lhs, some_big_object& rhs);
class X
{
private:
	some_big_object some_detail;
	std::mutex m;
public:
	X(some_big_object const& sd):some_detail(sd){}
	friend void swap(X& lhs, X& rhs)
	{
		if(&lhs == &rhs)
			return;
		std::lock(lhs.m, rhs.m);
		//std::adopt_lock指明互斥上已被鎖住，即互斥上有鎖存在
		std::lock_guard<std::mutex> lock_a(lhs.m, std::adopt_lock);
		std::lock_guard<std::mutex> lock_a(rhs.m, std::adopt_lock);
		swap(lhs.some_detail, rhs.some_detail);
    }
}

std::adopt_lock指明互斥上已被鎖住，即互斥上有鎖存在。

（2）使用std::unique_lock鎖，如下代碼：

//實(shí)例std::defer_lock將互斥保留為無鎖狀態(tài)
std::unique_lock<std::mutex> lock_a(lhs.m, std::defer_lock);
std::unique_lock<std::mutex> lock_b(lhs.m, std::defer_lock);
//到這里才對(duì)互斥加鎖
std::lock(lock_a, lock_b);

std::unique_lock占用更多的空間，也比std::lock_guard更慢，但是std::unique_lock對(duì)象可以不占用關(guān)聯(lián)的互斥，具備這份靈活性需要付出代價(jià)：需要存儲(chǔ)并且更新互斥信息。
std::defer_lock將互斥保留為無鎖狀態(tài)。

（3）C++17 提供一個(gè)新的RAII類模版std::scoped_lock<>。std::scoped_lock<>和std::scoped_guard<> 完全等價(jià)，只不過前者是可變參數(shù)類模版，接收各種互斥類型作為模版參數(shù)類表，還以多個(gè)互斥對(duì)象作為構(gòu)造函數(shù)的參數(shù)列表。

swap(X& lhs, X& rhs)
{
	if(&lhs == &rhs)
		return;
	std::scoped_lock guard(lhs.m, rhs.m);

	swap(lhs.some_detail, rhs.some_detail);	
}

std::scoped_lock guard(lhs.m, rhs.m)等價(jià)于 std::scoped_lock<std::mutex, std::mutex> guard(lhs.m, rhs.m);

2.5 如何使用std::unique_lock轉(zhuǎn)移互斥歸屬權(quán)？

因?yàn)?code>std::unique_lock實(shí)例不占有與之關(guān)聯(lián)的互斥，所以隨著其實(shí)例的轉(zhuǎn)移，互斥的歸屬權(quán)可以在多個(gè)std::unique_lock實(shí)例之間轉(zhuǎn)移。std::unique_lock是可移動(dòng)不可復(fù)制。

轉(zhuǎn)移有一種用途：準(zhǔn)許函數(shù)鎖定互斥，然后把互斥的歸屬權(quán)轉(zhuǎn)移給函數(shù)調(diào)用者，好讓它在同一鎖的保護(hù)下執(zhí)行其他操作。

看如下代碼，get_lock() 函數(shù)先鎖定互斥，接著對(duì)數(shù)據(jù)做前期準(zhǔn)備，再將歸屬權(quán)返回給調(diào)用者：

std::unique_lock<std::mutex> get_lock()
{
	extern std::mutex some_mutex;
	std::unique_lock<std::mutex> lk(some_mutex);
	prepare_data();
	return lk;
}
void process_data()
{
	std::unique_lock<std::mutex> lk(get_lock());
	do_something();
}

由于鎖lk是get_lock函數(shù)中聲明的std::unique_lock局部變量，因此代碼無需調(diào)用std::move()就能把它直接返回，編譯器會(huì)妥善調(diào)用移動(dòng)構(gòu)造函數(shù)。

2.6 如何使用std::unique_lock按合適的粒度加鎖？

粒度精細(xì)的鎖保護(hù)少量數(shù)據(jù)，而粒度粗大的鎖保護(hù)大量數(shù)據(jù)。鎖操作有兩個(gè)要點(diǎn)：

• 選擇足夠粗大的鎖粒度，確保目標(biāo)數(shù)據(jù)受到保護(hù)；
• 限制范圍，務(wù)求只在必要的操作過程中持鎖。

std::unique_lock類具有成員函數(shù)lock()、unlock()、try_lock()
可用std::unique_lock處理：假如代碼不再需要訪問共享數(shù)據(jù)，那么我們就調(diào)用unlock()解鎖；若以后需要重新訪問，則調(diào)用lock()加鎖。

void get_and_process_data()
{
	std::unique_lock<std::mutex> my_lock(the_mutex);
	some_class data_to_process = get_next_data_chunk();
	my_lock.unlock();
	//假定調(diào)用process()期間，互斥無須加鎖
	result_type result = process(data_to_process);
	my_lock.lock();
	//重新鎖住互斥，以寫出結(jié)果
	write_result(data_to_process, result);
}

若只用單獨(dú)一個(gè)互斥保護(hù)整個(gè)數(shù)據(jù)結(jié)構(gòu)，不但可以加劇鎖的爭(zhēng)奪，還將難以縮短縮短持鎖時(shí)間。假設(shè)某種操作需對(duì)同一個(gè)互斥全程加鎖，當(dāng)中步驟越多，則持鎖時(shí)間越久。這是一種雙重?fù)p失，恰恰加倍促使我們盡可能該用粒度精細(xì)的鎖。

比如刪除某數(shù)據(jù)，得先進(jìn)行查詢?cè)賱h除，可以先獲取拷貝對(duì)象數(shù)據(jù)，遍歷完再進(jìn)行刪除。減少鎖持續(xù)時(shí)間。

2.7 如果對(duì)很少更新的數(shù)據(jù)結(jié)構(gòu)該如何優(yōu)化加鎖？

采用std::mutex保護(hù)數(shù)據(jù)結(jié)構(gòu)過于嚴(yán)苛，原因是即便沒發(fā)生改動(dòng)，他照樣會(huì)禁止并發(fā)訪問。C++17標(biāo)準(zhǔn)庫提供std::share_mutex，我們需要采用新類型的互斥。由于新類型的互斥具有兩種不同的使用方式，因此通常被稱為讀寫互斥：允許單獨(dú)一個(gè)“寫線程”進(jìn)行完全排它訪問，也允許多個(gè)“讀線程”共享數(shù)據(jù)或并發(fā)訪問。

共享鎖即讀鎖，對(duì)應(yīng)std::shared_lock<std::shared_mutex>
排他鎖即寫鎖，對(duì)應(yīng)std::lock_guard<std::shared_mutex>和std::unique_guard<std::shared_mutex>

運(yùn)用 std::shared_mutex 保護(hù)數(shù)據(jù)結(jié)構(gòu)，代碼如下：

#include <map>
#include <string>
#include <mutex>
#include <shared_mutex>
class dns_enty;
class dns_cache
{
	std::map<std::string, dns_entry> entries;
	mutable std::shared_mutex entry_mutex;
public:
	dns_entry find_entry(std::string const& domain) const
	{
		std::shared_lock<std::shared_mutex> lk(entry_mytex);
		std::map<std::string, dns_entry>::const_iterator const it = entries.find(domain);
		return(it == entries.end()) ? dns_entry() : it->second;
	}
	void update_or_add_entry(std::string const& domain, dns_entry const& dns_details)
	{
		std::lock_guard<std::shared_mutex> lk(entry_mutex);
		entries[domain] = dns_details;
    }
	
};

2.8 如何遞歸加鎖？

使用std::mutex，再次對(duì)其重新加鎖就會(huì)出錯(cuò)，將導(dǎo)致未定義行為。
使用std::recursive_mutex允許同一線程對(duì)某互斥的統(tǒng)一實(shí)例多次加鎖，我們必須先釋放全部的鎖，才可以讓另一個(gè)線程鎖住該互斥。

例如：
若我們對(duì)它調(diào)用3次lock()，就必須調(diào)用3次unlock()。只要正確的使用std::lock_guard<std::recurive_mutex> 和 std::unique_guard<std::recurive_mutex>，它們便會(huì)處理好遞歸鎖。

3、多線程中如何在初始化過程中保護(hù)共享數(shù)據(jù)

用互斥實(shí)現(xiàn)線程安全的延遲初始化，代碼如下：

std::shared_ptr<some_resource> resource_ptr;
std::mutex resource_mutex;
void foo()
{
	//此處，全部線程都被迫循環(huán)運(yùn)行
	std::unique_lock<std::mutex> lk(resource_mutex);
	if(resource_ptr)
	{
		//僅有初始化需要保護(hù)數(shù)據(jù)
		resource_ptr.reset(new some_resource);
	}
	lk.unlock();
	resource_ptr->do_something();
}

不過數(shù)據(jù)為多線程使用，那么它們便無法并發(fā)訪問，線程只能毫無必要的運(yùn)行，因?yàn)槊總€(gè)線程都必須在互斥上輪候，等待查驗(yàn)數(shù)據(jù)是否已經(jīng)完成初始化。

為此，C++標(biāo)準(zhǔn)庫中提供了std::once_flag類和std::call_once()函數(shù)。
令所有線程共同調(diào)用std::call_once函數(shù)，從而確保在該調(diào)用返回時(shí)，指針初始化由其中某線程安全且唯一完成（通過適合的同步機(jī)制）。必要同步數(shù)據(jù)則有std::once_flag實(shí)例存儲(chǔ)，每個(gè)std::call_flag實(shí)例對(duì)應(yīng)一次不同的初始化。相比顯示使用互斥，std::call_once()函數(shù)的額外開銷往往更低，特別是在初始已經(jīng)完成的情況下，所以如果功能符合需求就應(yīng)優(yōu)先使用。

延遲初始化代碼如下：

std::shared_ptr<some_resource> resource_ptr;
//實(shí)例存儲(chǔ)
std::once_flag resource_flag;
void init_resource()
{
	resource_ptr.reset(new some_resource);
}
void foo()
{
	//初始化函數(shù)準(zhǔn)確地被唯一一次調(diào)用
	std::call_once(resource_flag, init_resource);
	resource_ptr->do_something();
}

利用std::call_once()函數(shù)對(duì)類X的數(shù)據(jù)成員實(shí)施線程安全的延遲初始化：

class X
{
private:
	connection_info connection_details;
	connection_handle connection;
	std::once_flag connection_init_flag;
	void open_connection()
	{
		connection = connection_manager.open(connection_details);
    }
public:
	X(connection_info const& connection_details_);
		connection_details(connection_details_)
		{}
	void send_data(data_packet const& data)
	{
		std::call_once(connection_init_flag, &X::open_connection, this);
		connection.send_data(data);
	}
	data_packet receive_data()
	{
		std::call_once(connection_init_flag, &X::open_connection, this);
		return connection.receive_data();
	}
};

某些類的代碼只需用到唯一一個(gè)全局實(shí)例，這種情形可用以下方法代替std::call_once():

class my_class;
//線程安全的初始化，C++11標(biāo)準(zhǔn)保證其正確性
my_class& get_my_class_instance()
{
	static my_class inctance;
	return instance;
}

多個(gè)線程可以安全地調(diào)用get_my_class_instance()，而無需擔(dān)憂初始化的條件競(jìng)爭(zhēng)。

4、小結(jié)

世上無難事，只怕有心人。

點(diǎn)擊進(jìn)入系列文章目錄文章來源地址http://www.zghlxwxcb.cn/news/detail-477744.html

到了這里，關(guān)于C++并發(fā)線程 - 如何線程間共享數(shù)據(jù)【詳解：如何使用鎖操作】的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！