Python多進(jìn)程中避免死鎖問題的六種策略
更新時間：2025年12月09日 08:26:08 作者：三維空間
死鎖是Python多進(jìn)程開發(fā)中最容易踩的坑之一,一旦出現(xiàn)會導(dǎo)致進(jìn)程卡死、程序無響應(yīng),甚至需要強制終止才能恢復(fù),所以本文給大家介紹了Python多進(jìn)程中避免死鎖問題的六種策略,需要的朋友可以參考下
一、先搞懂：多進(jìn)程死鎖到底是什么？
1. 死鎖的核心定義
死鎖是指兩個或多個進(jìn)程互相持有對方需要的鎖（或資源），且都不釋放自己持有的鎖，導(dǎo)致所有進(jìn)程都陷入“等待對方釋放資源”的無限阻塞狀態(tài)。
2. 死鎖產(chǎn)生的4個必要條件（缺一不可）
只有同時滿足以下4個條件，才會觸發(fā)死鎖，打破任意一個條件就能避免死鎖：
- 互斥條件：資源（如鎖、文件）只能被一個進(jìn)程占用；
- 請求與保持條件：進(jìn)程持有一個資源的同時，又請求另一個被占用的資源；
- 不剝奪條件：進(jìn)程已持有的資源不能被強制剝奪，只能主動釋放；
- 循環(huán)等待條件：多個進(jìn)程形成“你等我、我等你”的循環(huán)等待鏈。
3. 多進(jìn)程死鎖典型場景（新手高頻踩坑）
import multiprocessing
import time
場景：兩個進(jìn)程互相等待對方的鎖
def process1(lock1, lock2):
進(jìn)程1先拿lock1，再嘗試拿lock2
lock1.acquire()
print("進(jìn)程1獲取了lock1，等待lock2...")
time.sleep(1) 放大死鎖概率
lock2.acquire() 此時lock2已被進(jìn)程2持有，進(jìn)程1阻塞
print("進(jìn)程1獲取了lock2，執(zhí)行任務(wù)")
lock1.release()
lock2.release()
def process2(lock1, lock2):
進(jìn)程2先拿lock2，再嘗試拿lock1
lock2.acquire()
print("進(jìn)程2獲取了lock2，等待lock1...")
time.sleep(1)
lock1.acquire() 此時lock1已被進(jìn)程1持有，進(jìn)程2阻塞
print("進(jìn)程2獲取了lock1，執(zhí)行任務(wù)")
lock1.release()
lock2.release()
if __name__ == "__main__":
lock1 = multiprocessing.Lock()
lock2 = multiprocessing.Lock()
p1 = multiprocessing.Process(target=process1, args=(lock1, lock2))
p2 = multiprocessing.Process(target=process2, args=(lock1, lock2))
p1.start()
p2.start()
p1.join()
p2.join()
print("程序結(jié)束") 永遠(yuǎn)不會執(zhí)行到這一行
現(xiàn)象：程序輸出“進(jìn)程1獲取了lock1，等待lock2...”和“進(jìn)程2獲取了lock2，等待lock1...”后卡死，無法繼續(xù)執(zhí)行。
二、避免死鎖的6個核心策略（從易到難）
1. 策略1：使用with語句自動釋放鎖（最推薦）
with語句會在代碼塊執(zhí)行完成后自動釋放鎖，即使代碼拋出異常也能保證鎖釋放，從根本上避免“忘記release()”導(dǎo)致的死鎖。
修復(fù)上述死鎖案例（僅改鎖的使用方式）：
def process1(lock1, lock2):
with lock1: 自動acquire()，代碼塊結(jié)束自動release()
print("進(jìn)程1獲取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("進(jìn)程1獲取了lock2，執(zhí)行任務(wù)")
def process2(lock1, lock2):
with lock2:
print("進(jìn)程2獲取了lock2，等待lock1...")
time.sleep(1)
with lock1:
print("進(jìn)程2獲取了lock1，執(zhí)行任務(wù)")
注意：這個修改僅解決“鎖未釋放”的問題，但上述場景仍會因“循環(huán)等待”觸發(fā)死鎖，需結(jié)合策略2使用。
2. 策略2：統(tǒng)一鎖的獲取順序（打破循環(huán)等待）
死鎖的核心誘因之一是“進(jìn)程獲取鎖的順序不一致”，只要讓所有進(jìn)程按相同的順序獲取鎖，就能打破循環(huán)等待條件。
最終修復(fù)死鎖案例：
def process1(lock1, lock2):
統(tǒng)一先拿lock1，再拿lock2
with lock1:
print("進(jìn)程1獲取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("進(jìn)程1獲取了lock2，執(zhí)行任務(wù)")
def process2(lock1, lock2):
同樣先拿lock1，再拿lock2（不再先拿lock2）
with lock1:
print("進(jìn)程2獲取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("進(jìn)程2獲取了lock2，執(zhí)行任務(wù)")
執(zhí)行結(jié)果：進(jìn)程1先獲取lock1，進(jìn)程2等待lock1；進(jìn)程1執(zhí)行完成釋放鎖后，進(jìn)程2獲取lock1和lock2，無死鎖。
3. 策略3：給鎖添加超時時間（避免無限等待）
multiprocessing.Lock本身不支持超時，但可使用multiprocessing.RLock（可重入鎖）或threading.Lock（多進(jìn)程中需通過Manager傳遞）的acquire(timeout)方法，設(shè)置超時時間，超時后放棄獲取鎖，避免無限阻塞。
示例：帶超時的鎖獲取
import multiprocessing
import time
def process1(lock1, lock2):
獲取lock1，超時時間3秒
if lock1.acquire(timeout=3):
print("進(jìn)程1獲取了lock1，等待lock2...")
time.sleep(1)
獲取lock2，超時時間3秒
if lock2.acquire(timeout=3):
print("進(jìn)程1獲取了lock2，執(zhí)行任務(wù)")
lock2.release()
else:
print("進(jìn)程1獲取lock2超時，釋放lock1")
lock1.release()
else:
print("進(jìn)程1獲取lock1超時")
def process2(lock1, lock2):
if lock2.acquire(timeout=3):
print("進(jìn)程2獲取了lock2，等待lock1...")
time.sleep(1)
if lock1.acquire(timeout=3):
print("進(jìn)程2獲取了lock1，執(zhí)行任務(wù)")
lock1.release()
else:
print("進(jìn)程2獲取lock1超時，釋放lock2")
lock2.release()
else:
print("進(jìn)程2獲取lock2超時")
if __name__ == "__main__":
通過Manager創(chuàng)建支持超時的鎖
manager = multiprocessing.Manager()
lock1 = manager.Lock()
lock2 = manager.Lock()
p1 = multiprocessing.Process(target=process1, args=(lock1, lock2))
p2 = multiprocessing.Process(target=process2, args=(lock1, lock2))
p1.start()
p2.start()
p1.join()
p2.join()
print("程序結(jié)束")
執(zhí)行結(jié)果：進(jìn)程1獲取lock1，進(jìn)程2獲取lock2；雙方等待對方的鎖超時后，釋放自己持有的鎖，程序正常結(jié)束，無死鎖。
4. 策略4：減少鎖的使用范圍（最小化鎖持有時間）
只在“必須保護(hù)共享資源”的代碼塊加鎖，執(zhí)行完核心邏輯后立即釋放鎖，縮短鎖的持有時間，降低多個進(jìn)程同時爭用鎖的概率。
反面案例（鎖持有時間過長）：
def write_file(num, lock):
lock.acquire()
非核心邏輯（耗時），卻持有鎖
time.sleep(5) 模擬耗時操作
with open("test.txt", "a") as f:
f.write(f"進(jìn)程{num}寫入\n")
lock.release()
正面案例（僅核心邏輯加鎖）：
def write_file(num, lock):
非核心邏輯（耗時），不持有鎖
time.sleep(5)
僅寫入文件時加鎖
with lock:
with open("test.txt", "a") as f:
f.write(f"進(jìn)程{num}寫入\n")
5. 策略5：使用單鎖替代多鎖（簡化資源競爭）
如果多個鎖的作用是保護(hù)同一類共享資源（如多個文件都屬于“數(shù)據(jù)文件”），可合并為一個全局鎖，避免多鎖嵌套導(dǎo)致的死鎖。
示例：單鎖替代多鎖
import multiprocessing
import time
全局鎖（替代多個文件鎖）
global_lock = multiprocessing.Lock()
寫入文件A
def write_file_a(num):
with global_lock:
with open("file_a.txt", "a") as f:
f.write(f"進(jìn)程{num}寫入文件A\n")
time.sleep(1)
寫入文件B
def write_file_b(num):
with global_lock:
with open("file_b.txt", "a") as f:
f.write(f"進(jìn)程{num}寫入文件B\n")
time.sleep(1)
if __name__ == "__main__":
p1 = multiprocessing.Process(target=write_file_a, args=(1,))
p2 = multiprocessing.Process(target=write_file_b, args=(2,))
p1.start()
p2.start()
p1.join()
p2.join()
print("寫入完成")
6. 策略6：使用死鎖檢測工具（進(jìn)階）
對于復(fù)雜的多進(jìn)程場景，可通過工具檢測潛在的死鎖：
- 內(nèi)置工具：
multiprocessing.active_children()查看活躍進(jìn)程，結(jié)合日志定位阻塞的進(jìn)程； - 第三方工具：
py-spy（進(jìn)程分析工具），可實時查看進(jìn)程調(diào)用棧，定位死鎖位置：
安裝py-spy pip install py-spy 分析進(jìn)程（替換為你的進(jìn)程ID） py-spy dump --pid 12345
三、死鎖排查：快速定位問題的3個方法
- 查看進(jìn)程狀態(tài)：用
ps -ef | grep python查看進(jìn)程是否處于“D狀態(tài)”（不可中斷睡眠，大概率死鎖）； - 添加日志：在每個
acquire()和release()前后打印日志，定位哪個鎖導(dǎo)致阻塞； - 簡化場景：將復(fù)雜的多進(jìn)程邏輯拆分為最小可復(fù)現(xiàn)的demo，逐步排查死鎖誘因。
四、總結(jié)：避免死鎖的核心原則
- 自動釋放：優(yōu)先用
with語句管理鎖，杜絕“忘記release()”； - 順序一致：所有進(jìn)程按相同順序獲取多把鎖，打破循環(huán)等待；
- 超時兜底：給鎖添加超時時間，避免無限阻塞；
- 最小持有：僅在核心邏輯加鎖，縮短鎖的持有時間；
- 簡化鎖結(jié)構(gòu)：能用單鎖就不用多鎖，減少嵌套爭用。
遵循這些原則，99%的多進(jìn)程死鎖問題都能被規(guī)避。新手建議從“with語句+統(tǒng)一鎖順序”這兩個最基礎(chǔ)的策略入手，先保證程序無死鎖，再根據(jù)場景優(yōu)化性能。
以上就是Python多進(jìn)程中避免死鎖問題的六種策略的詳細(xì)內(nèi)容，更多關(guān)于Python多進(jìn)程避免死鎖的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要介紹了Python中幾個比較常見的名詞解釋,本文解釋同樣適應(yīng)其它編程語言,本文講解了循環(huán)、迭代、遞歸、遍歷等名詞的含義,需要的朋友可以參考下2015-07-07
文章講述了在使用PyCharm開發(fā)Flask應(yīng)用時,啟動地址與手動指定的IP端口不一致的問題,通過修改PyCharm的運行配置,將Flask項目的運行模式從Flask模式改為Python模式來解決此問題,需要的朋友可以參考下2025-03-03
這篇文章主要介紹了Python如何批量更改圖像尺寸統(tǒng)一大小問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-03-03
用python寫PDF轉(zhuǎn)換器的實現(xiàn)
這篇文章主要介紹了用python寫PDF轉(zhuǎn)換器的實現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-10-10
在python中獲取div的文本內(nèi)容并和想定結(jié)果進(jìn)行對比詳解
今天小編就為大家分享一篇在python中獲取div的文本內(nèi)容并和想定結(jié)果進(jìn)行對比詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01
Python Django中的STATIC_URL 設(shè)置和使用方式
這篇文章主要介紹了Python Django中的STATIC_URL 設(shè)置和使用方式，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-03-03
這篇文章主要介紹了tensorflow之讀取jpg圖像長和寬實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06
在Python中,抽象基類是一類特殊的類,它不能被實例化,主要用于作為基類被其他子類繼承,本文小編就來和大家深入介紹一下抽象基類的具體使用,有需要的小伙伴可以參考一下2025-06-06
Python爬蟲之Selenium中frame/iframe表單嵌套頁面
這篇文章主要介紹了Python爬蟲之Selenium中frame/iframe表單嵌套頁面，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
詳解Python OpenCV圖像分割算法的實現(xiàn)
圖像分割是指根據(jù)灰度、色彩、空間紋理、幾何形狀等特征把圖像劃分成若干個互不相交的區(qū)域。本文就來和大家聊聊OpenCV的圖像分割算法及基于輪廓的字符分離，感興趣的可以了解一下2022-08-08