死鎖的原理非常簡(jiǎn)單，用一句話(huà)就可以描述完。就是當(dāng)多線(xiàn)程訪(fǎng)問(wèn)多個(gè)鎖的時(shí)候，不同的鎖被不同的線(xiàn)程持有，它們都在等待其他線(xiàn)程釋放出鎖來(lái)，于是便陷入了永久等待。比如A線(xiàn)程持有1號(hào)鎖，等待2號(hào)鎖，B線(xiàn)程持有2號(hào)鎖等待1號(hào)鎖，那么它們永遠(yuǎn)也等不到執(zhí)行的那天，這種情況就叫做死鎖。

關(guān)于死鎖有一個(gè)著名的問(wèn)題叫做哲學(xué)家就餐問(wèn)題，有5個(gè)哲學(xué)家圍坐在一起，他們每個(gè)人需要拿到兩個(gè)叉子才可以吃飯。如果他們同時(shí)拿起自己左手邊的叉子，那么就會(huì)永遠(yuǎn)等待右手邊的叉子釋放出來(lái)。這樣就陷入了永久等待，于是這些哲學(xué)家都會(huì)餓死。

這是一個(gè)很形象的模型，因?yàn)樵谟?jì)算機(jī)并發(fā)場(chǎng)景當(dāng)中，一些資源的數(shù)量往往是有限的。很有可能出現(xiàn)多個(gè)線(xiàn)程搶占的情況，如果處理不好就會(huì)發(fā)生大家都獲取了一個(gè)資源，然后在等待另外的資源的情況。

對(duì)于死鎖的問(wèn)題有多種解決方法，這里我們介紹比較簡(jiǎn)單的一種，就是對(duì)這些鎖進(jìn)行編號(hào)。我們規(guī)定當(dāng)一個(gè)線(xiàn)程需要同時(shí)持有多個(gè)鎖的時(shí)候，必須要按照序號(hào)升序的順序?qū)@些鎖進(jìn)行訪(fǎng)問(wèn)。通過(guò)上下文管理器我們可以很容易實(shí)現(xiàn)這一點(diǎn)。

上下文管理器

首先我們來(lái)簡(jiǎn)單介紹一下上下文管理器，上下文管理器我們其實(shí)經(jīng)常使用，比如我們經(jīng)常使用的 with語(yǔ)句 就是一個(gè)上下文管理器的經(jīng)典使用。當(dāng)我們通過(guò)with語(yǔ)句打開(kāi)文件的時(shí)候，它會(huì)自動(dòng)替我們處理好文件讀取之后的關(guān)閉以及拋出異常的處理，可以節(jié)約我們大量的代碼。

同樣我們也可以自己定義一個(gè)上下文處理器，其實(shí)很簡(jiǎn)單，我們只需要實(shí)現(xiàn)__enter__和__exit__這兩個(gè)函數(shù)即可。__enter__函數(shù)用來(lái)實(shí)現(xiàn)進(jìn)入資源之前的操作和處理，那么顯然__exit__函數(shù)對(duì)應(yīng)的就是使用資源結(jié)束之后或者是出現(xiàn)異常的處理邏輯。有了這兩個(gè)函數(shù)之后，我們就有了自己的上下文處理類(lèi)了。

我們來(lái)看一個(gè)樣例：

class Sample: def __enter__(self): print(’enter resources’) return self def __exit__(self, exc_type, exc_val, exc_tb): print(’exit’) # print(exc_type) # print(exc_val) # print(exc_tb) def doSomething(self): a = 1/1 return adef getSample(): return Sample()if __name__ == ’__main__’: with getSample() as sample: print(’do something’) sample.doSomething()

當(dāng)我們運(yùn)行這段代碼的時(shí)候，屏幕上打印的結(jié)果和我們的預(yù)期是一致的。

我們觀察一下__exit__函數(shù)，會(huì)發(fā)現(xiàn)它的參數(shù)有4個(gè)，后面的三個(gè)參數(shù)對(duì)應(yīng)的是拋出異常的情況。type對(duì)應(yīng)異常的類(lèi)型，val對(duì)應(yīng)異常時(shí)的輸出值，trace對(duì)應(yīng)異常拋出時(shí)的運(yùn)行堆棧。這些信息都是我們排查異常的時(shí)候經(jīng)常需要用到的信息，通過(guò)這三個(gè)字段，我們可以根據(jù)我們的需要對(duì)可能出現(xiàn)的異常進(jìn)行自定義的處理。

實(shí)現(xiàn)上下文管理器并不一定要通過(guò)類(lèi)實(shí)現(xiàn)，Python當(dāng)中也提供了上下文管理的注解，通過(guò)使用注解我們可以很方便地實(shí)現(xiàn)上下文管理。我們同樣也來(lái)看一個(gè)例子：

import timefrom contextlib import contextmanager@contextmanagerdef timethis(label): start = time.time() try: yield finally: end = time.time() print(’{}: {}’.format(label, end - start))with timethis(’timer’): pass

在這個(gè)方法當(dāng)中yield之前的部分相當(dāng)于__enter__函數(shù)，yield之后的部分相當(dāng)于__exit__。如果出現(xiàn)異常會(huì)在try語(yǔ)句當(dāng)中拋出，那么我們編寫(xiě)except對(duì)異常進(jìn)行處理即可。

避免死鎖

了解了上下文管理器之后，我們要做的就是 在lock的外面包裝一層 ，使得我們?cè)讷@取和釋放鎖的時(shí)候可以根據(jù)我們的需要，對(duì)鎖進(jìn)行排序，按照升序的順序進(jìn)行持有。

這段代碼源于Python的著名進(jìn)階書(shū)籍《Python cookbook》，非常經(jīng)典：

from contextlib import contextmanager# 用來(lái)存儲(chǔ)local的數(shù)據(jù)_local = threading.local()@contextmanagerdef acquire(*locks): # 對(duì)鎖按照id進(jìn)行排序 locks = sorted(locks, key=lambda x: id(x)) # 如果已經(jīng)持有鎖當(dāng)中的序號(hào)有比當(dāng)前更大的，說(shuō)明策略失敗 acquired = getattr(_local,’acquired’,[]) if acquired and max(id(lock) for lock in acquired) >= id(locks[0]): raise RuntimeError(’Lock Order Violation’) # 獲取所有鎖 acquired.extend(locks) _local.acquired = acquired try: for lock in locks: lock.acquire() yield finally: # 倒敘釋放 for lock in reversed(locks): lock.release() del acquired[-len(locks):]

這段代碼寫(xiě)得非常漂亮，可讀性很高，邏輯我們都應(yīng)該能看懂，但是有一個(gè)小問(wèn)題是這里用到了 threading.local 這個(gè)組件。

它是一個(gè)多線(xiàn)程場(chǎng)景當(dāng)中的 共享變量 ，雖然說(shuō)是共享的，但是對(duì)于每個(gè)線(xiàn)程來(lái)說(shuō)讀取到的值都是獨(dú)立的。聽(tīng)起來(lái)有些難以理解，其實(shí)我們可以將它理解成一個(gè)dict，dict的key是每一個(gè)線(xiàn)程的id，value是一個(gè)存儲(chǔ)數(shù)據(jù)的dict。每個(gè)線(xiàn)程在訪(fǎng)問(wèn)local變量的時(shí)候，都相當(dāng)于先通過(guò)線(xiàn)程id獲取了一個(gè)獨(dú)立的dict，再對(duì)這個(gè)dict進(jìn)行的操作。

看起來(lái)我們?cè)谑褂玫臅r(shí)候直接使用了_local，這是因?yàn)橥ㄟ^(guò)線(xiàn)程id先進(jìn)行查詢(xún)的步驟在其中封裝了。不明就里的話(huà)可能會(huì)覺(jué)得有些難以理解。

我們?cè)賮?lái)看下這個(gè)acquire的使用：

x_lock = threading.Lock()y_lock = threading.Lock()def thread_1(): while True: with acquire(x_lock, y_lock): print(’Thread-1’)def thread_2(): while True: with acquire(y_lock, x_lock): print(’Thread-2’)t1 = threading.Thread(target=thread_1)t1.start()t2 = threading.Thread(target=thread_2)t2.start()

運(yùn)行一下會(huì)發(fā)現(xiàn)沒(méi)有出現(xiàn)死鎖的情況，但如果我們把代碼稍加調(diào)整，寫(xiě)成這樣，那么就會(huì)觸發(fā)異常了。

def thread_1(): while True: with acquire(x_lock): with acquire(y_lock): print(’Thread-1’)def thread_2(): while True: with acquire(y_lock): with acquire(x_lock): print(’Thread-1’)

因?yàn)槲覀儼焰i寫(xiě)成了層次結(jié)構(gòu)，這樣就沒(méi)辦法進(jìn)行排序保證持有的有序性了，那么就會(huì)觸發(fā)我們代碼當(dāng)中定義的異常。

最后我們?cè)賮?lái)看下哲學(xué)家就餐問(wèn)題，通過(guò)我們自己實(shí)現(xiàn)的acquire函數(shù)我們可以非常方便地解決他們死鎖吃不了飯的問(wèn)題。

import threadingdef philosopher(left, right): while True: with acquire(left,right): print(threading.currentThread(), ’eating’)# 叉子的數(shù)量NSTICKS = 5chopsticks = [threading.Lock() for n in range(NSTICKS)]for n in range(NSTICKS): t = threading.Thread(target=philosopher, args=(chopsticks[n],chopsticks[(n+1) % NSTICKS])) t.start()

關(guān)于死鎖的問(wèn)題，對(duì)鎖進(jìn)行排序 只是其中的一種解決方案 ，除此之外還有很多解決死鎖的模型。比如我們可以讓線(xiàn)程在嘗試持有新的鎖失敗的時(shí)候主動(dòng)放棄所有目前已經(jīng)持有的鎖，比如我們可以設(shè)置機(jī)制檢測(cè)死鎖的發(fā)生并對(duì)其進(jìn)行處理等等。發(fā)散出去其實(shí)有很多種方法，這些方法起作用的原理各不相同，其中涉及大量操作系統(tǒng)的基礎(chǔ)概念和知識(shí)，感興趣的同學(xué)可以深入研究一下這個(gè)部分，一定會(huì)對(duì)操作系統(tǒng)以及鎖的使用有一個(gè)深刻的認(rèn)識(shí)。

以上就是python 多線(xiàn)程死鎖問(wèn)題的解決方案的詳細(xì)內(nèi)容，更多關(guān)于python 多線(xiàn)程死鎖的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！