设置动态调度 celerybeat

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 字段安排它。

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

Answer 1

要在指定的日期和时间执行任务，您可以在调用docs 中提到的任务时使用apply_async 的eta 属性

创建通知对象后，您可以将您的任务称为

# 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()