Python脚本如何重启失败定时任务

本文目录导读：

1. 通用定时任务管理器脚本
2. 更简单的守护进程版本
3. 使用系统定时任务检查
4. 使用supervisor替代方案（推荐）
5. 使用建议

我可以帮你编写一个用于重启失败定时任务的Python脚本,这里提供几种常见场景的解决方案：

通用定时任务管理器脚本

import subprocess
import time
import logging
import sys
from datetime import datetime
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('task_restart.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
class TaskRestarter:
    def __init__(self, task_name, command, check_interval=60, max_restarts=3):
        """
        初始化任务重启器
        :param task_name: 任务名称
        :param command: 启动任务的命令
        :param check_interval: 检查间隔（秒）
        :param max_restarts: 最大重启次数
        """
        self.task_name = task_name
        self.command = command
        self.check_interval = check_interval
        self.max_restarts = max_restarts
        self.restart_count = 0
        self.process = None
    def start_task(self):
        """启动任务"""
        try:
            logging.info(f"正在启动任务: {self.task_name}")
            self.process = subprocess.Popen(
                self.command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            logging.info(f"任务启动成功，PID: {self.process.pid}")
            return True
        except Exception as e:
            logging.error(f"启动任务失败: {e}")
            return False
    def check_task_status(self):
        """检查任务运行状态"""
        if self.process is None:
            return False
        # 检查进程是否还在运行
        poll = self.process.poll()
        if poll is not None:
            logging.warning(f"任务已停止，退出码: {poll}")
            # 获取输出信息
            stdout, stderr = self.process.communicate()
            if stdout:
                logging.info(f"标准输出: {stdout.decode('utf-8', errors='ignore')}")
            if stderr:
                logging.error(f"错误输出: {stderr.decode('utf-8', errors='ignore')}")
            return False
        return True
    def restart_task(self):
        """重启任务"""
        if self.restart_count >= self.max_restarts:
            logging.error(f"任务 {self.task_name} 已达到最大重启次数 {self.max_restarts}")
            return False
        try:
            # 先停止旧任务
            if self.process and self.process.poll() is None:
                logging.info(f"正在停止旧任务...")
                self.process.terminate()
                try:
                    self.process.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    logging.warning("强制终止任务")
                    self.process.kill()
            # 等待一下再重启
            time.sleep(2)
            # 重启任务
            self.restart_count += 1
            logging.info(f"第 {self.restart_count} 次尝试重启任务")
            return self.start_task()
        except Exception as e:
            logging.error(f"重启任务失败: {e}")
            return False
    def run(self):
        """主运行循环"""
        logging.info(f"开始监控任务: {self.task_name}")
        # 初始启动
        if not self.start_task():
            logging.error("初始启动失败")
            return
        try:
            while True:
                # 检查任务状态
                if not self.check_task_status():
                    # 任务失败，尝试重启
                    if not self.restart_task():
                        logging.error("无法重启任务，退出监控")
                        break
                    else:
                        # 重启成功后重置计数器（可选）
                        # self.restart_count = 0
                        pass
                # 等待检查间隔
                time.sleep(self.check_interval)
        except KeyboardInterrupt:
            logging.info("收到中断信号，正在停止...")
        finally:
            # 清理资源
            if self.process and self.process.poll() is None:
                self.process.terminate()
                self.process.wait(timeout=5)
                logging.info("任务已停止")
# 使用示例
if __name__ == "__main__":
    # 示例1: 重启Python脚本
    restarter = TaskRestarter(
        task_name="my_python_task",
        command="python3 /path/to/your/script.py",
        check_interval=30,  # 每30秒检查一次
        max_restarts=5      # 最多重启5次
    )
    restarter.run()

更简单的守护进程版本

import subprocess
import time
import os
import signal
import sys
def run_with_restart(command, max_restarts=3, check_interval=10):
    """
    运行命令并在失败时重启
    :param command: 要运行的命令
    :param max_restarts: 最大重启次数
    :param check_interval: 检查间隔（秒）
    """
    restart_count = 0
    while restart_count < max_restarts:
        try:
            print(f"启动命令: {command}")
            print(f"启动时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
            # 启动进程
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                universal_newlines=True
            )
            # 实时输出日志
            for line in process.stdout:
                print(line, end='')
            # 等待进程结束
            process.wait()
            # 检查退出码
            if process.returncode == 0:
                print("任务正常结束")
                break
            else:
                print(f"任务异常退出，退出码: {process.returncode}")
        except KeyboardInterrupt:
            print("\n收到中断信号，退出监控")
            process.terminate()
            break
        except Exception as e:
            print(f"运行出错: {e}")
        # 重启逻辑
        restart_count += 1
        if restart_count < max_restarts:
            print(f"将在 {check_interval} 秒后尝试第 {restart_count} 次重启...")
            time.sleep(check_interval)
        else:
            print(f"已达到最大重启次数 {max_restarts}，退出")
# 使用示例
if __name__ == "__main__":
    # 监控并重启Python脚本
    run_with_restart(
        command="python3 your_script.py --some-argument",
        max_restarts=5,
        check_interval=10
    )

使用系统定时任务检查

import subprocess
import json
from datetime import datetime
def check_cron_status():
    """检查cron任务状态"""
    try:
        # 查看cron日志
        result = subprocess.run(
            ['grep', 'CRON', '/var/log/syslog'],
            capture_output=True,
            text=True,
            timeout=10
        )
        # 提取最近的任务执行记录
        lines = result.stdout.split('\n')
        recent_failures = []
        for line in lines[-100:]:  # 检查最近100行
            if 'FAILED' in line or 'error' in line.lower():
                recent_failures.append(line)
        return recent_failures
    except Exception as e:
        print(f"检查cron状态失败: {e}")
        return []
def restart_failed_cron_jobs():
    """重启失败的cron任务"""
    print(f"检查时间: {datetime.now()}")
    # 检查失败的cron任务
    failures = check_cron_status()
    if failures:
        print(f"发现 {len(failures)} 个失败的任务:")
        for failure in failures[:5]:  # 只显示前5个
            print(f"  - {failure}")
        # 重启cron服务
        try:
            print("正在重启cron服务...")
            subprocess.run(['systemctl', 'restart', 'cron'], check=True)
            print("cron服务重启成功")
        except subprocess.CalledProcessError:
            print("重启cron服务失败，尝试使用service命令...")
            try:
                subprocess.run(['service', 'cron', 'restart'], check=True)
                print("cron服务重启成功")
            except Exception as e:
                print(f"重启cron服务失败: {e}")
    else:
        print("未发现失败的任务")
if __name__ == "__main__":
    restart_failed_cron_jobs()

使用supervisor替代方案（推荐）

虽然这不是纯Python方案,但配合使用更简单：

import subprocess
import requests
import json
def restart_supervisor_process(process_name, supervisor_url='http://localhost:9001'):
    """
    通过Supervisor API重启进程
    """
    try:
        # 使用supervisorctl命令
        result = subprocess.run(
            ['supervisorctl', 'restart', process_name],
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode == 0:
            print(f"进程 {process_name} 重启成功")
            return True
        else:
            print(f"重启失败: {result.stderr}")
            return False
    except Exception as e:
        print(f"重启过程出错: {e}")
        return False
def monitor_and_restart(process_list, check_interval=60):
    """
    监控并重启特定进程
    """
    import time
    while True:
        for process in process_list:
            # 检查进程状态
            result = subprocess.run(
                ['supervisorctl', 'status', process],
                capture_output=True,
                text=True
            )
            if 'RUNNING' not in result.stdout:
                print(f"进程 {process} 不在运行状态，尝试重启...")
                restart_supervisor_process(process)
        time.sleep(check_interval)
# 使用示例
if __name__ == "__main__":
    monitor_and_restart(['your_task_name1', 'your_task_name2'])

使用建议

1. 对于简单的Python脚本：使用第一种方案，直接在脚本内部重启
2. 对于生产环境：推荐使用supervisor或systemd管理
3. 设置合理的重启次数：避免无限重启导致资源浪费
4. 添加告警通知：在多次重启失败后发送告警
5. 记录详细日志：便于问题定位

选择哪种方案取决于你的具体需求和使用场景。