IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen python案例 246

本文目录导读：

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. 记录详细日志：便于问题定位

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

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python脚本如何删除过期定时任务

下一篇Python脚本如何记录定时任务日志

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝