如何在 Python 中不使用 Celery 调度任务及替代方案
每个后端开发者可能都遇到过这种情况。你正在用 FastAPI 或 Flask 构建一个 Web 服务，项目规模还很小，运行在单台服务器上。然后产品经理说："嘿，我们每天早上 9 点给客户发送摘要，每小时重置过期购物车怎么样？"
第一反应是直接加上 Celery 配合 Celery Beat 和 Redis。但仅仅为了两个后台任务就启动一个消息代理、配置 worker、监控守护进程，显得有些过度。第二个选择是系统 cron，但这会把逻辑分散到应用代码和服务器配置中。你希望调度器嵌入代码中，能够处理 2 个任务，并在需要时扩展到多个节点。
这就是 APScheduler（高级 Python 调度器）库派上用场的地方。
这个库能做什么
APScheduler 已经存在一段时间了，项目目前正在向 4.0 版本过渡。第四个分支仍处于预发布状态，但它已经显著重构了库的架构。
以前它主要是一个用于单个进程的本地调度器，但现在这个工具已经发展成为一个成熟的分布式队列和调度系统。最简单的用例仍然只需要三行代码。
核心概念：声明常规的 Python 函数或协程，并附加一个带有必要执行条件的触发器。
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()
async def send_digest():
print("Отправляем утренний дайджест...")
# Запуск каждый будний день в 9 утра
scheduler.add_job(send_digest, 'cron', day_of_week='mon-fri', hour=9, minute=0)
scheduler.start()
库会处理时间跟踪、计算偏移量，并在正确的时刻调用函数。
调度选项
提供四种内置触发器类型：
- Cron 触发器。熟悉的 Linux 风格语法。可以灵活设置星期几、月份、具体小时和分钟。
- 时间间隔。每隔 N 秒、分钟或小时运行一次。适用于定期轮询外部 API。
- 日历触发器。当间隔取决于月份或年份长度时需要。例如，每个月第一天中午准时运行任务。
- 一次性运行。仅在指定的未来日期和时间触发一次。
如果内置条件不够用，触发器可以通过复合规则组合，也可以编写自定义类实现自定义逻辑。
防止常见故障
在实践中，后台任务经常遇到超载和延迟问题。APScheduler 包含几个有用的机制，这些在基于 3 和 4 构建的自定义解决方案中经常被忽视。
首先，限制并发运行。如果你配置了每五分钟执行一次大型报表导出，但上一次运行卡了七分钟，调度器不会并行启动第二个实例，也不会让数据库过载。
其次，5 参数。它会在任务开始时间添加随机延迟。假设你有一百个 worker，它们都需要在 00:00 精确刷新缓存。没有 jitter，数据库会收到瞬时负载峰值。加上几秒钟的随机偏移，负载就会均匀分布。
第三，错过触发时间处理。如果服务器在负载下冻结或正在重启，调度器会检查任务延迟了多久。如果延迟在可接受范围内，任务会运行；如果不在范围内，则会被跳过而不会造成队列积压。
存储和分布式模式
对于简单的脚本，任务可以保存在内存中。但如果服务重启，调度计划会被重置。为了避免这个问题，库支持持久化存储：
- PostgreSQL
- MySQL
- SQLite
- MongoDB
在第四版中，调度器学会了在分布式集群中工作。多个应用实例连接到共享数据库和事件代理（支持 Redis、PostgreSQL LISTEN/NOTIFY 和 MQTT）。
这实现了水平扩展：一个节点宕机，其他节点会从队列中接管任务。
# Пример концепции работы с постоянным хранилищем
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {
'default': SQLAlchemyJobStore(url='postgresql+asyncpg://user:pass@localhost/mydb')
}
scheduler = AsyncIOScheduler(jobstores=jobstores)
同步和异步代码
该库很好地适配现代 Python 技术栈。提供不同的调度器实现：
- 6 和与 Trio 的集成，适用于 FastAPI、Litestar 或 Aiohttp 等现代异步后端。
- 7 和 8 适用于同步脚本、Django 和 Flask。
你不需要用 9 将异步函数包装成同步变通方案——调度器在本机事件循环中原生 await 协程。
适用场景
我通常在三种典型场景中使用 APScheduler：
- 小型微服务和机器人。当部署 Celery 或 RQ 显得过度，但又需要在进程内直接执行周期性任务时。
- 延迟用户操作。例如，在配送后恰好 24 小时发送一封邮件询问用户对订单进行评价。
- 清理临时数据以及定期从外部系统同步参考数据。
值得在项目中使用吗
如果你使用 Python 编写并且需要可预测的基于时间的任务执行，APScheduler 是最成熟的方案之一。
目前唯一的细微差别：3.x 和 4.0 版本之间的过渡期。第三个分支已经在生产环境中经过多年测试，最大程度稳定，但在分布式操作方面有局限性。4.0 版本带来了现代架构和扩展能力，但作者坦诚警告在最终发布之前可能会有破坏性 API 变更。
对于当前的生产环境，使用稳定的 10 分支更安全，而 4.0 版本值得在新项目或新微服务中尝试，同时密切关注即将到来的变化。
相关项目