Python多进程中避免死锁问题的六种策略
更新时间：2025年12月09日 08:26:08 作者：三维空间
死锁是Python多进程开发中最容易踩的坑之一,一旦出现会导致进程卡死、程序无响应,甚至需要强制终止才能恢复,所以本文给大家介绍了Python多进程中避免死锁问题的六种策略,需要的朋友可以参考下
一、先搞懂：多进程死锁到底是什么？
1. 死锁的核心定义
死锁是指两个或多个进程互相持有对方需要的锁（或资源），且都不释放自己持有的锁，导致所有进程都陷入“等待对方释放资源”的无限阻塞状态。
2. 死锁产生的4个必要条件（缺一不可）
只有同时满足以下4个条件，才会触发死锁，打破任意一个条件就能避免死锁：
- 互斥条件：资源（如锁、文件）只能被一个进程占用；
- 请求与保持条件：进程持有一个资源的同时，又请求另一个被占用的资源；
- 不剥夺条件：进程已持有的资源不能被强制剥夺，只能主动释放；
- 循环等待条件：多个进程形成“你等我、我等你”的循环等待链。
3. 多进程死锁典型场景（新手高频踩坑）
import multiprocessing
import time
场景：两个进程互相等待对方的锁
def process1(lock1, lock2):
进程1先拿lock1，再尝试拿lock2
lock1.acquire()
print("进程1获取了lock1，等待lock2...")
time.sleep(1) 放大死锁概率
lock2.acquire() 此时lock2已被进程2持有，进程1阻塞
print("进程1获取了lock2，执行任务")
lock1.release()
lock2.release()
def process2(lock1, lock2):
进程2先拿lock2，再尝试拿lock1
lock2.acquire()
print("进程2获取了lock2，等待lock1...")
time.sleep(1)
lock1.acquire() 此时lock1已被进程1持有，进程2阻塞
print("进程2获取了lock1，执行任务")
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("程序结束") 永远不会执行到这一行
现象：程序输出“进程1获取了lock1，等待lock2...”和“进程2获取了lock2，等待lock1...”后卡死，无法继续执行。
二、避免死锁的6个核心策略（从易到难）
1. 策略1：使用with语句自动释放锁（最推荐）
with语句会在代码块执行完成后自动释放锁，即使代码抛出异常也能保证锁释放，从根本上避免“忘记release()”导致的死锁。
修复上述死锁案例（仅改锁的使用方式）：
def process1(lock1, lock2):
with lock1: 自动acquire()，代码块结束自动release()
print("进程1获取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("进程1获取了lock2，执行任务")
def process2(lock1, lock2):
with lock2:
print("进程2获取了lock2，等待lock1...")
time.sleep(1)
with lock1:
print("进程2获取了lock1，执行任务")
注意：这个修改仅解决“锁未释放”的问题，但上述场景仍会因“循环等待”触发死锁，需结合策略2使用。
2. 策略2：统一锁的获取顺序（打破循环等待）
死锁的核心诱因之一是“进程获取锁的顺序不一致”，只要让所有进程按相同的顺序获取锁，就能打破循环等待条件。
最终修复死锁案例：
def process1(lock1, lock2):
统一先拿lock1，再拿lock2
with lock1:
print("进程1获取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("进程1获取了lock2，执行任务")
def process2(lock1, lock2):
同样先拿lock1，再拿lock2（不再先拿lock2）
with lock1:
print("进程2获取了lock1，等待lock2...")
time.sleep(1)
with lock2:
print("进程2获取了lock2，执行任务")
执行结果：进程1先获取lock1，进程2等待lock1；进程1执行完成释放锁后，进程2获取lock1和lock2，无死锁。
3. 策略3：给锁添加超时时间（避免无限等待）
multiprocessing.Lock本身不支持超时，但可使用multiprocessing.RLock（可重入锁）或threading.Lock（多进程中需通过Manager传递）的acquire(timeout)方法，设置超时时间，超时后放弃获取锁，避免无限阻塞。
示例：带超时的锁获取
import multiprocessing
import time
def process1(lock1, lock2):
获取lock1，超时时间3秒
if lock1.acquire(timeout=3):
print("进程1获取了lock1，等待lock2...")
time.sleep(1)
获取lock2，超时时间3秒
if lock2.acquire(timeout=3):
print("进程1获取了lock2，执行任务")
lock2.release()
else:
print("进程1获取lock2超时，释放lock1")
lock1.release()
else:
print("进程1获取lock1超时")
def process2(lock1, lock2):
if lock2.acquire(timeout=3):
print("进程2获取了lock2，等待lock1...")
time.sleep(1)
if lock1.acquire(timeout=3):
print("进程2获取了lock1，执行任务")
lock1.release()
else:
print("进程2获取lock1超时，释放lock2")
lock2.release()
else:
print("进程2获取lock2超时")
if __name__ == "__main__":
通过Manager创建支持超时的锁
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("程序结束")
执行结果：进程1获取lock1，进程2获取lock2；双方等待对方的锁超时后，释放自己持有的锁，程序正常结束，无死锁。
4. 策略4：减少锁的使用范围（最小化锁持有时间）
只在“必须保护共享资源”的代码块加锁，执行完核心逻辑后立即释放锁，缩短锁的持有时间，降低多个进程同时争用锁的概率。
反面案例（锁持有时间过长）：
def write_file(num, lock):
lock.acquire()
非核心逻辑（耗时），却持有锁
time.sleep(5) 模拟耗时操作
with open("test.txt", "a") as f:
f.write(f"进程{num}写入\n")
lock.release()
正面案例（仅核心逻辑加锁）：
def write_file(num, lock):
非核心逻辑（耗时），不持有锁
time.sleep(5)
仅写入文件时加锁
with lock:
with open("test.txt", "a") as f:
f.write(f"进程{num}写入\n")
5. 策略5：使用单锁替代多锁（简化资源竞争）
如果多个锁的作用是保护同一类共享资源（如多个文件都属于“数据文件”），可合并为一个全局锁，避免多锁嵌套导致的死锁。
示例：单锁替代多锁
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"进程{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"进程{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：使用死锁检测工具（进阶）
对于复杂的多进程场景，可通过工具检测潜在的死锁：
- 内置工具：
multiprocessing.active_children()查看活跃进程，结合日志定位阻塞的进程； - 第三方工具：
py-spy（进程分析工具），可实时查看进程调用栈，定位死锁位置：
安装py-spy pip install py-spy 分析进程（替换为你的进程ID） py-spy dump --pid 12345
三、死锁排查：快速定位问题的3个方法
- 查看进程状态：用
ps -ef | grep python查看进程是否处于“D状态”（不可中断睡眠，大概率死锁）； - 添加日志：在每个
acquire()和release()前后打印日志，定位哪个锁导致阻塞； - 简化场景：将复杂的多进程逻辑拆分为最小可复现的demo，逐步排查死锁诱因。
四、总结：避免死锁的核心原则
- 自动释放：优先用
with语句管理锁，杜绝“忘记release()”； - 顺序一致：所有进程按相同顺序获取多把锁，打破循环等待；
- 超时兜底：给锁添加超时时间，避免无限阻塞；
- 最小持有：仅在核心逻辑加锁，缩短锁的持有时间；
- 简化锁结构：能用单锁就不用多锁，减少嵌套争用。
遵循这些原则，99%的多进程死锁问题都能被规避。新手建议从“with语句+统一锁顺序”这两个最基础的策略入手，先保证程序无死锁，再根据场景优化性能。
以上就是Python多进程中避免死锁问题的六种策略的详细内容，更多关于Python多进程避免死锁的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了Python装饰器用法,结合实例形式对比分析了Python装饰器的定义与使用技巧,需要的朋友可以参考下2017-07-07
这篇文章主要介绍了Python中的闭包总结,本文讲解了闭包的概念、为什么使用闭包、使用闭包实例等内容,需要的朋友可以参考下2014-09-09
这篇文章主要为大家详细介绍了如何使用Python实现获取当前脚本目录的绝对路径,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下2025-11-11
在本篇文章里小编给大家整理了关于python3代码输出嵌套式对象实例详解内容，有兴趣的朋友们可以学习下。2020-12-12
这篇文章主要为大家介绍了Python变量的作用域，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助2021-12-12
字符串是一个很长的话题，纵然现在开始第三部分，但是也不能完全说尽。因为字符串是自然语言中最复杂的东西，也是承载功能最多的，计算机高级语言编程，要解决自然语言中的问题，让自然语言中完成的事情在计算机上完成，所以，也不得不有更多的话题。2014-09-09
今天小编就为大家分享一篇解决pandas展示数据输出时列名不能对齐的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2019-11-11
使用qt quick-ListView仿微信好友列表和聊天列表的示例代码
本文以微信好友列表为例给大家学习listview的相关知识，通过实例demo给大家详解qt quick-ListView仿微信好友列表和聊天列表的实现方法，需要的朋友参考下吧2021-06-06
下面小编就为大家分享一篇python实现校园网自动登录的示例讲解，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-04-04
今天小编就为大家分享一篇关于Python3转换html到pdf的不同解决方案，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧2019-03-03
最新评论