Python Watchdog脚本实现子脚本启停与故障自动重启方案问询

需求描述 #

需要编写watchdog.py脚本，用于管理test1.py、test2.py这类以独立循环运行的封闭脚本。核心功能包括：按需启动/停止子脚本，子脚本运行出错时自动重启。此前尝试使用Popen和多线程方案，但未实现有效的监控与重启逻辑，需正确的实现流程及示例代码。

现有方案不足 #

当前在Linux环境下通过subprocess.Popen启动子脚本，用pkill命令终止，但该方案并非最优，需更合理的实现方式。

测试实现结构 #

• watchdog.py：主脚本，通过无限循环按需启动、停止、重启子脚本
• watchdog.ini：配置文件，定义子脚本及启停、重启标记，可在watchdog.py运行时手动或通过其他脚本修改
• test1.py：测试脚本，每10秒打印一次"Hello Test 1"，无运行错误
• test2.py：测试脚本，每10秒打印一次"Hello Test 2"，运行两次后触发错误，可通过配置文件的重启标记触发自动重启

所有文件放在同一目录，运行watchdog.py即可测试，目前缺少子脚本全路径配置功能，可按需修改。

watchdog.py 代码 #

import time
import os
import subprocess
import configparser

# 监控脚本的运行间隔（秒）
interval = 10

# 读取配置文件
def read_config(section, key):
    config = configparser.ConfigParser()
    config.read('watchdog.ini')
    return config[section][key]

# 启动子脚本作为子进程
def start_script(scriptname):
    sub_id = f"sub_id_{scriptname}"
    cmd = ['python', f"{scriptname}.py"]
    globals()[sub_id] = subprocess.Popen(cmd)

# 终止子脚本进程
def stop_script(scriptname):
    command = f'pkill -f {scriptname}.py'
    os.system(command)

# 检查子脚本运行状态
def running_script(scriptname):
    sub_id = f'sub_id_{scriptname}'
    if sub_id in globals():
        proc = globals()[sub_id]
        if proc.poll() is None:
            return 'running'
        else:
            return 'stoped'
    else:
        return 'not defined'

# 主循环
while True:
    config = configparser.ConfigParser()
    config.read('watchdog.ini')
    all_section_names = config.sections()
    
    for section_name in all_section_names:
        # 获取子脚本的启停、重启标记
        start = int(read_config(section_name, 'start'))
        stop = int(read_config(section_name, 'stop'))
        restart = int(read_config(section_name, 'restart'))
        
        # 获取子脚本运行状态
        status = running_script(section_name)
        
        # 启动逻辑：start=1且未启动过
        if start == 1 and status == 'not defined':
            start_script(section_name)
            print(f"{section_name} started")
        
        # 停止逻辑：stop=1且脚本处于运行状态
        if stop == 1 and status != 'stoped' and status != 'not defined':
            stop_script(section_name)
            print(f"{section_name} stoped")
        
        # 重启逻辑：restart=1且脚本已停止
        if restart == 1 and status == 'stoped':
            stop_script(section_name)
            start_script(section_name)
            print(f"{section_name} restarted")
    
    time.sleep(interval)

watchdog.ini 配置 #

[test1]
start = 1
stop = 0
restart = 0

[test2]
start = 1
stop = 0
restart = 1

test1.py 测试脚本 #

# 每10秒打印一次，无运行错误
import time

while True:
    print('Hello Test 1')
    time.sleep(10)

test2.py 测试脚本 #

# 每10秒打印两次后触发错误，用于测试自动重启
import time

while True:
    print("Hello Test 2")
    time.sleep(10)
    print("Hello Test 2")
    time.sleep(10)
    get.value  # 触发未定义变量错误

内容的提问来源于stack exchange，提问作者Phantom001