Apscheduler 多次执行作业

Question

我有一个使用 uwsgi（有 10 个工作人员）+ ngnix 运行的 django 应用程序。我正在使用 apscheduler 进行调度。每当我安排一个作业时，它都会被执行多次。从这些答案ans1，ans2 我知道这是因为调度程序是在 uwsgi 的每个工作人员中启动的。我通过按照answer 中的建议将调度程序绑定到套接字并在数据库中保持状态，从而对调度程序进行了条件初始化，这样只会启动调度程序的一个实例，但仍然存在同样的问题，有时也存在创建作业时，发现调度程序未运行，作业保持挂起且未执行。

我正在使用以下代码在 django 应用程序的 url 中初始化 apscheduler。这将在应用程序启动时启动调度程序。

def job_listener(ev):
    print('event',ev)


job_defaults = {
    'coalesce': True,  
    'max_instances': 1
}

scheduler = BackgroundScheduler(job_defaults=job_defaults, timezone=TIME_ZONE, daemon=False)
scheduler.add_jobstore(MongoDBJobStore(client=client), 'default')
scheduler.add_executor(ThreadPoolExecutor(), 'default')
scheduler.add_executor(ProcessPoolExecutor(),'processpool')
scheduler.add_listener(job_listener)


def initialize_scheduler():
    try:
        if scheduler_db_conn.find_one():
            print('scheduler already running')
            return True
        scheduler.start()
        scheduler_db_conn.save({'status': True})
        print('---------------scheduler started --------------->')
        return True
    except:
        return False

我使用以下代码来创建作业。

from scheduler_conf import scheduler
def create_job(arg_list):
    try:
        print('scheduler status-->',scheduler.running)
        job = scheduler.add_job(**arg_list)
        return True
    except:
        print('error in creating Job')
        return False

我无法正确配置和运行调度程序。我已经参考了apschedule中的所有线程，但仍然没有解决方案。

• 如果我不限制在每个工作人员中运行多个调度程序，则作业会执行多次。
• 但是，如果我限制在一个工作器中只运行一个调度程序，一些作业会保持挂起并且不会执行。

解决办法是什么？

Answer 1

让我们考虑以下事实：

(1) 默认情况下，UWSGI 将您的 Django 应用程序预加载到 UWSGI 主进程的内存中在分叉其工作人员之前。

(2) UWSGI 从 master 中“分叉”worker，这意味着它们本质上被复制到每个 worker 的内存中。由于fork() 的实现方式，子进程（即工作者）不会继承父进程的线程。

(3) 当您调用BackgroundScheduler.start() 时，会创建一个线程，该线程负责在任何w​​orker/master 调用此函数时执行作业。

您必须做的就是在创建任何工作人员之前在主进程上调用BackgroundScheduler.start()。通过这样做，当工人被创建时，他们将不会继承 BackgroundScheduler 线程（上面的#2），因此不会执行任何作业（但他们仍然可以通过与作业存储通信来调度/修改/删除作业！）。

为此，只需确保在任何函数/模块实例化您的应用程序中调用 BackgroundScheduler.start()。例如，在以下 Django 项目结构中，我们（可能）希望在 wsgi.py 中执行此代码，这是 UWSGI 服务器的入口点。：

mysite/
manage.py
mysite/
    __init__.py
    settings.py
    urls.py
    wsgi.py

陷阱：

不要“在 django 应用程序的 urls 中初始化 [e] apscheduler....这将在应用程序启动时启动调度程序。” 这些可能由每个工作人员加载，因此 @ 987654329@被多次执行。

不要在"lazy-app"模式下启动UWSGI服务器，这将在每个workers创建后加载应用程序。

不要使用默认（内存）作业存储运行 BackgroundScheduler。这将在所有工人之间造成脑裂综合症。您希望对作业上执行的所有 CRUD 操作强制执行单点真实性，就像使用 MongoDB 一样。

This post 可能会为您提供更多详细信息，仅在 Gunicorn（WSGI 服务器）环境中。

Answer 2

假设您想在应用程序中使用默认作业存储和默认执行器运行 BackgroundScheduler：

from apscheduler.schedulers.background import BackgroundScheduler


scheduler = BackgroundScheduler()

这将为您提供一个带有名为“default”的 MemoryJobStore 和一个名为“default”的 ThreadPoolExecutor 的 BackgroundScheduler，默认最大线程数为 10。

现在，假设您想减少线程数，并且还想调整新作业的默认值并设置不同的时区。下面的例子会让你明白：

from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler

job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler(job_defaults=job_defaults, timezone=utc)