Django APscheduler 阻止更多工作人员运行计划任务答案

【问题标题】：Django APscheduler prevent more workers running scheduled taskDjango APscheduler 阻止更多工作人员运行计划任务
【发布时间】：2022-07-10 18:48:43
【问题描述】：

我在 Django 中使用 APScheduler，在 Windows IIS 上运行我的后台脚本。问题是， taks 会运行多次。如果我在我的 PC 上运行相同的程序，它只会运行一次，但是当我上传到 Windows 服务器（托管我的 Django 应用程序）时，它会运行更多次。我想这与工人数量有关吗？作业是计划好的，但每次作业任务完成时，它就像运行随机数量的实例一样。第一次，然后是 2，然后是 10，然后是 2。即使我有 'replace_existing=True，coalesce=True，misfire_grace_time = 1，max_instances = 1'

planer_zad.py

from apscheduler.schedulers.background import BackgroundScheduler
from blog.views import cron_mail_overdue

def start():

scheduler.add_job(cron_mail_overdue, "cron", hour=7, minute=14, day_of_week='mon-sun', id="task002", replace_existing=True, coalesce= True, misfire_grace_time = 10, max_instances = 1)

scheduler.start()

apps.py

from django.apps import AppConfig


class BlogConfig(AppConfig):
    name = 'blog'

    def ready(self):
        #print('Starting Scheduler...')
        from .planer import planer_zad
        planer_zad.start()

为了测试，我尝试了“间隔”：

scheduler.add_job(cron_mail_overdue, "interval", minutes=1, id="task002", replace_existing=True, coalesce= True, misfire_grace_time = 10, max_instances = 1)

试过了：

    scheduler = BackgroundScheduler({
        'apscheduler.executors.default': {
            'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
            'max_workers': '1'
        },
        'apscheduler.executors.processpool': {
            'type': 'processpool',
            'max_workers': '1'
        },
        'apscheduler.job_defaults.coalesce': 'True',
        'apscheduler.job_defaults.max_instances': '1',
        'apscheduler.timezone': 'UTC',
    })


scheduler.add_job(cron_mail_overdue, "cron", hour=9, minute=3, second=00, day_of_week='mon-sun', id="task002", replace_existing=True, coalesce= True, misfire_grace_time = 10, max_instances = 1)

scheduler.start()

不起作用。有时它只运行一次，然后运行 12 次。

【问题讨论】：

以上问题你找到解决办法了吗？
很遗憾没有。我朝另一个方向前进，使用任务计划程序。

标签： django cron worker apscheduler

【解决方案1】：

只测试对象是否已经存在于 ready() 中：

# django/myapp/apps.py
from django.apps import AppConfig
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

class BlogConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'blog'

    def __init__(self, app_name, app_module):
        super(BlogConfig, self).__init__(app_name, app_module)
        self.my_pubnub_listener  = None

    def ready(self):
        if self.my_pubnub_listener is None:
            background_scheduler = BackgroundScheduler()
            background_scheduler.add_job(task1, CronTrigger.from_crontab('* * * * *')) # Every minutes (debug).
            background_scheduler.start()
            return background_scheduler

    def task1(self):
        print("cron task is working")

您可以稍后再调用它：

# api.py
from django.apps import apps

@router.get("/background-task")
def background_task(request):
    """
        Run a background task.
    """
    user = request.user
    blog_config= apps.get_app_config('blog')
    background_scheduler = blog_config.background_scheduler
    return {"status": "Success", "True": str(background_scheduler)}

【讨论】：