Python脚本如何排查线程卡死问题

wen python案例 2026-06-27 245

本文目录导读：

Python脚本如何排查线程卡死问题

终极武器：打印所有线程的堆栈信息
针对不同原因的专项排查
自动化监控与诊断脚本
总结排查流程
重要提示

排查Python线程卡死问题，通常需要从检测卡死位置和分析卡死原因两个方向入手，由于Python的GIL（全局解释器锁）机制，纯计算密集型任务通常不会导致线程完全“卡死”，更多是死锁、I/O阻塞、资源竞争或C扩展阻塞。

以下是一个系统性的排查脚本和思路,可以直接运行或集成到项目中。

终极武器：打印所有线程的堆栈信息

当程序看起来卡死时，最直接有效的方法是向进程发送信号，强制打印所有线程的堆栈，这是排查的第一步。

方法A：使用 `faulthandler` 模块（推荐，Python 3.3+）

此方法可以在不终止进程的情况下,将堆栈输出到stderr或日志文件。

import faulthandler
import signal
import time
import threading
import sys
# 启用故障处理程序
faulthandler.enable()
# 可选：当收到 SIGUSR1 信号时，打印所有线程的堆栈（Linux/macOS）
# Windows 不支持 SIGUSR1，可以使用其他信号或定时器
# faulthandler.register(signal.SIGUSR1)
# 模拟一个死锁场景
lock1 = threading.Lock()
lock2 = threading.Lock()
def worker_a():
    with lock1:
        print("Worker A acquired lock1, waiting for lock2...")
        time.sleep(1)
        with lock2:
            pass
def worker_b():
    with lock2:
        print("Worker B acquired lock2, waiting for lock1...")
        time.sleep(1)
        with lock1:
            pass
# 启动线程
t1 = threading.Thread(target=worker_a)
t2 = threading.Thread(target=worker_b)
t1.start()
t2.start()
# 主线程模拟卡死
time.sleep(3)
print("Main thread is stuck. Dumping stack traces...")
# 触发堆栈打印（手动调用）
faulthandler.dump_traceback()

运行效果：
你会看到类似下面的输出,清晰地显示每个线程卡在哪一行代码：

Current thread 0x000000010d5e2000 (most recent call first):
  File "/path/to/script.py", line 18, in worker_a
    with lock2:
  ...
Thread 0x000000010d9ea000 (most recent call first):
  File "/path/to/script.py", line 25, in worker_b
    with lock1:
  ...

方法B：使用 `sys._current_frames()`（无需额外模块）

在可疑的卡死点,通过另一个线程或信号处理函数调用。

import sys
import traceback
import threading
import time
def dump_threads():
    """打印所有线程的堆栈信息"""
    for thread_id, frame in sys._current_frames().items():
        stack = traceback.extract_stack(frame)
        print(f"\n--- Thread ID: {thread_id} ---")
        for filename, lineno, name, line in stack:
            print(f"  File: \"{filename}\", line {lineno}, in {name}")
            if line:
                print(f"    {line.strip()}")
    print("\n--- End of dump ---")
# 启动一个监控线程，每隔5秒检查一次
def monitor():
    while True:
        time.sleep(5)
        # 这里可以增加条件判断，例如检测某个变量是否长时间未更新
        print("[Monitor] Dumping threads...")
        dump_threads()
monitor_thread = threading.Thread(target=monitor, daemon=True)
monitor_thread.start()
# ... 你的业务逻辑 ...

针对不同原因的专项排查

死锁（最常见）

特征：多个线程相互等待对方持有的锁。
排查方法：使用上面的堆栈打印，查看是否存在 Lock.acquire() 或 RLock.acquire() 等待。
自动检测：可以使用 threading_utils 或自定义装饰器来监控锁获取超时。

示例：带有超时的锁获取

import threading
import time
lock = threading.Lock()
def safe_acquire(lock, timeout=5):
    """尝试获取锁，超时则打印警告"""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if lock.acquire(blocking=False):
            return True
        time.sleep(0.1)
    print(f"WARNING: Failed to acquire lock {lock} within {timeout}s")
    return False
# 使用时
# if safe_acquire(some_lock):
#     try:
#         ...
#     finally:
#         some_lock.release()

I/O 阻塞

特征：线程卡在 socket.recv()、file.read()、subprocess.Popen.wait() 等操作上。
排查方法：检查堆栈中是否有 select.select()、recv、read 等调用。
解决方案：
- 设置 Socket 超时：socket.setdefaulttimeout(10)。
- 使用 concurrent.futures.ThreadPoolExecutor 配合 Future.result(timeout=5)。

C 扩展阻塞（释放了 GIL）

特征：堆栈显示程序正在执行 C 扩展代码（如 numpy, pandas, pybind11 等），且 GIL 已释放。
难点：Python 无法中断或探查 C 代码内部,堆栈信息有限。
排查方法：
1. 使用 gdb 附加到进程：gdb python <pid>，然后执行 thread apply all bt。
2. 升级库版本或联系 C 扩展维护者。

线程饥饿（公平性问题）

特征：某个线程逻辑上一直在运行，但始终获取不到执行权（通常是死循环或 GIL 调度问题）。
排查方法：用 dump_threads() 发现该线程状态不是 waiting 或 locked，而是 running,但堆栈显示它在一个循环中。

自动化监控与诊断脚本

整合以上方法,实现一个可复用的监控装饰器或上下文管理器。

import sys
import threading
import traceback
import time
from functools import wraps
class ThreadMonitor:
    """线程监控器，定时检查线程状态"""
    def __init__(self, check_interval=5, dump_on_stuck=True):
        self._check_interval = check_interval
        self._dump_on_stuck = dump_on_stuck
        self._threads_status = {}
        self._running = True
        self._monitor_thread = threading.Thread(target=self._run, daemon=True)
        self._monitor_thread.start()
    def _run(self):
        while self._running:
            time.sleep(self._check_interval)
            current_frames = sys._current_frames()
            # 检查是否有线程长时间卡在同一个位置
            for thread in threading.enumerate():
                if thread is threading.main_thread() or thread is self._monitor_thread:
                    continue
                thread_id = thread.ident
                frame = current_frames.get(thread_id)
                if frame:
                    stack = traceback.extract_stack(frame)
                    if stack:
                        last_frame = stack[-1]
                        key = (last_frame.filename, last_frame.lineno)
                        if thread_id in self._threads_status:
                            last_key = self._threads_status[thread_id]
                            if key == last_key:
                                print(f"[Monitor] Thread {thread.name} seems stuck at {last_frame.filename}:{last_frame.lineno}")
                                if self._dump_on_stuck:
                                    # 打印更详细的堆栈
                                    traceback.print_stack(frame)
                        self._threads_status[thread_id] = key
    def stop(self):
        self._running = False
# 使用示例
monitor = ThreadMonitor(check_interval=2)  # 每2秒检查一次
def long_running_task():
    time.sleep(100)  # 模拟一个看似卡死的线程
t = threading.Thread(target=long_running_task, name="Worker")
t.start()
t.join()

总结排查流程

立即获取堆栈：使用 faulthandler.dump_traceback() 或 sys._current_frames()。
观察卡住位置：
- acquire() -> 死锁。
- recv() / read() / accept() -> I/O 阻塞,检查超时设置。
- wait() (Condition) -> 等待通知未收到。
- join() -> 等待的子线程可能卡住了。
检查资源使用：lsof -p <pid> (Linux) 查看文件描述符；top -H -p <pid> 查看线程 CPU/内存。
使用外部工具：
- gdb + py-bt (如果安装了python-dbg)。
- strace (Linux) 跟踪系统调用。
代码审查：检查是否存在循环等待、忘记释放锁、I/O 未设置超时、线程间通知丢失等问题。