Jak przekierować łańcuch zadań do określonej kolejki w selerach?

Question

Kiedy trasa zadanie do konkretnej kolejki to działa:

task.apply_async(queue='beetroot') 

Ale jeśli utworzyć łańcuch:

chain = task | task 

A potem piszę

chain.apply_async(queue='beetroot') 

Wydaje ignorować słowo kluczowe kolejki i przypisuje do domyślnej kolejki "selekcji".

Byłoby miło, gdyby seler obsługiwał routing w łańcuchach - wszystkie zadania wykonywane sekwencyjnie w tej samej kolejce.

Answer 1

Ok Mam to wymyślone.

Trzeba dodać wymagane opcje wykonanie jak kolejce = lub odliczania = definicji podzadania lub poprzez częściowe:

definicji podzadanie:

from celery import subtask 

chain = subtask('task', queue = 'beetroot') | subtask('task', queue = 'beetroot') 

częściowy:

chain = task.s().apply_async(queue = 'beetroot') | task.s().apply_async(queue = 'beetroot') 

Następnie wykonuje się łańcuch przez:

chain.apply_async() 

lub

chain.delay() 

a zadania zostaną wysłane do kolejki „buraków”. Dodatkowe argumenty wykonania w tym ostatnim poleceniu nic nie dadzą. Byłoby miło zastosować wszystkie te argumenty wykonawcze na poziomie Łańcucha (lub Grupy lub dowolnych innych Elementów pierwotnych na płótnie).

Answer 2

robię to tak:

subtask = task.s(*myargs, **mykwargs).set(queue=myqueue) 
mychain = celery.chain(subtask, subtask2, ...) 
mychain.apply_async() 

Answer 3

Jest to dość późno, ale nie sądzę, kod świadczone przez @mpaf jest całkowicie poprawne.

Kontekst: W moim przypadku mam dwie podzadania, z których pierwsza dostarcza wartość zwracaną, która jest przekazywana do drugiej jako argument wejściowy. Miałem problemy z wykonaniem drugiego zadania do wykonania - widziałem w dziennikach, że Seler uznał drugie zadanie za callback pierwszego, ale nigdy nie wykona drugiego.

To był mój kod łańcuch non-working -:

from celery import chain 

chain(
    module.task1.s(arg), 
    module.task2.s() 
).apply_async(countdown=0.1, queue='queuename') 

Korzystanie składni przedstawione w odpowiedzi @ mpaf za, mam oba zadania do wykonania, ale kolejność wykonywania była przypadkowa, a drugi podzadanie nie został potwierdzony jako wywołanie zwrotne pierwszego. Mam pomysł, aby przejrzeć dokumenty, w jaki sposób jawnie ustawić kolejkę na podzadanie.

Jest to kod roboczy -:

chain(
    module.task1.s(arg).set(queue='queuename'), 
    module.task2.s().set(queue='queuename') 
).apply_async(countdown=0.1)