设置动态调度芹菜节拍

Question

我的通知模型中有send_time字段。我想在此时向所有移动客户端发送通知。

我现在正在做的是，我已经创建了一个任务并计划了它每分钟一次

tasks.py

@app.task(name='app.tasks.send_notification')
def send_notification():
   # here is logic to filter notification that fall inside that 1 minute time span 
   cron.push_notification()

settings.py

CELERYBEAT_SCHEDULE = {
    'send-notification-every-1-minute': {
        'task': 'app.tasks.send_notification',
        'schedule': crontab(minute="*/1"),
    },
}

一切都按预期工作。

问题：

有没有办法根据send_time字段安排任务，所以我不必每分钟都安排任务。

更具体地说，我想创建一个新的任务实例，因为我的通知模型获得新条目并根据该记录send_time字段安排它。

注意：我正在使用芹菜与 django 的新集成，而不是 django-celery 包

Answer 1

要在指定的日期和时间执行任务，您可以在调用任务时使用eta apply_async 的属性，如文档中所述

创建通知对象后，您可以将任务调用为

# here obj is your notification object, you can send extra information in kwargs
send_notification.apply_async(kwargs={'obj_id':obj.id}, eta=obj.send_time)

注意：send_time应datetime。

Answer 2

您必须使用PeriodicTask和CrontabSchedule来计划可以从djcelery.models导入的任务。

所以代码将是这样的：

from djcelery.models import PeriodicTask, CrontabSchedule
crontab, created = CrontabSchedule.objects.get_or_create(minute='*/1')
periodic_task_obj, created = PeriodicTask.objects.get_or_create(name='send_notification', task='send_notification', crontab=crontab, enabled=True)

注意：您必须编写任务的完整路径，例如"app.tasks.send_notification"

---

您可以在通知模型post_save计划通知任务，如下所示：

@post_save
def schedule_notification(sender, instance, *args, **kwargs):
    """
    instance is notification model object
    """
    # create crontab according to your notification object.
    # there are more options you can pass like day, week_day etc while creating Crontab object.
    crontab, created = CrontabSchedule.objects.get_or_create(minute=instance.send_time.minute, hour=instance.send_time.hour)
    periodic_task_obj, created = PeriodicTask.objects.get_or_create(name='send_notification', task='send_notification_{}'.format(instance.pk))
    periodic_task_obj.crontab = crontab
    periodic_task_obj.enabled = True
    # you can also pass kwargs to your task like this
    periodic_task_obj.kwargs = json.dumps({"notification_id": instance.pk})
    periodic_task_obj.save()