django - 确保只有一个工作人员在运行多个工作人员的 Pyramid 网络应用程序中启动 apscheduler 事件

Question

我们有一个用 Pyramid 制作的网络应用程序，并通过 gunicorn+nginx 提供服务。它适用于 8 个工作线程/进程
我们需要工作，我们选择了调度程序。这是我们启动它的方式

from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from apscheduler.scheduler import Scheduler

rerun_monitor = Scheduler()
rerun_monitor.start()
rerun_monitor.add_interval_job(job_to_be_run,\
            seconds=JOB_INTERVAL)

问题是 gunicorn 的所有工作进程都选择了调度程序。我们尝试实现文件锁定，但它似乎不是一个足够好的解决方案。什么是确保在任何给定时间只有一个工作进程选择预定事件并且没有其他线程在下一个 JOB_INTERVAL 之前选择它的最佳方法？ ?
如果我们决定稍后切换到 apache2+modwsgi，该解决方案甚至需要使用 mod_wsgi。它需要与作为服务员的单进程开发服务器一起使用。
来自赏金赞助者的更新
我面临着 OP 描述的相同问题，只是使用 Django 应用程序。如果原始问题，我很确定添加此细节不会有太大变化。出于这个原因，为了获得更多的可见性，我还用 django 标记了这个问题。 .

Answer 1

因为 Gunicorn 从 8 个 worker 开始(在你的例子中)，这个 fork 应用程序 8 次分为 8 个进程。这 8 个进程来自 大师 进程，它监视他们的每个状态并能够添加/删除 worker 。

每个进程都会获得您的 APScheduler 对象的副本，该对象最初是您的主进程的 APScheduler 的精确副本。这导致每个“nth”工作人员(进程)执行每个作业总共“n”次。

解决此问题的方法是使用以下选项运行 gunicorn:

env/bin/gunicorn module_containing_app:app -b 0.0.0.0:8080 --workers 3 --preload

--preload标志告诉 Gunicorn “在 fork 工作进程之前加载应用程序”。通过这样做，每个工作人员都“获得了应用程序的副本，已经由 Master 实例化，而不是实例化应用程序本身”。这意味着以下代码仅在 Master 进程中执行一次:

rerun_monitor = Scheduler()
rerun_monitor.start()
rerun_monitor.add_interval_job(job_to_be_run,\
            seconds=JOB_INTERVAL)

此外，我们需要设置 工作商店 是 以外的任何东西:内存: . 这样，虽然每个工作人员都是自己的独立进程，无法与其他 7 个工作人员进行通信，但通过使用本地数据库(而不是内存)，我们保证了作业存储上的 CRUD 操作的单点真实性。

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

rerun_monitor = Scheduler(
    jobstores={'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')})
rerun_monitor.start()
rerun_monitor.add_interval_job(job_to_be_run,\
            seconds=JOB_INTERVAL)

最后，我们要使用 背景调度器 因为它实现了 start() .当我们调用 start()在 BackgroundScheduler 中，一个新线程在后台启动，负责调度/执行作业。这很重要，因为请记住在步骤 (1) 中，由于我们的 --preload flag 我们只执行 start()功能一次，在Master Gunicorn进程中。根据定义， fork 的进程不继承其父进程的线程，所以每个工作人员都不会运行 BackgroundScheduler 线程。

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

rerun_monitor = BackgroundScheduler(
    jobstores={'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')})
rerun_monitor.start()
rerun_monitor.add_interval_job(job_to_be_run,\
            seconds=JOB_INTERVAL)

由于这一切，每个 Gunicorn worker 都有一个 APScheduler，它被欺骗进入“已启动”状态，但实际上并没有运行，因为它丢弃了它的父线程!每个实例还能够更新作业存储数据库，只是不执行任何作业!

查看 flask-APScheduler用于在 Web 服务器(如 Gunicorn)中运行 APScheduler 的快速方法，并为每个作业启用 CRUD 操作。