我该在Django应用程序中的哪里注册rq-scheduler作业？

Question

我想使用django_rq和rq-scheduler 进行离线任务，但是我不确定在哪里调用rq-scheduler的能力来安排重复任务。目前，我已将我的调度添加到我的应用程序中的tasks.py模块，并在__init__.py中导入该模块。不过，一定有更好的方法吧？

提前感谢。

- Carson

---

我认为 - 没有地方。每次重新启动应用程序服务器时，作业将被添加到调度程序队列中（每次启动时任务都会增加）。作为解决方法，您可以在每次启动之前清除/删除rq:scheduler:scheduled_jobs键，然后再添加它们。但请记住避免在工作进程中添加/删除作业 - 我不知道如何实现这一点。 - marcinn

Answer 1

我新建了一个自定义管理命令，用于修改和替换django_rq中包含的rqscheduler命令。这里提供了一个示例：https://github.com/rq/rq-scheduler/issues/51#issuecomment-362352497。

- Pete

---

在我看来，这应该是被接受的答案。这个方法优雅地解决了问题。如果你没有看到Pete做了什么，他已经对内置的rqscheduler管理命令进行了子类化，以便同时调用scheduler.schedule命令。我曾经遇到过schedule方法运行多次的问题，而这个方法解决了这个问题。感谢@Pete！ - davidscolgan

这是一个很棒的解决方案。启动调度程序进程并安排周期性任务。 - madmanick

Answer 2

我发现最好的运行位置是从你的AppConfig中的apps.py。

def ready(self):
    scheduler = django_rq.get_scheduler('default')

    # Delete any existing jobs in the scheduler when the app starts up
    for job in scheduler.get_jobs():
        job.delete()

    # Have 'mytask' run every 5 minutes
    scheduler.schedule(datetime.utcnow(), 'mytask', interval=60*5)

- Aaron C. de Bruyn

---

Django rq如何检测此调度程序？需要通过调用函数从某个地方启动它吗？如果不需要，那么在运行python manage.py rqscheduler时如何自动启动它？ - Sheesh Mohsin

在Django的新版本中，您可以创建一个apps.py文件来配置您的应用程序。请参阅此处获取最新信息：https://docs.djangoproject.com/en/1.9/ref/applications/ - Aaron C. de Bruyn

需要删除作业吗？因为我可以看到队列在每次应用程序重新启动时都会增加。 - Kishan Mehta

如果应用程序停止，队列中尚未处理的任何内容都将保留在队列中，直到有其他进程对其进行处理。 - Aaron C. de Bruyn

如果有人使用不同于“default”队列，那么使用scheduler.schedule()方法的queue_name参数指定队列名称会非常有用。 - Erik Telepovský

如果您正在使用gunicorn或其他具有多个工作进程的Web服务器运行Django应用程序，则需要小心，因为这种方法很可能会在计划中复制任务，并且它们将被执行多次。因此，例如，预定的电子邮件将发送两次（可能不是期望的结果）。 - BernarditoLuis

Answer 3

我在我的项目应用程序（基于Django）__init__模块中添加了调度功能，但使用小函数进行包装，以防止重复排队作业。调度策略可能取决于您的特定需求（例如，您可能需要对作业参数进行额外检查）。

以下是适合我需求并且可行的代码：

import django_rq
from collections import defaultdict
import tasks

scheduler = django_rq.get_scheduler('default')

jobs = scheduler.get_jobs()
functions = defaultdict(lambda: list())

map(lambda x: functions[x.func].append(x.meta.get('interval')), jobs)

now = datetime.datetime.now()

def schedule_once(func, interval):
    """
    Schedule job once or reschedule when interval changes
    """
    if not func in functions or not interval in functions[func]\
            or len(functions[func])>1:

        # clear all scheduled jobs for this function
        map(scheduler.cancel, filter(lambda x: x.func==func, jobs))

        # schedule with new interval
        scheduler.schedule(now+datetime.timedelta(seconds=interval), func,
                interval=interval)

schedule_once(tasks.some_task_a, interval=60*5)
schedule_once(tasks.some_task_b, interval=120)

此外，我将此代码片段封装起来以避免在包级别上导入。

def init_scheduler():
    # paste here initialization code

init_scheduler()

- marcinn

---

我刚刚遇到了“在settings.py中安排任务”的恶梦。显然，每次导入时，它都会重新添加计划的任务。我的任务本来应该每小时触发一次，但现在却一次性堆积了50个...;) - Aaron C. de Bruyn

Answer 4

您应该使用Django命令来运行计划任务 https://docs.djangoproject.com/en/3.2/howto/custom-management-commands/

像这样

输入图像描述在这里

class Command(BaseCommand):

  def handle(self, *args, **options):
    scheduler = django_rq.get_scheduler('crontab_job')
    for job in scheduler.get_jobs():
        scheduler.cancel(job)
    # 定时任务例子1
    scheduler.cron(
        "*/3 * * * *",  # 每周一零点零时零分执行 0 0 * * 0    测试可以使用 */3 * * * * 每3分钟执行一次
        func=gong_an_job,  # Function to be queued
        kwargs={'msg': '我是王龙飞1，我喜欢修仙', 'number': 1},  # Keyword arguments passed into function when executed
        repeat=None,  # Repeat this number of times (None means repeat forever)
        queue_name='crontab_job',  # In which queue the job should be put in
        use_local_timezone=False  # Interpret hours in the local timezone
    )
    # 定时任务例子2
    scheduler.cron(
        "*/5 * * * *",  # 每周一零点零时零分执行 0 0 * * 0    测试可以使用 */3 * * * * 每3分钟执行一次
        func=gong_an_job,  # Function to be queued
        kwargs={'msg': '我是王龙飞222222，我喜欢修仙', 'number': 22222},  # Keyword arguments passed into function when executed
        repeat=None,  # Repeat this number of times (None means repeat forever)
        queue_name='crontab_job',  # In which queue the job should be put in
        use_local_timezone=False  # Interpret hours in the local timezone
    )

创建定时任务

python manage.py rq_crontab_job

检查Crontab作业并将其加入队列。

python manage.py rqscheduler --queue crontab_job

运行定时任务的 crontab 命令。

python manage.py rqworker crontab_job

我认为第一个答案很好，但在多进程环境中可能会出现一些问题，你只需要运行一次来创建crontab任务！

- dragonfly-wang

---

你正在回答一个非常古老的问题。你认为你的答案比其他答案更好吗？如果是，请编辑你的回复，清楚地说明这一点，并展示之前的答案可能已经不再相关。此外，请在格式上做出努力。 - mozway