Python中WSGI项目计划任务的最佳实践是什么？

场景就是，比如 Flask、Django 写的 WSGI 应用吧，提供 Web 服务，同时希望定期（比如每 10 分钟）执行一些任务（清缓存、更新计算内容等等）

如果我对 WSGI 原理没理解错的话，假如把计划任务写进 Flask，那么如果希望以 WSGI 多进程的方式提高性能，实际会起多个 WSGI + Flask 组成的服务进程，这样计划任务就也是多个，在每 10 分钟 n 个进程就会执行 n 次定期任务吧？

所以想请教一下我的理解有没有错误，考虑周全一些的话计划任务应当如何实现呢？
Python中WSGI项目计划任务的最佳实践是什么？

7 回复

celery


对于在WSGI项目（比如Flask、Django应用）里跑计划任务，核心思路是把任务调度器独立于Web服务进程。直接在WSGI进程里搞time.sleep或者threading.Timer是坑，会阻塞请求。通常用这三种路子：

  1. 用操作系统的调度器：最省心。写独立的Python脚本执行任务，然后用cron（Linux）或Task Scheduler（Windows）来定时调用。和WSGI应用完全解耦，通过数据库、消息队列或者共享文件来通信。
  2. 用消息队列：专业选择。搞个Celery，搭配RabbitMQ或Redis做broker。Web应用把任务丢到队列，Celery的worker进程在后台异步执行，支持定时（用celery beat）。这是生产环境最常用的，扩展性好。
  3. 用APScheduler等库内嵌：如果任务非常轻量，且你不想引入外部组件，可以在应用启动时（比如在if __name__ == '__main__':里）启动一个后台线程运行APScheduler。但要特别注意：在像gunicorn这类多worker预分支（pre-fork）模型下，这会导致每个worker都启动自己的调度器，任务可能重复执行。通常需要配合文件锁或环境变量来确保只有主进程启动调度器。

简单代码示例（APScheduler内嵌，适用于单进程开发环境）：

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)

def my_scheduled_task():
    print("定时任务执行了！")

# 创建调度器
scheduler = BackgroundScheduler()
scheduler.add_job(func=my_scheduled_task, trigger="interval", seconds=60) # 每60秒执行一次
scheduler.start()

@app.route('/')
def hello():
    return "Hello, World!"

if __name__ == '__main__':
    # 注意：在生产WSGI服务器（如gunicorn）下，这样启动可能会导致多个调度器实例。
    # 仅适用于单进程的调试环境。
    app.run()

总结建议：生产环境优先考虑Celery或系统cron。

写个 service 然后定时去调用一下？

Celery 处理定时任务的模块

你的理解有问题

这取决于你的定时任务是怎么触发的，如果你在 flask 应用中触发，自然是有多少应用就会有多少次触发，如果你只是在代码中定义定时任务，触发由其他进程来管理，自然就是只触发一次，楼上们也给出了一些解决方案，celery 就是另外的进程来触发你的定时任务

linux 系统定时任务了解一下~
django 的 django-crontab 了解一下~
坏处是上面那两种貌似都不支持 Windows

回到顶部