So senden Sie periodische Aufgaben an bestimmte Warteschlangen in Sellery

Standardmäßig Sellerie senden alle Aufgaben, 'Sellerie' Warteschlange, aber Sie können dieses Verhalten ändern, durch hinzufügen von zusätzliche parameter:

@task(queue='celery_periodic')
def recalc_last_hour():
    log.debug('sending new task')
    recalc_hour.delay(datetime(2013, 1, 1, 2)) # for example

Scheduler-Einstellungen:

CELERYBEAT_SCHEDULE = {
   'installer_recalc_hour': {
        'task': 'stats.installer.tasks.recalc_last_hour',
        'schedule': 15  # every 15 sec for test
    },
}
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

Laufen Arbeiter:

python manage.py celery worker -c 1 -Q celery_periodic -B -E

Diese Regelung funktioniert nicht wie erwartet: diese Arbeiter sendet periodische Aufgaben 'Sellerie' queue, nicht 'celery_periodic'. Wie kann ich das beheben?

P. S. Sellerie==3.0.16

InformationsquelleAutor der Frage Artem Mezhenin | 2013-06-06

  1. 28

    Periodisch gesendet werden, um Warteschlangen celerybeat.Sie tun können alles, was wir tun, mit Sellerie-api. Hier ist die Liste der Konfigurationen kommt mit celerybeat.

    http://celery.readthedocs.org/en/latest/userguide/periodic-tasks.html#available-fields

    In Ihrem Fall

    CELERYBEAT_SCHEDULE = {
   'installer_recalc_hour': {
        'task': 'stats.installer.tasks.recalc_last_hour',
        'schedule': 15  # every 15 sec for test,
        'options': {'queue' : 'celery_periodic'} ##options are mapped to apply_async options
    },
}

    InformationsquelleAutor der Antwort abhi shukla

  2. 18

    Fand ich die Lösung für dieses problem:

    1) als Erstes änderte ich die Art und Weise zur Konfiguration von periodischen tasks. Ich verwendet @periodic_task decorator wie diese:

    @periodic_task(run_every=crontab(minute='5'),
               queue='celery_periodic',
               options={'queue': 'celery_periodic'})
def recalc_last_hour():
    dt = datetime.utcnow()
    prev_hour = datetime(dt.year, dt.month, dt.day, dt.hour) \
                - timedelta(hours=1)
    log.debug('Generating task for hour %s', str(prev_hour))
    recalc_hour.delay(prev_hour)

    2) ich schrieb celery_periodic zweimal in der params zu @periodic_task:

    • queue='celery_periodic' option wird verwendet, wenn Sie aufrufen Aufgabe von code (.Verzögerung oder .apply_async)

    • options={'queue': 'celery_periodic'} option wird verwendet, wenn Sellerie beat ruft es.

    Ich bin mir sicher, das gleiche ist möglich, wenn Sie möchten, konfigurieren Sie wiederkehrende Aufgaben mit CELERYBEAT_SCHEDULE variable.

    UPD. Diese Lösung korrekt für beide DB-basierende und Datei-basierte storage für CELERYBEAT_SCHEDULER.

    InformationsquelleAutor der Antwort Artem Mezhenin

  3. 2

    Und wenn Sie mit djcelery Datenbank-scheduler, können Sie die Warteschlange auf die Ausführung Optionen -> Feld Warteschlange ein

    InformationsquelleAutor der Antwort Mounir

Schreibe einen Kommentar