本文目录导读：
我来帮你写一个死锁线程定位的简易脚本,这里提供 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
核心功能
- 死锁模拟：演示经典的死锁场景
- 线程状态检测：监控线程等待状态
- 资源锁分析：检测锁的获取顺序
- 系统级检测：监控D状态线程和文件锁
这个脚本可以帮助你快速定位和诊断死锁问题。
