設置動態調度celerybeat

Question

我在我的Notification模型中有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

要在指定的日期和時間執行任務yo ü可以使用的apply_asynceta屬性，而在docs

提到創作的通知後調用任務對象，你可以打電話給你的任務，

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