|
在上篇中,，我們已經(jīng)討論過如何去實(shí)現(xiàn)一個(gè) Map 了,，并且也討論了諸多優(yōu)化點(diǎn)。在下篇中,，我們將繼續(xù)討論如何實(shí)現(xiàn)一個(gè)線程安全的 Map,。說到線程安全，需要從概念開始說起,。 線程安全就是如果你的代碼塊所在的進(jìn)程中有多個(gè)線程在同時(shí)運(yùn)行，而這些線程可能會(huì)同時(shí)運(yùn)行這段代碼,。如果每次運(yùn)行結(jié)果和單線程運(yùn)行的結(jié)果是一樣的,，而且其他的變量的值也和預(yù)期的是一樣的，就是線程安全的,。 如果代碼塊中包含了對共享數(shù)據(jù)的更新操作,，那么這個(gè)代碼塊就可能是非線程安全的。但是如果代碼塊中類似操作都處于臨界區(qū)之中,，那么這個(gè)代碼塊就是線程安全的,。 通常有以下兩類避免競爭條件的方法來實(shí)現(xiàn)線程安全： 通常在線程安全的問題中，最常見的代碼塊就是函數(shù),。讓函數(shù)具有線程安全的最有效的方式就是使其可重入,。如果某個(gè)進(jìn)程中所有線程都可以并發(fā)的對函數(shù)進(jìn)行調(diào)用，并且無論他們調(diào)用該函數(shù)的實(shí)際執(zhí)行情況怎么樣,，該函數(shù)都可以產(chǎn)生預(yù)期的結(jié)果,，那么就可以說這個(gè)函數(shù)是可重入的。 如果一個(gè)函數(shù)把共享數(shù)據(jù)作為它的返回結(jié)果或者包含在它返回的結(jié)果中,，那么該函數(shù)就肯定不是一個(gè)可重入的函數(shù),。任何內(nèi)含了操作共享數(shù)據(jù)的代碼的函數(shù)都是不可重入的函數(shù)。 為了實(shí)現(xiàn)線程安全的函數(shù),，把所有代碼都置放于臨界區(qū)中是可行的,。但是互斥量的使用總會(huì)耗費(fèi)一定的系統(tǒng)資源和時(shí)間,，使用互斥量的過程總會(huì)存在各種博弈和權(quán)衡。所以請合理使用互斥量保護(hù)好那些涉及共享數(shù)據(jù)操作的代碼,。 注意：可重入只是線程安全的充分不必要條件,，并不是充要條件。這個(gè)反例在下面會(huì)講到,。 如果變量已經(jīng)被本地化,，所以每個(gè)線程都有自己的私有副本。這些變量通過子程序和其他代碼邊界保留它們的值,，并且是線程安全的,，因?yàn)檫@些變量都是每個(gè)線程本地存儲(chǔ)的，即使訪問它們的代碼可能被另一個(gè)線程同時(shí)執(zhí)行,，依舊是線程安全的,。 對象一旦初始化以后就不能改變。這意味著只有只讀數(shù)據(jù)被共享,，這也實(shí)現(xiàn)了固有的線程安全性,。可變（不是常量）操作可以通過為它們創(chuàng)建新對象,，而不是修改現(xiàn)有對象的方式去實(shí)現(xiàn),。 Java，C＃和 Python 中的字符串的實(shí)現(xiàn)就使用了這種方法,。 第一類方法都比較簡單,，通過代碼改造就可以實(shí)現(xiàn)。但是如果遇到一定要進(jìn)行線程中共享數(shù)據(jù)的情況,，第一類方法就解決不了了,。這時(shí)候就出現(xiàn)了第二類解決方案，利用線程同步的方法來解決線程安全問題,。 今天就從線程同步開始說起,。 在多線程的程序中，多以共享數(shù)據(jù)作為線程之間傳遞數(shù)據(jù)的手段,。由于一個(gè)進(jìn)程所擁有的相當(dāng)一部分虛擬內(nèi)存地址都可以被該進(jìn)程中所有線程共享,，所以這些共享數(shù)據(jù)大多是以內(nèi)存空間作為載體的。如果兩個(gè)線程同時(shí)讀取同一塊共享內(nèi)存但獲取到的數(shù)據(jù)卻不同,，那么程序很容易出現(xiàn)一些 bug,。 為了保證共享數(shù)據(jù)一致性，最簡單并且最徹底的方法就是使該數(shù)據(jù)成為一個(gè)不變量,。當(dāng)然這種絕對的方式在大多數(shù)情況下都是不可行的,。比如函數(shù)中會(huì)用到一個(gè)計(jì)數(shù)器，記錄函數(shù)被調(diào)用了幾次,，這個(gè)計(jì)數(shù)器肯定就不能被設(shè)為常量,。那這種必須是變量的情況下,，還要保證共享數(shù)據(jù)的一致性，這就引出了臨界區(qū)的概念,。 臨界區(qū)的出現(xiàn)就是為了使該區(qū)域只能被串行的訪問或者執(zhí)行。臨界區(qū)可以是某個(gè)資源,，也可以是某段代碼,。保證臨界區(qū)最有效的方式就是利用線程同步機(jī)制,。 先介紹2種共享數(shù)據(jù)同步的方法,。 在同一時(shí)刻，只允許一個(gè)線程處于臨界區(qū)之內(nèi)的約束稱為互斥,，每個(gè)線程在進(jìn)入臨界區(qū)之前,，都必須先鎖定某個(gè)對象，只有成功鎖定對象的線程才能允許進(jìn)入臨界區(qū),，否則就會(huì)阻塞,。這個(gè)對象稱為互斥對象或者互斥量。 一般我們?nèi)粘Ｕf的互斥鎖就能達(dá)到這個(gè)目的,。 互斥量可以有多個(gè),，它們所保護(hù)的臨界區(qū)也可以有多個(gè)。先從簡單的說起,，一個(gè)互斥量和一個(gè)臨界區(qū),。 上圖就是一個(gè)互斥量和一個(gè)臨界區(qū)的例子,。當(dāng)線程1先進(jìn)入臨界區(qū)的時(shí)候,，當(dāng)前臨界區(qū)處于未上鎖的狀態(tài)，于是它便先將臨界區(qū)上鎖,。線程1獲取到臨界區(qū)里面的值,。 這個(gè)時(shí)候線程2準(zhǔn)備進(jìn)入臨界區(qū)，由于線程1把臨界區(qū)上鎖了,，所以線程2進(jìn)入臨界區(qū)失敗,，線程2由就緒狀態(tài)轉(zhuǎn)成睡眠狀態(tài)。線程1繼續(xù)對臨界區(qū)的共享數(shù)據(jù)進(jìn)行寫入操作,。 當(dāng)線程1完成所有的操作以后,，線程1調(diào)用解鎖操作。當(dāng)臨界區(qū)被解鎖以后,，會(huì)嘗試喚醒正在睡眠的線程2,。線程2被喚醒以后，由睡眠狀態(tài)再次轉(zhuǎn)換成就緒狀態(tài),。線程2準(zhǔn)備進(jìn)入臨界區(qū),，當(dāng)臨界區(qū)此處處于未上鎖的狀態(tài),，線程2便將臨界區(qū)上鎖。 經(jīng)過 read,、write 一系列操作以后,，最終在離開臨界區(qū)的時(shí)候會(huì)解鎖。 線程在離開臨界區(qū)的時(shí)候,，一定要記得把對應(yīng)的互斥量解鎖,。這樣其他因臨界區(qū)被上鎖而導(dǎo)致睡眠的線程還有機(jī)會(huì)被喚醒。所以對同一個(gè)互斥變量的鎖定和解鎖必須成對的出現(xiàn),。既不可以對一個(gè)互斥變量進(jìn)行重復(fù)的鎖定,，也不能對一個(gè)互斥變量進(jìn)行多次的解鎖。 如果對一個(gè)互斥變量鎖定多次可能會(huì)導(dǎo)致臨界區(qū)最終永遠(yuǎn)阻塞,。可能有人會(huì)問了,，對一個(gè)未鎖定的互斥變成解鎖多次會(huì)出現(xiàn)什么問題呢,？ 在 Go 1.8 之前，雖然對互斥變量解鎖多次不會(huì)引起任何 goroutine 的阻塞,，但是它可能引起一個(gè)運(yùn)行時(shí)的恐慌,。Go 1.8 之前的版本，是可以嘗試恢復(fù)這個(gè)恐慌的,，但是恢復(fù)以后,，可能會(huì)導(dǎo)致一系列的問題，比如重復(fù)解鎖操作的 goroutine 會(huì)永久的阻塞,。所以 Go 1.8 版本以后此類運(yùn)行時(shí)的恐慌就變成了不可恢復(fù)的了,。所以對互斥變量反復(fù)解鎖就會(huì)導(dǎo)致運(yùn)行時(shí)操作，最終程序異常退出,。 在這種情況下,，極容易產(chǎn)生線程死鎖的情況。所以盡量不要讓不同的互斥量所保護(hù)的臨界區(qū)重疊,。 上圖這個(gè)例子中，一個(gè)臨界區(qū)中存在2個(gè)互斥量：互斥量 A 和互斥量 B,。 線程1先鎖定了互斥量 A ,，接著線程2鎖定了互斥量 B。當(dāng)線程1在成功鎖定互斥量 B 之前永遠(yuǎn)不會(huì)釋放互斥量 A,。同樣,，線程2在成功鎖定互斥量 A 之前永遠(yuǎn)不會(huì)釋放互斥量 B。那么這個(gè)時(shí)候線程1和線程2都因無法鎖定自己需要鎖定的互斥量，都由 ready 就緒狀態(tài)轉(zhuǎn)換為 sleep 睡眠狀態(tài),。這是就產(chǎn)生了線程死鎖了,。 線程死鎖的產(chǎn)生原因有以下幾種： 1.系統(tǒng)資源競爭2.進(jìn)程推薦順序非法3.死鎖必要條件（必要條件中任意一個(gè)不滿足，死鎖都不會(huì)發(fā)生）(1). 互斥條件(2). 不剝奪條件(3). 請求和保持條件(4). 循環(huán)等待條件 想避免線程死鎖的情況發(fā)生有以下幾種方法可以解決： 1.預(yù)防死鎖(1). 資源有序分配法（破壞環(huán)路等待條件）(2). 資源原子分配法（破壞請求和保持條件）2.避免死鎖銀行家算法3.檢測死鎖死鎖定理（資源分配圖化簡法）,，這種方法雖然可以檢測,，但是無法預(yù)防，檢測出來了死鎖還需要配合解除死鎖的方法才行,。 徹底解決死鎖有以下幾種方法： 1.剝奪資源2.撤銷進(jìn)程3.試鎖定 — 回退如果在執(zhí)行一個(gè)代碼塊的時(shí)候,，需要先后（順序不定）鎖定兩個(gè)變量，那么在成功鎖定其中一個(gè)互斥量之后應(yīng)該使用試鎖定的方法來鎖定另外一個(gè)變量,。如果試鎖定第二個(gè)互斥量失敗,，就把已經(jīng)鎖定的第一個(gè)互斥量解鎖，并重新對這兩個(gè)互斥量進(jìn)行鎖定和試鎖定,。 如上圖,，線程2在鎖定互斥量 B 的時(shí)候,，再試鎖定互斥量 A，此時(shí)鎖定失敗,，于是就把互斥量 B 也一起解鎖,。接著線程1會(huì)來鎖定互斥量 A。此時(shí)也不會(huì)出現(xiàn)死鎖的情況,。 4.固定順序鎖定 這種方式就是讓線程1和線程2都按照相同的順序鎖定互斥量，都按成功鎖定互斥量1以后才能去鎖定互斥量2 ,。這樣就能保證在一個(gè)線程完全離開這些重疊的臨界區(qū)之前,，不會(huì)有其他同樣需要鎖定那些互斥量的線程進(jìn)入到那里。 多個(gè)臨界區(qū)和多個(gè)互斥量的情況就要看是否會(huì)有沖突的區(qū)域,，如果出現(xiàn)相互交集的沖突區(qū)域，后進(jìn)臨界區(qū)的線程就會(huì)進(jìn)入睡眠狀態(tài),，直到該臨界區(qū)的線程完成任務(wù)以后,，再被喚醒。 一般情況下,，應(yīng)該盡量少的使用互斥量,。每個(gè)互斥量保護(hù)的臨界區(qū)應(yīng)該在合理范圍內(nèi)并盡量大。但是如果發(fā)現(xiàn)多個(gè)線程會(huì)頻繁出入某個(gè)較大的臨界區(qū),，并且它們之間經(jīng)常存在訪問沖突,，那么就應(yīng)該把這個(gè)較大的臨界區(qū)劃分的更小一點(diǎn)，并使用不同的互斥量保護(hù)起來。這樣做的目的就是為了讓等待進(jìn)入同一個(gè)臨界區(qū)的線程數(shù)變少,，從而降低線程被阻塞的概率,，并減少它們被迫進(jìn)入睡眠狀態(tài)的時(shí)間，這從一定程度上提高了程序的整體性能,。 在說另外一個(gè)線程同步的方法之前,，回答一下文章開頭留下的一個(gè)疑問：可重入只是線程安全的充分不必要條件，并不是充要條件,。這個(gè)反例在下面會(huì)講到,。 這個(gè)問題最關(guān)鍵的一點(diǎn)在于：mutex 是不可重入的。 舉個(gè)例子： 在下面這段代碼中,，函數(shù) increment_counter 是線程安全的,，但不是可重入的。 #include int increment_counter () { static int counter=0; static pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER; pthread_mutex_lock(&mutex); // only allow one thread to increment at a time ++counter; // store value before any other threads increment it further int result=counter; pthread_mutex_unlock(&mutex); return result; } 上面的代碼中,，函數(shù) increment_counter 可以在多個(gè)線程中被調(diào)用,，因?yàn)橛幸粋€(gè)互斥鎖 mutex 來同步對共享變量 counter 的訪問。但是如果這個(gè)函數(shù)用在可重入的中斷處理程序中,，如果在 pthread_mutex_lock(&mutex) 和 pthread_mutex_unlock(&mutex) 之間產(chǎn)生另一個(gè)調(diào)用函數(shù) increment_counter 的中斷,，則會(huì)第二次執(zhí)行此函數(shù)，此時(shí)由于 mutex 已被 lock,，函數(shù)會(huì)在 pthread_mutex_lock(&mutex) 處阻塞,，并且由于 mutex 沒有機(jī)會(huì)被 unlock，阻塞會(huì)永遠(yuǎn)持續(xù)下去,。簡言之,，問題在于 pthread 的 mutex 是不可重入的。 解決辦法是設(shè)定 PTHREAD_MUTEX_RECURSIVE 屬性,。然而對于給出的問題而言,，專門使用一個(gè) mutex 來保護(hù)一次簡單的增量操作顯然過于昂貴，因此 c++11 中的 原子變量 提供了一個(gè)可使此函數(shù)既線程安全又可重入（而且還更簡潔）的替代方案： #include int increment_counter () { static std::atomic counter(0); // increment is guaranteed to be done atomically int result=++counter; return result; } 在 Go 中,，互斥量在標(biāo)準(zhǔn)庫代碼包 sync 中的 Mutex 結(jié)構(gòu)體表示的,。sync.Mutex 類型只有兩個(gè)公開的指針方法，Lock 和 Unlock,。前者用于鎖定當(dāng)前的互斥量,，后者則用于對當(dāng)前的互斥量進(jìn)行解鎖。 在線程同步的方法中,，還有一個(gè)可以與互斥量相提并論的同步方法,，條件變量。 條件變量與互斥量不同,，條件變量的作用并不是保證在同一時(shí)刻僅有一個(gè)線程訪問某一個(gè)共享數(shù)據(jù),，而是在對應(yīng)的共享數(shù)據(jù)的狀態(tài)發(fā)生變化時(shí)，通知其他因此而被阻塞的線程。條件變量總是與互斥變量組合使用的,。 這類問題其實(shí)很常見,。先用生產(chǎn)者消費(fèi)者的例子來舉例。 如果不用條件變量,，只用互斥量,，來看看會(huì)發(fā)生什么后果。 生產(chǎn)者線程在完成添加操作之前,，其他的生產(chǎn)者線程和消費(fèi)者線程都無法進(jìn)行操作,。同一個(gè)商品也只能被一個(gè)消費(fèi)者消費(fèi)。 如果只用互斥量,，可能會(huì)出現(xiàn)2個(gè)問題,。 1.生產(chǎn)者線程獲得了互斥量以后，卻發(fā)現(xiàn)商品已滿,，無法再添加新的商品了,。于是該線程就會(huì)一直等待。新的生產(chǎn)者也進(jìn)入不了臨界區(qū),，消費(fèi)者也無法進(jìn)入,。這時(shí)候就死鎖了。2.消費(fèi)者線程獲得了互斥量以后,，卻發(fā)現(xiàn)商品是空的,，無法消費(fèi)了。這個(gè)時(shí)候該線程也是會(huì)一直等待,。新的生產(chǎn)者和消費(fèi)者也都無法進(jìn)入。這時(shí)候同樣也死鎖了,。 這就是只用互斥量無法解決的問題,。在多個(gè)線程之間，急需一套同步的機(jī)制,，能讓這些線程都協(xié)作起來,。 條件變量就是大家熟悉的 P - V 操作了,。這塊大家應(yīng)該比較熟悉,，所以簡單的過一下。 P 操作就是 wait 操作,，它的意思就是阻塞當(dāng)前線程,，直到收到該條件變量發(fā)來的通知。 V 操作就是 signal 操作,，它的意思就是讓該條件變量向至少一個(gè)正在等待它通知的線程發(fā)送通知,，以表示某個(gè)共享數(shù)據(jù)的狀態(tài)已經(jīng)變化。 Broadcast 廣播通知，它的意思就是讓條件變量給正在等待它通知的所有線程發(fā)送通知,，以表示某個(gè)古董交易共享數(shù)據(jù)的狀態(tài)已經(jīng)發(fā)生改變,。 signal 可以操作多次,，如果操作3次,，就代表發(fā)了3次信號(hào)通知。如上圖,。 P - V 操作設(shè)計(jì)美妙之處在于，P 操作的次數(shù)與 V 操作的次數(shù)是相同的,。wait 多少次,，signal 對應(yīng)的有多少次?？瓷蠄D,，這個(gè)循環(huán)就是這么的奇妙。 這個(gè)問題可以形象的描述成像上圖這樣,，門衛(wèi)守護(hù)著臨界區(qū)的安全。售票廳記錄著當(dāng)前 semaphone 的值,，它也控制著門衛(wèi)是否打開臨界區(qū),。 臨界區(qū)只允許一個(gè)線程進(jìn)入,，當(dāng)已經(jīng)有一個(gè)線程了,，再來一個(gè)線程，就會(huì)被 lock 住,。售票廳也會(huì)記錄當(dāng)前阻塞的線程數(shù),。 當(dāng)之前的線程離開以后,，售票廳就會(huì)告訴門衛(wèi),，允許一個(gè)線程進(jìn)入臨界區(qū)。 用 P-V 偽代碼來描述生產(chǎn)者消費(fèi)者： 初始變量： semaphore mutex=1; // 臨界區(qū)互斥信號(hào)量 semaphore empty=n; // 空閑緩沖區(qū)個(gè)數(shù) semaphore full=0; // 緩沖區(qū)初始化為空 生產(chǎn)者線程： producer() { while(1) { produce an item in nextp; P(empty); P(mutex); add nextp to buffer; V(mutex); V(full); } } 消費(fèi)者線程： consumer() { while(1) { P(full); P(mutex); remove an item from buffer; V(mutex); V(empty); consume the item; } } 雖然在生產(chǎn)者和消費(fèi)者單個(gè)程序里面 P,，V 并不是成對的,，但是整個(gè)程序里面 P，V 還是成對的,。 讀者優(yōu)先,，寫進(jìn)程被延遲。只要有讀者在讀,，后來的讀者都可以隨意進(jìn)來讀,。 讀者要先進(jìn)入 rmutex ，查看 readcount,，然后修改 readcout 的值,，最后再去讀數(shù)據(jù)。對于每個(gè)讀進(jìn)程都是寫者,，都要進(jìn)去修改 readcount 的值,，所以還要單獨(dú)設(shè)置一個(gè) rmutex 互斥訪問。 初始變量： int readcount=0; // 讀者數(shù)量 semaphore rmutex=1; // 保證更新 readcount 互斥 semaphore wmutex=1; // 保證讀者和寫著互斥的訪問文件 讀者線程： reader() { while(1) { P(rmutex); // 準(zhǔn)備進(jìn)入,，修改 readcount,，“開門” if(readcount==0) { // 說明是第一個(gè)讀者 P(wmutex); // 拿到”鑰匙”，阻止寫線程來寫 } readcount ++; V(rmutex); reading; P(rmutex); // 準(zhǔn)備離開 readcount --; if(readcount==0) { // 說明是最后一個(gè)讀者 V(wmutex); // 交出”鑰匙”,，讓寫線程來寫 } V(rmutex); // 離開,，“關(guān)門” } } 寫者線程： writer() { while(1) { P(wmutex); writing; V(wmutex); } } 有寫者寫，禁止后面的讀者來讀,。在寫者前的讀者,，讀完就走。只要有寫者在等待,，禁止后來的讀者進(jìn)去讀,。 初始變量： int readcount=0; // 讀者數(shù)量 semaphore rmutex=1; // 保證更新 readcount 互斥 semaphore wmutex=1; // 保證讀者和寫著互斥的訪問文件 semaphore w=1; // 用于實(shí)現(xiàn)“寫者優(yōu)先” 讀者線程： reader() { while(1) { P(w); // 在沒有寫者的時(shí)候才能請求進(jìn)入 P(rmutex); // 準(zhǔn)備進(jìn)入,，修改 readcount,，“開門” if(readcount==0) { // 說明是第一個(gè)讀者 P(wmutex); // 拿到”鑰匙”，阻止寫線程來寫 } readcount ++; V(rmutex); V(w); reading; P(rmutex); // 準(zhǔn)備離開 readcount --; if(readcount==0) { // 說明是最后一個(gè)讀者 V(wmutex); // 交出”鑰匙”,，讓寫線程來寫 } V(rmutex); // 離開,，“關(guān)門” } } 寫者線程： writer() { while(1) { P(w); P(wmutex); writing; V(wmutex); V(w); } } 假設(shè)有五位哲學(xué)家圍坐在一張圓形餐桌旁，做以下兩件事情之一：吃飯,，或者思考,。吃東西的時(shí)候，他們就停止思考,，思考的時(shí)候也停止吃東西。餐桌中間有一大碗意大利面,，每兩個(gè)哲學(xué)家之間有一只餐叉,。因?yàn)橛靡恢徊筒婧茈y吃到意大利面，所以假設(shè)哲學(xué)家必須用兩只餐叉吃東西,。他們只能使用自己左右手邊的那兩只餐叉,。哲學(xué)家就餐問題有時(shí)也用米飯和筷子而不是意大利面和餐叉來描述，因?yàn)楹苊黠@,，吃米飯必須用兩根筷子,。 初始變量： semaphore chopstick[5]={1,1,1,1,1}; // 初始化信號(hào)量 semaphore mutex=1; // 設(shè)置取筷子的信號(hào)量 哲學(xué)家線程： Pi() { do { P(mutex); // 獲得取筷子的互斥量 P(chopstick[i]); // 取左邊的筷子 P(chopstick[ (i + 1) % 5 ]); // 取右邊的筷子 V(mutex); // 釋放取筷子的信號(hào)量 eat; V(chopstick[i]); // 放回左邊的筷子 V(chopstick[ (i + 1) % 5 ]); // 放回右邊的筷子 think; }while(1); } |
|
|
