🕒 Python 中的定时任务实现

在 Python 中，我们可以使用 APScheduler 和 schedule 这两个库来实现定时任务。它们分别适用于不同复杂度和需求的项目。

🧰 APScheduler

安装依赖

[tool.poetry.dependencies]
python = "^3.9"
apscheduler = "^3.10.4"

示例代码

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timedelta

def sch_test(job_type):
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('时间：{}, {}测试apscheduler'.format(now, job_type))

task = BlockingScheduler(timezone="Asia/Shanghai")
task.add_job(func=sch_test, args=('一次性任务',), trigger='date', next_run_time=datetime.now() + timedelta(seconds=3))
task.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5')
task.add_job(func=sch_test, args=('定时任务',), trigger='interval', hours=1)
task.add_job(func=sch_test, args=('定时任务',), trigger='cron', second='*/5', jitter=30, misfire_grace_time=60)
task.add_job(func=sch_test, args=('循环任务',), trigger='interval', seconds=3)
task.add_job(func=sch_test, args=('每天22:00执行的任务',), trigger='cron', hour=22, minute=0)
task.start()

功能说明

• date：在特定时间执行一次任务。
• cron：按 cron 表达式周期执行任务。
• interval：每隔固定时间间隔执行任务。
• 支持时区设置（如 Asia/Shanghai）。
• 支持任务延迟容忍（misfire_grace_time）和随机延迟（jitter）。

---

🧵 schedule

安装

pip install schedule

示例代码

import schedule
import time
import traceback

def make_loop(job, minutes=None, hours=None):
    job()
    
    if minutes or hours:
        if minutes:
            schedule.every(minutes).minutes.do(job)
        elif hours:
            schedule.every(hours).hours.do(job)

        while True:
            schedule.run_pending()
            time.sleep(5)

def test():
    print('test')

def job():
    try:
        test()
    except:
        traceback.print_exc()

make_loop(job, minutes=1)  # 每1分钟执行一次

功能说明

• 简单易用，适合轻量级定时任务。
• 支持按分钟、小时等周期执行。
• 配合 while 循环持续运行任务。

---

📝 进阶用法：带日志记录的定时任务

from apscheduler.schedulers.blocking import BlockingScheduler
from zoneinfo import ZoneInfo
from loguru import logger

logger.add("logs/my_log.log", rotation="1 day", level="INFO")

def main():
    logger.info("开始执行 main 函数")
    try:
        # do task
        logger.info("执行完成")
    except Exception as e:
        logger.error(f"执行过程中发生错误: {e}", exc_info=True)

if __name__ == "__main__":
    main()
    sched = BlockingScheduler(timezone=ZoneInfo("Asia/Shanghai"))
    sched.add_job(main, 'cron', hour=5, minute=0)

    logger.info("启动定时任务调度器")
    try:
        sched.start()
    except (KeyboardInterrupt, SystemExit):
        logger.info("调度器已停止")

说明

• 使用 loguru 记录日志，便于调试和监控。
• 使用 zoneinfo 设置时区。
• 定时在每天早上 5:00 执行任务。

---

📌 总结：

• 如果你需要功能强大、支持多种触发器和持久化的任务调度，推荐使用 APScheduler。
• 如果你只需要简单的定时执行逻辑，schedule 是一个轻量而实用的选择。