如何写死锁线程定位简易脚本

本文目录导读：

1. Python 版本 - 模拟死锁并检测
2. Bash 版本 - 系统级死锁检测脚本
3. 使用说明
4. 核心功能

我来帮你写一个死锁线程定位的简易脚本,这里提供 Python 和 Bash 两种版本：

Python 版本 - 模拟死锁并检测

#!/usr/bin/env python3
"""
简易死锁检测脚本
"""
import threading
import time
import sys
import traceback
import signal
class DeadlockDetector:
    def __init__(self):
        self.lock1 = threading.Lock()
        self.lock2 = threading.Lock()
        self.threads_info = {}
        self.running = True
    def task1(self):
        """线程1：先获取lock1，再获取lock2"""
        thread_id = threading.current_thread().name
        self.threads_info[thread_id] = "尝试获取 lock1"
        print(f"[{thread_id}] 获取 lock1...")
        with self.lock1:
            self.threads_info[thread_id] = "已获取 lock1，等待 lock2"
            print(f"[{thread_id}] 已获取 lock1，等待获取 lock2...")
            time.sleep(0.1)
            with self.lock2:
                self.threads_info[thread_id] = "已获取 lock1 和 lock2"
                print(f"[{thread_id}] 成功获取所有锁")
    def task2(self):
        """线程2：先获取lock2，再获取lock1（造成死锁）"""
        thread_id = threading.current_thread().name
        self.threads_info[thread_id] = "尝试获取 lock2"
        print(f"[{thread_id}] 获取 lock2...")
        with self.lock2:
            self.threads_info[thread_id] = "已获取 lock2，等待 lock1"
            print(f"[{thread_id}] 已获取 lock2，等待获取 lock1...")
            time.sleep(0.1)
            with self.lock1:
                self.threads_info[thread_id] = "已获取 lock1 和 lock2"
                print(f"[{thread_id}] 成功获取所有锁")
    def detect_deadlock(self):
        """检测死锁状态"""
        print("\n=== 死锁检测 ===")
        print("当前线程状态:")
        for thread_id, status in self.threads_info.items():
            print(f"  {thread_id}: {status}")
        # 检查是否有线程长时间等待
        deadlocked = False
        for thread_id, status in self.threads_info.items():
            if "等待" in status:
                print(f"  ⚠️ {thread_id} 可能处于死锁状态")
                deadlocked = True
        if deadlocked:
            print("\n🚨 检测到可能的死锁！")
            print("解决方案建议:")
            print("1. 统一锁的获取顺序")
            print("2. 使用 try_lock() 设置超时")
            print("3. 使用 threading.Lock() 的 acquire(timeout)")
        else:
            print("\n✅ 当前未检测到死锁")
        print("=" * 30)
    def run_with_deadlock(self):
        """模拟死锁场景"""
        print("开始模拟死锁场景...\n")
        t1 = threading.Thread(target=self.task1, name="线程1")
        t2 = threading.Thread(target=self.task2, name="线程2")
        t1.start()
        t2.start()
        # 设置超时检测
        timeout = 2
        t1.join(timeout=timeout)
        t2.join(timeout=timeout)
        if t1.is_alive() or t2.is_alive():
            print(f"\n⏰ 线程在 {timeout} 秒后仍未完成")
            self.detect_deadlock()
        else:
            print("\n✅ 所有线程正常完成")
        # 强制结束（仅为演示）
        self.running = False
def simple_deadlock_detector():
    """简易死锁检测器（不模拟死锁，只提供检测功能）"""
    print("简易死锁检测器")
    print("-" * 30)
    print("检测方法:")
    print("1. 检查线程状态")
    print("2. 分析锁的获取顺序")
    print("3. 检测循环等待条件")
    print()
    # 获取当前所有线程信息
    for thread in threading.enumerate():
        print(f"线程: {thread.name}")
        print(f"  活动状态: {thread.is_alive()}")
        print(f"  是否是守护线程: {thread.daemon}")
        print()
if __name__ == "__main__":
    print("选择运行模式:")
    print("1. 死锁模拟与检测")
    print("2. 简单线程状态检测")
    choice = input("请输入选择 (1/2): ")
    if choice == "1":
        detector = DeadlockDetector()
        detector.run_with_deadlock()
    else:
        simple_deadlock_detector()

Bash 版本 - 系统级死锁检测脚本

#!/bin/bash
# 系统级死锁检测脚本
echo "=== 系统死锁检测脚本 ==="
echo "时间: $(date)"
echo ""
# 检测系统负载
check_system_load() {
    echo "1. 系统负载检测:"
    uptime
    echo ""
}
# 检测进程状态
check_process_status() {
    echo "2. 进程状态检测:"
    echo "  正在运行的进程数: $(ps aux | wc -l)"
    echo "  僵尸进程: $(ps aux | grep -c "Z")"
    echo ""
}
# 检测线程D状态（不可中断睡眠，可能是死锁）
check_d_state() {
    echo "3. 检测D状态线程（可能死锁）:"
    ps aux | head -1
    ps aux | grep " D" 
    if [ $? -ne 0 ]; then
        echo "  未发现D状态线程"
    fi
    echo ""
}
# 检测文件锁
check_file_locks() {
    echo "4. 文件锁检测:"
    lsof 2>/dev/null | head -5
    echo "  总文件锁数: $(lsof 2>/dev/null | wc -l)"
    echo ""
}
# 检测特定进程的线程
check_process_threads() {
    echo "5. 请选择要检测的进程:"
    echo "  正在运行的Java进程:"
    jps 2>/dev/null || ps aux | grep java | grep -v grep
    read -p "  输入进程PID (或按Enter跳过): " pid
    if [ ! -z "$pid" ]; then
        echo "  进程 $pid 的线程信息:"
        if [ -d /proc/$pid/task ]; then
            echo "  线程数: $(ls /proc/$pid/task | wc -l)"
            echo "  线程列表:"
            for tid in $(ls /proc/$pid/task); do
                status=$(cat /proc/$pid/task/$tid/status 2>/dev/null | grep "State" | head -1)
                echo "    TID: $tid - $status"
            done
        fi
    fi
    echo ""
}
# 使用strace检测死锁（需要root权限）
check_with_strace() {
    echo "6. strace检测（需要root权限）:"
    if [ "$EUID" -eq 0 ]; then
        read -p "  输入要跟踪的PID: " pid
        if [ ! -z "$pid" ]; then
            echo "  执行 strace -p $pid -e trace=lock 3秒..."
            timeout 3 strace -p $pid -e trace=lock 2>&1 | head -20
        fi
    else
        echo "  需要root权限执行strace检测"
        echo "  请使用: sudo $0"
    fi
    echo ""
}
# 模拟死锁（用于测试）
simulate_deadlock() {
    echo "可选: 创建测试死锁场景"
    read -p "是否创建测试死锁? (y/n): " create
    if [ "$create" = "y" ]; then
        echo "创建死锁测试..."
        python3 << 'EOF'
import threading
import time
class TestDeadlock:
    def __init__(self):
        self.lock1 = threading.Lock()
        self.lock2 = threading.Lock()
    def thread1(self):
        with self.lock1:
            time.sleep(0.1)
            with self.lock2:
                print("Thread 1 done")
    def thread2(self):
        with self.lock2:
            time.sleep(0.1)
            with self.lock1:
                print("Thread 2 done")
test = TestDeadlock()
t1 = threading.Thread(target=test.thread1)
t2 = threading.Thread(target=test.thread2)
t1.start()
t2.start()
print("死锁测试线程已启动，PID:", threading.current_thread().ident)
time.sleep(5)  # 保持运行
EOF
    fi
}
# 主函数
main() {
    echo "═══════════════════════════════"
    echo "    系统死锁检测工具"
    echo "═══════════════════════════════"
    echo ""
    check_system_load
    check_process_status
    check_d_state
    check_file_locks
    read -p "是否进行详细检测? (y/n): " detail
    if [ "$detail" = "y" ]; then
        check_process_threads
        check_with_strace
    fi
    read -p "是否创建测试死锁? (y/n): " test
    if [ "$test" = "y" ]; then
        simulate_deadlock
    fi
    echo ""
    echo "检测完成!"
    echo "提示: 如果发现死锁，考虑以下解决方案:"
    echo "1. 重启相关进程"
    echo "2. 检查并修复锁的顺序"
    echo "3. 增加超时机制"
    echo "4. 使用更高级的死锁检测工具如 Valgrind"
}
main

使用说明

Python 版本：

# 运行
python3 deadlock_detector.py

Bash 版本：

# 添加执行权限
chmod +x deadlock_detector.sh
# 运行（部分功能需要root）
sudo ./deadlock_detector.sh

核心功能

1. 死锁模拟：演示经典的死锁场景
2. 线程状态检测：监控线程等待状态
3. 资源锁分析：检测锁的获取顺序
4. 系统级检测：监控D状态线程和文件锁

这个脚本可以帮助你快速定位和诊断死锁问题。