Django/Seler wiele kolejek na localhost - routing nie działa

Question

Obserwowałem selera docs, aby zdefiniować 2 kolejki na moim komputerze.

Moje ustawienia seler:

CELERY_ALWAYS_EAGER = True 
CELERY_TASK_RESULT_EXPIRES = 60 # 1 mins 
CELERYD_CONCURRENCY = 2 
CELERYD_MAX_TASKS_PER_CHILD = 4 
CELERYD_PREFETCH_MULTIPLIER = 1 
CELERY_CREATE_MISSING_QUEUES = True 
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'), 
    Queue('feeds', Exchange('feeds'), routing_key='arena.social.tasks.#'), 
) 
CELERY_ROUTES = { 
    'arena.social.tasks.Update': { 
     'queue': 'fs_feeds', 
    }, 
} 

i otwarte dwa okna terminala, w virtualenv mojego projektu i pobiegł następujące polecenia:

terminal_1$ celery -A arena worker -Q default -B -l debug --purge -n deafult_worker 
terminal_2$ celery -A arena worker -Q feeds -B -l debug --purge -n feeds_worker 

co ja dostać jest to, że wszystkie zadania są przetwarzane przez obie kolejki.

Moim celem jest posiadanie jednej kolejki do przetwarzania tylko jednego zadania zdefiniowanego w CELERY_ROUTES i domyślnej kolejki do przetwarzania wszystkich innych zadań.

ja również po tej SO question, rabbitmqctl list_queues powraca celery 0 i działa rabbitmqctl list_bindings powraca exchange celery queue celery [] dwukrotnie. Restartowanie króliczego serwera niczego nie zmieniło.

Answer 1

Ok, więc to wymyśliłem. Poniżej znajduje się cała moja konfiguracja, ustawienia i sposób prowadzenia selera, dla tych, którzy mogą zastanawiać się nad tym samym, co moje pytanie.

Ustawienia

CELERY_TIMEZONE = TIME_ZONE 
CELERY_ACCEPT_CONTENT = ['json', 'pickle'] 
CELERYD_CONCURRENCY = 2 
CELERYD_MAX_TASKS_PER_CHILD = 4 
CELERYD_PREFETCH_MULTIPLIER = 1 

# celery queues setup 
CELERY_DEFAULT_QUEUE = 'default' 
CELERY_DEFAULT_EXCHANGE_TYPE = 'topic' 
CELERY_DEFAULT_ROUTING_KEY = 'default' 
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'), 
    Queue('feeds', Exchange('feeds'), routing_key='long_tasks'), 
) 
CELERY_ROUTES = { 
    'arena.social.tasks.Update': { 
     'queue': 'feeds', 
     'routing_key': 'long_tasks', 
    }, 
} 

Jak uruchomić seler?

zacisk - zakładka 1:

celery -A proj worker -Q default -l debug -n default_worker 

ten rozpocznie pierwszy pracownik, który zużywa zadania z domyślnej kolejki. UWAGA! -n default_worker nie jest koniecznością dla pierwszego pracownika, ale jest koniecznością, jeśli masz uruchomione jakiekolwiek inne instancje selera. Ustawienie -n worker_name jest takie samo jak --hostname=default@%h.

zacisk - zakładka 2:

celery -A proj worker -Q feeds -l debug -n feeds_worker 

ten rozpocznie drugi robotnik, że konsumenci zadania z zewnętrznej kolejki. Uwaga: -n feeds_worker, jeśli używasz -l debug (log level = debug), zobaczysz, że obaj pracownicy synchronizują między nimi.

zacisk - zakładka 3:

celery -A proj beat -l debug 

ten rozpocznie bicie, wykonywanie zadań zgodnie z harmonogramem w swojej CELERYBEAT_SCHEDULE. Nie musiałem zmieniać zadania ani CELERYBEAT_SCHEDULE.

Na przykład, to jak wygląda moich CELERYBEAT_SCHEDULE dla zadania, które powinny Idź do kan kolejki:

CELERYBEAT_SCHEDULE = { 
    ... 
    'update_feeds': { 
     'task': 'arena.social.tasks.Update', 
     'schedule': crontab(minute='*/6'), 
    }, 
    ... 
} 

Jak widać, nie ma potrzeby dodawania 'options': {'routing_key': 'long_tasks'} lub określając, co powinno iść w kolejce.Ponadto, jeśli zastanawiasz się, dlaczego Update jest wielkimi literami, jest to jego niestandardowe zadanie zdefiniowane jako podklasy celery.Task.