it编程 > 前端脚本 > Python

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

2人参与 • 2026-08-07 • 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定时任务内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

赞 (0)

您想发表意见！！点此发布评论

推荐阅读

一文详解Python文件操作与路径处理

08-07
Python结合PyQt6实现屏幕截图工具(附源码)

Python结合PyQt6实现屏幕截图工具(附源码)

08-07

Python基础入门之模块搜索路径sys.path用法详解

08-07
Python+PyQt6实现将PDF文件转高清图片

Python+PyQt6实现将PDF文件转高清图片

08-07
Python+PyQt6实现跨平台串口设备热插拔检测

Python+PyQt6实现跨平台串口设备热插拔检测

08-07
Python difflib库实现文本差异比较与合并实战指南

Python difflib库实现文本差异比较与合并实战指南

08-07

猜你喜欢

版权声明：本文内容由互联网用户贡献，该文观点仅代表作者本人。本站仅提供信息存储服务，不拥有所有权，不承担相关法律责任。 如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 2386932994@qq.com 举报，一经查实将立刻删除。

发表评论