python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python APScheduler定时任务

Python中定时任务APScheduler框架使用教程

作者：detayun

在 Python 开发中,定时任务是刚需场景,而APScheduler又是Python生态中最通用的定时任务框架,具有轻量、易用、功能全面的特点,下面小编就和大家详细讲讲它的具体应用吧

一、APScheduler 简介

在 Python 开发中，定时任务是刚需场景：爬虫定时抓取、数据同步、日志清理、报表生成、服务巡检等。

原生 while + sleep 写法存在致命缺陷：

APScheduler（Advanced Python Scheduler） 是 Python 生态最标准、最轻量、最通用的定时任务框架，适配脚本、爬虫、后端服务、Web 项目，支持 Linux Crontab 定时、间隔轮询、定点单次执行，线程/进程并发、任务持久化、动态启停，是中小型项目定时任务首选方案。

核心优势

二、环境安装

pip install apscheduler

三、四大核心组件（必须掌握）

APScheduler 采用四大组件解耦设计，理解这四个组件就掌握了 80% 原理：

  1. Scheduler 调度器：任务总控制器，负责管理所有任务生命周期（启动、暂停、停止、调度）。
  2. Trigger 触发器：决定任务什么时候执行、多久执行一次，核心三种：date / interval / cron。
  3. Executor 执行器：真正执行任务的工人，支持线程池、进程池，解决任务阻塞问题。
  4. JobStore 任务存储：保存定时任务配置，默认内存存储（重启失效），支持数据库持久化。

四、五种调度器选型（场景对照）

不同项目必须选对应调度器，否则会阻塞、不生效、冲突：

调度器适用场景特点
BlockingScheduler独立脚本、爬虫脚本、单机定时服务阻塞主线程，独占进程，最常用
BackgroundSchedulerFlask/Django Web 项目后台运行，不阻塞主程序
AsyncIOSchedulerasync/await 异步项目适配异步任务
GeventSchedulergevent 协程项目协程调度
TornadoSchedulerTornado 框架项目框架适配

爬虫/独立 Python 脚本首选：BlockingScheduler

五、三大触发器完整详解

1. date 触发器（一次性任务）

指定某个时间点执行一次，执行完毕自动销毁任务，适合临时定时、单次执行场景。

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

def once_task():
    print(f"一次性任务执行：{datetime.now()}")

scheduler = BlockingScheduler()
# 指定时间执行一次
scheduler.add_job(once_task, "date", run_date="2026-12-31 18:00:00")
scheduler.start()

2. interval 触发器（固定间隔轮询）

每隔指定时间执行一次，支持秒/分钟/小时/天，适合高频巡检、实时轮询爬虫。

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

def loop_task():
    print(f"间隔轮询任务：{datetime.now()}")

scheduler = BlockingScheduler()
# 每 3 秒执行一次
scheduler.add_job(loop_task, "interval", seconds=3)
# 常用：minutes=5 / hours=1 / days=1
scheduler.start()

3. cron 触发器（重点、生产最常用）

类 Linux Crontab 表达式，精准控制时分秒、周、月，适合固定时段定时任务（爬虫每日多时段更新）。

常用规则参数

实战示例（爬虫经典配置）

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

def cron_spider_task():
    print(f"定时爬虫执行成功：{datetime.now()}")

scheduler = BlockingScheduler()

# 每天 8/10/12/15/17/22 整点执行
scheduler.add_job(
    func=cron_spider_task,
    trigger="cron",
    hour="8,10,12,15,17,22",
    minute="0",
    second="0",
    id="spider_cron_task"
)

scheduler.start()

六、高阶核心用法（生产必备）

1. 任务传参（任意函数支持传参）

通过 args 位置参数、kwargs 关键字参数传参。

注意：单参数 args=(xxx,) 末尾逗号不能省略，必须是元组！

from apscheduler.schedulers.blocking import BlockingScheduler

def task_demo(name, mode="定时执行"):
    print(f"任务：{name}，执行模式：{mode}")

scheduler = BlockingScheduler()
scheduler.add_job(
    func=task_demo,
    trigger="cron",
    hour=8,
    minute=0,
    args=("SMS新闻爬虫",),
    kwargs={"mode": "每日定点轮询"},
    id="task_param_demo"
)
scheduler.start()

2. 启动立即执行一次（解决首次等待问题）

默认 cron 任务只会等待下一个时间点，不会立刻执行。通过 next_run_time=datetime.now() 实现：启动立刻跑一次 + 后续正常定时。

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

def spider_task():
    print("爬虫任务执行")

scheduler = BlockingScheduler()
scheduler.add_job(
    func=spider_task,
    trigger="cron",
    hour="8,22",
    minute="0",
    next_run_time=datetime.now()
)
scheduler.start()

3. 自定义线程池/进程池（解决任务阻塞）

APScheduler 默认单线程执行，耗时任务会阻塞后续所有任务。手动配置执行器，适配爬虫 IO 密集场景：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

# 自定义并发池
executors = {
    "default": ThreadPoolExecutor(100),   # IO爬虫用线程池
    "processpool": ProcessPoolExecutor(1) # 计算任务用进程池
}

scheduler = BlockingScheduler(executors=executors)

4. 任务动态管理（增删启停）

# 根据id删除任务
scheduler.remove_job("task_id")

# 暂停任务
scheduler.pause_job("task_id")

# 恢复任务
scheduler.resume_job("task_id")

# 查看所有任务
print(scheduler.get_jobs())

七、生产级完整模板（爬虫专用）

整合 装饰器日志、异常捕获、并发配置、立即执行、定时调度，可直接上线：

import traceback
from functools import wraps
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

# 任务日志装饰器
def task_logger(task_name):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"[{datetime.now()}]【{task_name}】任务开始")
            try:
                res = func(*args, **kwargs)
                print(f"[{datetime.now()}]【{task_name}】任务执行成功")
                return res
            except Exception as e:
                err = traceback.format_exc()
                print(f"[{datetime.now()}]【{task_name}】任务异常：{repr(e)}\n{err}")
                raise
        return wrapper
    return decorator

# 业务任务
@task_logger("SMS新闻爬虫")
def spider_task():
    # 此处替换为你的爬虫逻辑
    pass

if __name__ == "__main__":
    # 并发配置
    executors = {
        "default": ThreadPoolExecutor(100),
        "processpool": ProcessPoolExecutor(1)
    }
    scheduler = BlockingScheduler(executors=executors)

    # 添加定时任务
    scheduler.add_job(
        func=spider_task,
        trigger="cron",
        hour="8,10,12,15,17,22",
        minute="0",
        second="0",
        id="sms_spider_task",
        next_run_time=datetime.now()
    )

    print("定时调度器启动成功")
    scheduler.start()

八、高频踩坑总结

装饰器失效：禁止 from xxx import * 全局导入，会覆盖被装饰函数，导致装饰器不生效。

任务不立即执行：Cron 默认等待下一个时间点，必须加 next_run_time=datetime.now()。

任务串行阻塞：默认单线程，耗时任务卡死后续任务，必须配置 ThreadPoolExecutor。

传参报错：单参数 args=("test",) 必须带逗号，否则不是元组。

重启任务丢失：默认内存存储，常驻服务建议开启 MySQL/Redis 持久化。

Job ID 重复：每个任务 ID 必须唯一，重复会直接覆盖旧任务。

九、APScheduler 与其他定时框架对比

框架优点缺点适用场景
APScheduler轻量、灵活、支持多触发器、并发无分布式集群90% 中小型项目、爬虫、自动化
schedule极简、上手快功能弱、无并发、无持久化简单测试脚本
Celery Beat分布式、强大稳定重、需消息队列大型分布式服务

十、总结

APScheduler 是 Python 定时任务最均衡、最通用的解决方案：

到此这篇关于Python中定时任务APScheduler框架使用教程的文章就介绍到这了,更多相关Python APScheduler定时任务内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文