代码收藏家技术教程 2025-01-09

Linux 定时任务操作详解及python简单的任务管理器

Linux 定时任务操作

在 Linux 中，定时任务操作主要通过 cron 工具来实现。cron 是一个基于时间的作业调度器，允许用户在指定的时间或周期内执行预定的任务。

1. 查看当前用户的定时任务

使用 crontab 命令查看当前用户的定时任务：

crontab -l

2. 编辑定时任务

使用 crontab -e 命令编辑当前用户的定时任务。此命令将打开一个编辑器（通常是 vi 或 nano），允许你添加、删除或修改任务：

crontab -e

3. 定时任务格式

定时任务的格式通常如下：

* * * * * <command>
┬ ┬ ┬ ┬ ┬
│ │ │ │ │
│ │ │ │ └─── 星期几 (0 - 7) (Sunday is 0 or 7)
│ │ │ └────── 月份 (1 - 12)
│ │ └────────── 日 (1 - 31)
│ └────────────── 小时 (0 - 23)
└───────────────── 分钟 (0 - 59)

示例

30 2 * * * /path/to/command：每天凌晨 2:30 执行 /path/to/command。

0 9 * * 1-5 /path/to/script.sh：每周一到周五的 9:00 执行 /path/to/script.sh。

4. 定时任务的常见例子

每小时执行一次脚本：

0 * * * * /path/to/script.sh

每天凌晨 2:00 执行备份：

0 2 * * * /usr/bin/rsync -av /source /backup

每月 1 号执行清理日志：

0 0 1 * * /path/to/cleanup_logs.

每分钟执行一次：

* * * * * /path/to/check_process.sh

5. Cron Job 的时间字段补充说明

分钟 (0-59)

小时 (0-23)

日 (1-31)

月份 (1-12)

星期几 (0-6 或者 Sun-Sat，0 和 7 都代表周日)

6. 特殊字符用法

星号 ()：表示“每”。

例如， * * * * 表示每分钟、每小时、每天、每月、每周都会执行。

逗号 (,)：用来指定多个值。

例如，1,15 * * * * 表示每小时的第 1 分钟和第 15 分钟执行任务。

连字符 (-)：用来指定范围。

例如，5-10 * * * * 表示每小时的 5 分钟到 10 分钟执行任务。

斜杠 (/)：用于设置步长（例如，表示每隔几个时间单位执行一次任务）。

例如，*/5 * * * * 表示每 5 分钟执行一次任务。

7. 设置系统级定时任务（系统级 cron）

系统级的定时任务通常是管理员级别的，通过修改 /etc/crontab 文件来设置：

sudo vi /etc/crontab

格式：

* * * * * username <command>

例如，系统级任务：

0 3 * * * root /path/to/backup.sh

表示每天 3:00 AM 执行 backup.sh 脚本。

8. Cron 日志

在某些系统中，cron 作业的执行会被记录在 /var/log/syslog 或 /var/log/cron 中。你可以查看日志来调试 cron 作业：

tail -f /var/log/cron

9. 其他有用的命令

查看特定用户的定时任务：

sudo crontab -u username -l

移除当前用户的所有定时任务：

crontab -r

查看 cron 服务状态：

sudo systemctl status cron

10. 限制 cron 执行环境

默认情况下，cron 执行任务时不会加载用户的 shell 配置文件（如 .bashrc），因此如果任务依赖于某些环境变量，可能需要在 cron 文件中显式指定它们。

例如，设置 PATH 环境变量：

PATH=/usr/bin:/bin:/usr/sbin:/sbin
* * * * * /path/to/command

11. 使用python进入任务调度管理器

import schedule
import time
import sys

class TaskSchedulerManager:
    def __init__(self):
        self.tasks = {}

    def list_tasks(self):
        """列出当前所有定时任务"""
        if not self.tasks:
            print("当前没有任何任务。")
        else:
            print("当前所有任务：")
            for task_name, task_info in self.tasks.items():
                print(f"任务名称: {task_name}, 间隔: {task_info['interval']}, 上次执行: {task_info['last_run']}")

    def create_task(self, task_name, job_func, interval, unit="minutes"):
        """创建一个新的定时任务"""
        # 根据单位（分钟/小时/秒）设置任务间隔
        if unit == "minutes":
            job = schedule.every(interval).minutes.do(job_func)
        elif unit == "hours":
            job = schedule.every(interval).hours.do(job_func)
        elif unit == "seconds":
            job = schedule.every(interval).seconds.do(job_func)

        self.tasks[task_name] = {
            'job': job,
            'interval': f"{interval} {unit}",
            'last_run': None
        }

        print(f"任务 '{task_name}' 已创建，计划每 {interval} {unit} 执行。")

    def delete_task(self, task_name):
        """删除指定的任务"""
        if task_name in self.tasks:
            job = self.tasks[task_name]['job']
            schedule.cancel_job(job)
            del self.tasks[task_name]
            print(f"任务 '{task_name}' 已删除。")
        else:
            print(f"错误：任务 '{task_name}' 不存在。")

    def modify_task(self, task_name, new_interval, unit="minutes"):
        """修改任务的执行间隔"""
        if task_name in self.tasks:
            # 先删除旧任务
            self.delete_task(task_name)
            # 然后重新创建任务
            job_func = self.tasks[task_name]['job'].job_func
            self.create_task(task_name, job_func, new_interval, unit)
            print(f"任务 '{task_name}' 的执行间隔已修改为每 {new_interval} {unit}。")
        else:
            print(f"错误：任务 '{task_name}' 不存在。")

    def run(self):
        """开始运行任务调度"""
        while True:
            schedule.run_pending()  # 检查并运行所有挂起的任务
            time.sleep(1)  # 每秒钟检查一次


# 示例任务函数
def job_task_1():
    print("任务 1 执行！")

def job_task_2():
    print("任务 2 执行！")

def main():
    manager = TaskSchedulerManager()

    while True:
        print("\n定时任务管理器")
        print("1. 查看所有任务")
        print("2. 创建新任务")
        print("3. 删除任务")
        print("4. 修改任务间隔")
        print("5. 启动任务调度")
        print("6. 退出")

        choice = input("请输入选择的操作（1-6）：")

        if choice == "1":
            manager.list_tasks()
        elif choice == "2":
            task_name = input("请输入任务名称：")
            interval = int(input("请输入任务间隔："))
            unit = input("请输入时间单位（minutes/hours/seconds）：").lower()
            if unit not in ["minutes", "hours", "seconds"]:
                print("无效的时间单位，请重新输入。")
                continue
            if task_name == "任务1":
                manager.create_task(task_name, job_task_1, interval, unit)
            elif task_name == "任务2":
                manager.create_task(task_name, job_task_2, interval, unit)
            else:
                print("无效的任务名称。")
        elif choice == "3":
            task_name = input("请输入任务名称：")
            manager.delete_task(task_name)
        elif choice == "4":
            task_name = input("请输入任务名称：")
            new_interval = int(input("请输入新的任务间隔："))
            unit = input("请输入新的时间单位（minutes/hours/seconds）：").lower()
            if unit not in ["minutes", "hours", "seconds"]:
                print("无效的时间单位，请重新输入。")
                continue
            manager.modify_task(task_name, new_interval, unit)
        elif choice == "5":
            print("启动任务调度...")
            manager.run()
        elif choice == "6":
            sys.exit("退出定时任务管理器")
        else:
            print("无效的选择，请重新输入。")

if __name__ == "__main__":
    main()