Sellerie ein Broker mehrere Warteschlangen und Arbeiter

Question

Ich habe eine Python-Datei namens tasks.py, in dem ich 4 einzelne Aufgaben definieren. Ich möchte Sellerie konfigurieren, um 4 Warteschlangen zu verwenden, da jeder Warteschlange eine andere Anzahl von Arbeitern zugewiesen ist. Ich habe gelesen, dass ich route_task Eigentum verwenden sollte, aber ich versuchte mehrere Optionen und kein Erfolg.

Ich folgte diesem doc celery route_tasks docs

Mein Ziel 4 Arbeiter, für jede Aufgabe ein ausgeführt werden würde, und sie Aufgaben nicht aus verschiedenen Arbeiter in verschiedenen Warteschlangen mischen. Es ist möglich? Es ist ein guter Ansatz?

Wenn ich etwas tue, falsch ich glücklich sein würde, meinen Code zu ändern, um es

funktioniert

Hier meine Config ist so weit

tasks.py

app = Celery('tasks', broker='pyamqp://guest@localhost//') 
app.conf.task_default_queue = 'default' 
app.conf.task_queues = (
    Queue('queueA', routing_key='tasks.task_1'), 
    Queue('queueB', routing_key='tasks.task_2'), 
    Queue('queueC', routing_key='tasks.task_3'), 
    Queue('queueD', routing_key='tasks.task_4') 
) 


@app.task 
def task_1(): 
    print "Task of level 1" 


@app.task 
def task_2(): 
    print "Task of level 2" 


@app.task 
def task_3(): 
    print "Task of level 3" 


@app.task 
def task_4(): 
    print "Task of level 4" 

Run Sellerie ein Arbeiter für jede Warteschlange

celery -A tasks worker --loglevel=debug -Q queueA --logfile=celery-A.log -n W1& 
celery -A tasks worker --loglevel=debug -Q queueB --logfile=celery-B.log -n W2& 
celery -A tasks worker --loglevel=debug -Q queueC --logfile=celery-C.log -n W3& 
celery -A tasks worker --loglevel=debug -Q queueD --logfile=celery-D.log -n W4& 

Answer 1

Es ist nicht notwendig, in komplexe Routing für subm zu bekommen Aufgaben in verschiedene Warteschlangen einfügen. Definieren Sie Ihre Aufgaben wie gewohnt.

from celery import celery 

app = Celery('tasks', broker='pyamqp://guest@localhost//') 

@app.task 
def task_1(): 
    print "Task of level 1" 


@app.task 
def task_2(): 
    print "Task of level 2" 

Jetzt, während die Aufgaben in die Warteschlange gestellt werden, die Aufgaben in die richtige Warteschlange stellen. Hier ist ein Beispiel, wie es geht.

In [12]: from tasks import * 

In [14]: result = task_1.apply_async(queue='queueA') 

In [15]: result = task_2.apply_async(queue='queueB') 

Dies wird die task_1 in der Warteschlange queueA und task_2 in queueB Namen setzen.

Jetzt können Sie Ihre Arbeiter einfach starten, um sie zu verbrauchen.

celery -A tasks worker --loglevel=debug -Q queueA --logfile=celery-A.log -n W1& 
celery -A tasks worker --loglevel=debug -Q queueB --logfile=celery-B.log -n W2&