Python多线程程序运行时，线程阻塞是较为常见的问题，可能导致程序响应变慢、任务无法推进甚至整体卡死。排查线程阻塞的核心目标是找到线程被卡住的具体位置，明确阻塞产生的原因，进而针对性优化。
线程阻塞的常见原因
在排查之前，先了解常见的阻塞诱因，能缩小排查范围：
- 等待I/O操作完成，比如文件读写、网络请求、数据库查询等
- 获取不到锁资源，多个线程竞争同一把锁时出现等待
- 调用了阻塞式的函数，比如
time.sleep()、threading.Event.wait()等 - 线程间死锁，多个线程互相持有对方需要的资源且不释放
基础排查方法：查看线程状态与堆栈
使用threading模块获取线程信息
Python内置的threading模块可以获取当前所有活跃线程的基本信息，先判断是否有线程处于长时间无响应的状态。
import threading
import time
def task():
# 模拟阻塞操作
time.sleep(100)
# 创建并启动线程
t = threading.Thread(target=task, name="block_task")
t.start()
# 遍历所有活跃线程，打印线程名称和状态
for thread in threading.enumerate():
print(f"线程名称: {thread.name}, 是否存活: {thread.is_alive()}")
通过sys._current_frames获取堆栈信息
要定位具体的阻塞位置，需要获取线程的调用堆栈，sys模块的_current_frames方法可以返回每个线程当前的堆栈帧，从而看到线程执行到哪一行代码。
import sys
import threading
import time
def block_func():
# 模拟阻塞的I/O操作
with open("test.txt", "r") as f:
f.read()
def monitor_threads():
# 每隔2秒打印所有线程的堆栈
while True:
print("===== 当前线程堆栈信息 =====")
for thread_id, frame in sys._current_frames().items():
# 找到对应的线程对象
thread = None
for t in threading.enumerate():
if t.ident == thread_id:
thread = t
break
thread_name = thread.name if thread else f"未知线程{thread_id}"
print(f"线程: {thread_name}")
# 遍历堆栈帧，打印调用路径
while frame:
filename = frame.f_code.co_filename
lineno = frame.f_lineno
func_name = frame.f_code.co_name
print(f" 文件: {filename}, 行号: {lineno}, 函数: {func_name}")
frame = frame.f_back
time.sleep(2)
# 启动阻塞线程和监控线程
t1 = threading.Thread(target=block_func, name="io_block_thread")
t2 = threading.Thread(target=monitor_threads, name="monitor_thread")
t1.start()
t2.start()
进阶排查：使用专业工具分析
py-spy工具采样分析
py-spy是一款轻量的Python性能分析工具，不需要修改代码就可以对运行中的Python程序进行采样，查看每个线程的执行状态，非常适合排查阻塞问题。
安装py-spy后，使用以下命令查看指定进程的所有线程状态：
# 假设目标Python进程ID为12345 py-spy dump --pid 12345
输出结果中会显示每个线程的调用栈，若某个线程的调用栈长时间停留在同一行，基本可以确定该位置就是阻塞点。
使用traceback模块打印异常堆栈
如果线程因为异常阻塞，或者想在指定位置主动打印堆栈，可以使用traceback模块：
import traceback
import threading
import time
def task():
try:
time.sleep(10)
except Exception as e:
# 打印异常堆栈
traceback.print_exc()
finally:
# 主动打印当前线程的堆栈
traceback.print_stack()
t = threading.Thread(target=task)
t.start()
t.join()
死锁场景的排查方法
死锁是线程阻塞的特殊场景，排查时可以重点检查锁的获取顺序：
import threading
import time
lock_a = threading.Lock()
lock_b = threading.Lock()
def task1():
# 先获取lock_a，再尝试获取lock_b
with lock_a:
print("task1获取lock_a")
time.sleep(1)
with lock_b:
print("task1获取lock_b")
def task2():
# 先获取lock_b，再尝试获取lock_a，形成死锁
with lock_b:
print("task2获取lock_b")
time.sleep(1)
with lock_a:
print("task2获取lock_a")
t1 = threading.Thread(target=task1)
t2 = threading.Thread(target=task2)
t1.start()
t2.start()
t1.join()
t2.join()
对于这类死锁问题，使用前面的堆栈分析方法，会看到两个线程分别卡在获取对方持有的锁的位置，此时调整锁的获取顺序，保证所有线程按相同顺序获取锁即可解决。
排查注意事项
- 排查时尽量在测试环境复现问题，避免影响生产环境程序运行
- 采样间隔不要设置太短，避免采样本身影响程序性能
- 若阻塞是偶发的，可以增加监控频率，记录阻塞发生的时间点和对应的堆栈信息
- 对于I/O类阻塞，可以检查对应的外部资源是否正常，比如网络是否通畅、文件是否存在等