设置动态调度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 字段安排它。

Note: i am using new integration of celery with django not django-celery package

Answer 1

您必须使用 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()

Answer 2

要在指定的日期和时间执行任务，您可以在调用 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)

Note: send_time should be datetime.

设置动态调度celerybeat

Set dynamic scheduling celerybeat

django

celery

celerybeat