如何写定时任务管理脚本

本文目录导读：

1. Linux Crontab 管理脚本
2. Python 定时任务管理脚本
3. 使用示例
4. Docker 版定时任务管理
5. 注意事项

我来介绍几种常见的定时任务管理脚本写法,涵盖Linux和跨平台方案。

Linux Crontab 管理脚本

基础任务管理脚本

#!/bin/bash
# 定时任务管理脚本
# crontab_manager.sh
CRON_DIR="${HOME}/cron_jobs"
LOG_DIR="${HOME}/cron_logs"
# 初始化目录
init_dirs() {
    mkdir -p "$CRON_DIR" "$LOG_DIR"
}
# 添加定时任务
add_job() {
    local schedule="$1"
    local command="$2"
    local job_name="$3"
    # 生成任务文件
    local job_file="${CRON_DIR}/${job_name}.sh"
    echo "#!/bin/bash" > "$job_file"
    echo "$command" >> "$job_file"
    chmod +x "$job_file"
    # 添加到crontab
    (crontab -l 2>/dev/null; echo "$schedule ${job_file} >> ${LOG_DIR}/${job_name}.log 2>&1") | crontab -
    echo "✅ 任务 $job_name 已添加"
}
# 列出所有任务
list_jobs() {
    echo "📋 当前定时任务列表："
    crontab -l 2>/dev/null || echo "暂无定时任务"
}
# 删除任务
remove_job() {
    local job_name="$1"
    local job_file="${CRON_DIR}/${job_name}.sh"
    if [ -f "$job_file" ]; then
        rm "$job_file"
        # 从crontab中删除相关行
        crontab -l | grep -v "$job_file" | crontab -
        echo "✅ 任务 $job_name 已删除"
    else
        echo "❌ 任务 $job_name 不存在"
    fi
}
# 启用/禁用任务
toggle_job() {
    local job_name="$1"
    local action="$2" # enable/disable
    local job_file="${CRON_DIR}/${job_name}.sh"
    if [ ! -f "$job_file" ]; then
        echo "❌ 任务 $job_name 不存在"
        return 1
    fi
    case "$action" in
        enable)
            chmod +x "$job_file"
            echo "✅ 任务 $job_name 已启用"
            ;;
        disable)
            chmod -x "$job_file"
            echo "⏸️  任务 $job_name 已禁用"
            ;;
    esac
}
# 查看任务日志
view_log() {
    local job_name="$1"
    local log_file="${LOG_DIR}/${job_name}.log"
    if [ -f "$log_file" ]; then
        tail -f "$log_file"
    else
        echo "❌ 日志文件不存在"
    fi
}
# 主菜单
main_menu() {
    echo "======================="
    echo "   定时任务管理器"
    echo "======================="
    echo "1. 添加任务"
    echo "2. 查看任务列表"
    echo "3. 删除任务"
    echo "4. 启用/禁用任务"
    echo "5. 查看任务日志"
    echo "6. 退出"
    echo "======================="
    read -p "请选择操作: " choice
    case $choice in
        1)
            read -p "任务名称: " name
            read -p "执行时间(如: */5 * * * *): " schedule
            read -p "执行命令: " cmd
            add_job "$schedule" "$cmd" "$name"
            ;;
        2)
            list_jobs
            ;;
        3)
            read -p "要删除的任务名称: " name
            remove_job "$name"
            ;;
        4)
            read -p "任务名称: " name
            read -p "操作(enable/disable): " action
            toggle_job "$name" "$action"
            ;;
        5)
            read -p "任务名称: " name
            view_log "$name"
            ;;
        6)
            exit 0
            ;;
        *)
            echo "❌ 无效选择"
            ;;
    esac
    main_menu
}
# 初始化
init_dirs
# 根据参数执行
case "$1" in
    add)
        add_job "$2" "$3" "$4"
        ;;
    list)
        list_jobs
        ;;
    remove)
        remove_job "$2"
        ;;
    log)
        view_log "$2"
        ;;
    *)
        main_menu
        ;;
esac

Python 定时任务管理脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
定时任务管理系统
支持: 一次性任务、周期性任务、依赖任务
"""
import json
import os
import sys
import time
import logging
import threading
from datetime import datetime, timedelta
import schedule
import subprocess
from typing import Dict, List, Optional
class TaskManager:
    """定时任务管理器"""
    def __init__(self, config_file: str = "tasks.json"):
        self.config_file = config_file
        self.tasks = {}
        self.running = False
        self.setup_logging()
        self.load_tasks()
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('task_manager.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('TaskManager')
    def load_tasks(self):
        """加载任务配置"""
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r') as f:
                self.tasks = json.load(f)
        else:
            self.tasks = {
                "tasks": [],
                "settings": {
                    "max_retries": 3,
                    "retry_delay": 60
                }
            }
            self.save_tasks()
    def save_tasks(self):
        """保存任务配置"""
        with open(self.config_file, 'w') as f:
            json.dump(self.tasks, f, indent=2, ensure_ascii=False)
    def add_task(self, name: str, command: str, schedule_type: str, 
                 schedule_value: str, description: str = ""):
        """添加任务"""
        task = {
            "id": len(self.tasks["tasks"]) + 1,
            "name": name,
            "command": command,
            "schedule_type": schedule_type,  # cron, interval, once, daily
            "schedule_value": schedule_value,
            "description": description,
            "enabled": True,
            "last_run": None,
            "next_run": None,
            "status": "pending"
        }
        self.tasks["tasks"].append(task)
        self.save_tasks()
        self.logger.info(f"任务 {name} 添加成功")
        return task
    def remove_task(self, task_id: int):
        """删除任务"""
        self.tasks["tasks"] = [t for t in self.tasks["tasks"] if t["id"] != task_id]
        self.save_tasks()
        self.logger.info(f"任务 {task_id} 已删除")
    def update_task(self, task_id: int, updates: Dict):
        """更新任务"""
        for task in self.tasks["tasks"]:
            if task["id"] == task_id:
                task.update(updates)
                self.save_tasks()
                self.logger.info(f"任务 {task_id} 已更新")
                return task
        return None
    def toggle_task(self, task_id: int):
        """启用/禁用任务"""
        for task in self.tasks["tasks"]:
            if task["id"] == task_id:
                task["enabled"] = not task["enabled"]
                self.save_tasks()
                status = "启用" if task["enabled"] else "禁用"
                self.logger.info(f"任务 {task_id} 已{status}")
                return task
        return None
    def execute_task(self, task: Dict):
        """执行任务"""
        self.logger.info(f"开始执行任务: {task['name']}")
        task["last_run"] = datetime.now().isoformat()
        task["status"] = "running"
        try:
            # 执行命令
            result = subprocess.run(
                task["command"],
                shell=True,
                capture_output=True,
                text=True,
                timeout=3600
            )
            if result.returncode == 0:
                task["status"] = "success"
                self.logger.info(f"任务 {task['name']} 执行成功")
                if result.stdout:
                    self.logger.debug(f"输出: {result.stdout}")
            else:
                task["status"] = "failed"
                self.logger.error(f"任务 {task['name']} 执行失败: {result.stderr}")
        except subprocess.TimeoutExpired:
            task["status"] = "timeout"
            self.logger.error(f"任务 {task['name']} 执行超时")
        except Exception as e:
            task["status"] = "error"
            self.logger.error(f"任务 {task['name']} 执行出错: {str(e)}")
        self.save_tasks()
        return task
    def schedule_task(self, task: Dict):
        """调度任务"""
        if not task["enabled"]:
            return
        name = task["name"]
        command = lambda: self.execute_task(task)
        if task["schedule_type"] == "interval":
            # 间隔执行: "30" 表示每30秒
            getattr(schedule.every(int(task["schedule_value"])).seconds, "do")(command)
        elif task["schedule_type"] == "minutes":
            # 每隔几分钟: "5" 表示每5分钟
            getattr(schedule.every(int(task["schedule_value"])).minutes, "do")(command)
        elif task["schedule_type"] == "hourly":
            # 每小时
            schedule.every().hour.do(command)
        elif task["schedule_type"] == "daily":
            # 每天特定时间: "09:00"
            schedule.every().day.at(task["schedule_value"]).do(command)
        elif task["schedule_type"] == "weekly":
            # 每周特定时间: "monday 09:00"
            day, time = task["schedule_value"].split()
            getattr(schedule.every(), day).at(time).do(command)
        elif task["schedule_type"] == "once":
            # 一次性任务
            delay_minutes = int(task["schedule_value"])
            threading.Timer(delay_minutes * 60, command).start()
    def start(self):
        """启动所有任务"""
        self.running = True
        self.logger.info("任务管理器启动")
        # 调度所有启用的任务
        for task in self.tasks["tasks"]:
            if task["enabled"]:
                self.schedule_task(task)
        # 主循环
        while self.running:
            schedule.run_pending()
            time.sleep(1)
    def stop(self):
        """停止任务管理器"""
        self.running = False
        self.logger.info("任务管理器停止")
    def get_statistics(self):
        """获取任务统计信息"""
        total = len(self.tasks["tasks"])
        enabled = len([t for t in self.tasks["tasks"] if t["enabled"]])
        success = len([t for t in self.tasks["tasks"] if t["status"] == "success"])
        failed = len([t for t in self.tasks["tasks"] if t["status"] == "failed"])
        return {
            "total": total,
            "enabled": enabled,
            "success": success,
            "failed": failed,
            "pending": total - enabled
        }
# 命令行接口
def cli():
    """命令行接口"""
    manager = TaskManager()
    if len(sys.argv) < 2:
        print("用法: python task_manager.py [command] [args]")
        print("命令:")
        print("  add <name> <command> <type> <value> [description]")
        print("  list")
        print("  remove <task_id>")
        print("  toggle <task_id>")
        print("  start")
        print("  stats")
        return
    command = sys.argv[1]
    if command == "add":
        if len(sys.argv) < 6:
            print("参数不足")
            return
        name = sys.argv[2]
        cmd = sys.argv[3]
        sched_type = sys.argv[4]
        sched_value = sys.argv[5]
        desc = sys.argv[6] if len(sys.argv) > 6 else ""
        manager.add_task(name, cmd, sched_type, sched_value, desc)
        print(f"任务 {name} 添加成功")
    elif command == "list":
        print("\n=== 任务列表 ===")
        for task in manager.tasks["tasks"]:
            status = "✅" if task["enabled"] else "⏸️"
            print(f"{status} [{task['id']}] {task['name']}")
            print(f"   命令: {task['command']}")
            print(f"   调度: {task['schedule_type']} {task['schedule_value']}")
            print(f"   状态: {task['status']}")
            print()
    elif command == "remove":
        if len(sys.argv) < 3:
            print("请指定任务ID")
            return
        manager.remove_task(int(sys.argv[2]))
    elif command == "toggle":
        if len(sys.argv) < 3:
            print("请指定任务ID")
            return
        manager.toggle_task(int(sys.argv[2]))
    elif command == "start":
        print("启动任务管理器...")
        try:
            manager.start()
        except KeyboardInterrupt:
            manager.stop()
            print("任务管理器已停止")
    elif command == "stats":
        stats = manager.get_statistics()
        print("\n=== 任务统计 ===")
        print(f"总任务数: {stats['total']}")
        print(f"启用: {stats['enabled']}")
        print(f"成功: {stats['success']}")
        print(f"失败: {stats['failed']}")
        print(f"待执行: {stats['pending']}")
if __name__ == "__main__":
    cli()

使用示例

# Linux Crontab 管理
chmod +x crontab_manager.sh
./crontab_manager.sh                    # 进入交互模式
./crontab_manager.sh list               # 列出任务
./crontab_manager.sh add "*/5 * * * *" "echo hello" "test_job"  # 添加任务
# Python 任务管理
python task_manager.py add "backup" "tar czf backup.tar.gz /data" "daily" "02:00" "每天备份"
python task_manager.py list
python task_manager.py start

Docker 版定时任务管理

# Dockerfile
FROM alpine:latest
RUN apk add --no-cache python3 py3-pip bash
RUN pip3 install schedule
COPY task_manager.py /app/
COPY crontab_manager.sh /app/
WORKDIR /app
CMD ["python3", "task_manager.py", "start"]

注意事项

1. 权限管理: 确保脚本有适当的执行权限
2. 日志轮转: 设置日志文件大小限制，避免磁盘满
3. 错误通知: 集成邮件或webhook通知机制
4. 监控告警: 监控任务执行状态
5. 并发控制: 避免同时执行大量任务

选择哪种方案取决于你的具体需求：

• 简单的Linux环境使用crontab脚本
• 需要复杂调度逻辑用Python方案
• 跨平台部署推荐Python方案