用 Python 写一个轻量级任务调度器：自动定时执行你的脚本（附完整实战项目）

适合自动化、脚本执行、监控任务、文件同步、爬虫定时采集等场景
完整示例可直接复制运行 ✔

---

📌 一、为什么要自己写一个任务调度器？

在实际工作中，我们会遇到很多“需要每天自动执行”的场景：

• 定时爬取网站数据
• 每天固定发邮件
• 每小时同步文件到服务器
• 每 5 分钟检测服务健康状况

虽然 Linux 有 Crontab、Windows 有 Task Scheduler，但它们：

❌ 不好跨平台
❌ 不好动态修改任务
❌ 不好写复杂逻辑（依赖、并发、异常通知）

▶ APScheduler = 超轻量 + 强大 + 跨平台 + 支持 Python 全生态

---

🔧 二、用 APScheduler 搭建一个任务调度器的核心能力

通过 APScheduler，你可以轻松获得一个类似 Server 的调度系统：

能力	说明
定时执行	每秒 / 每分钟 / 每天 / 每周 / Cron 表达式
异常重试	任务失败自动记录日志
多任务并行	默认线程池或进程池同时跑多个任务
动态添加/删除任务	运行中可随时新增
持久化任务	支持 SQLite / MySQL / Redis

---

🐍 三、安装 APScheduler

pip install apscheduler

---

🛠️ 四、构建你的“轻量级任务调度系统”

下面我们写一个基础版的调度器，实现：

✓ 每 10 秒打印一次日志
✓ 每分钟备份文件
✓ 每天晚上 23:30 自动执行脚本

---

✔ 核心代码：scheduler.py

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import shutil
import os

scheduler = BlockingScheduler()

# 任务 1：日志输出
def job_log():
    print(f"[{datetime.now()}] 定时任务运行中…")

# 任务 2：自动备份文件
def job_backup():
    source = "/path/to/source/file.txt"
    target = "/path/to/backup/file_backup.txt"
    shutil.copy(source, target)
    print(f"[{datetime.now()}] 文件已备份到：{target}")

# 任务 3：执行脚本（例如爬虫）
def job_run_script():
    os.system("python spider.py")  # 执行你自己的脚本
    print(f"[{datetime.now()}] 已执行 spider.py")

# 添加任务
scheduler.add_job(job_log, 'interval', seconds=10, id='log_task')
scheduler.add_job(job_backup, 'cron', minute='*', id='backup_task')
scheduler.add_job(job_run_script, 'cron', hour=23, minute=30, id='script_task')

if __name__ == '__main__':
    print("任务调度器启动中…")
    scheduler.start()

---

💡 五、让调度器更强：任务并行 + 异常捕获

APScheduler 默认是多线程的，可并行执行任务。

我们可以进一步增强，加入异常日志：

import logging

logging.basicConfig(
    filename="scheduler.log",
    level=logging.INFO,
    format="%(asctime)s — %(levelname)s — %(message)s"
)

def job_run_script():
    try:
        os.system("python spider.py")
        logging.info("任务执行成功 spider.py")
    except Exception as e:
        logging.error(f"任务失败：{str(e)}")

---

🔥 六、关键功能：动态添加任务（真实业务需求）

例如：你做一个平台，想让用户“随时添加定时任务”。

scheduler.add_job(job_run_script, 'interval', minutes=5, id='dynamic_job')

删除任务：

scheduler.remove_job('dynamic_job')

---

🧪 七、完整实战示例：构建一个自动化“网站监控报警系统”

业务场景：

每 1 分钟监控一次网站是否正常，如失败则自动发送邮件报警。

---

① 安装依赖

pip install apscheduler requests yagmail

---

② 实现监控脚本 monitor.py

import requests
import yagmail
from datetime import datetime

mailer = yagmail.SMTP("your_email@gmail.com", "your_email_password")

def check_site():
    url = "https://example.com"

    try:
        r = requests.get(url, timeout=3)
        if r.status_code != 200:
            raise Exception("状态码异常")
        print(f"{datetime.now()} — 网站正常")
    except Exception as e:
        print(f"{datetime.now()} — 网站异常：{str(e)}")
        mailer.send(
            to="admin@example.com",
            subject="【报警】网站访问失败",
            contents=f"网站异常时间：{datetime.now()}\n错误：{str(e)}"
        )

---

③ 把监控任务加入调度器 scheduler.py

from monitor import check_site

scheduler.add_job(check_site, 'interval', minutes=1, id='monitor_job')

---

④ 运行调度器

python scheduler.py

结果：

任务调度器启动中…
2025-02-10 10:02:00 — 网站正常
2025-02-10 10:03:00 — 网站正常
...
（异常时自动发送邮件）

---

📦 八、最终效果：一个可商用的小型任务调度系统

你现在拥有：

✓ 跨平台任务调度器
✓ 可执行任意 Python 脚本
✓ 支持定时任务、Cron 表达式
✓ 支持动态任务
✓ 支持异常告警（邮件）
✓ 支持扩展为 Web API 服务

这套能力可以轻松升级成：

▶ 自动化监控平台
▶ 文件同步系统
▶ 数据采集系统
▶ 数据清洗定时任务
▶ 企业自动化运维系统

---