Sequentielle Aufgabenausführung in Sellerie

Question

Ich habe einige "schwere" Anfragen an die Datenbank, die ich mit Sellerie ausführen werde. Wenn man berücksichtigt, dass sie "schwer" sind, möchte ich sie nacheinander (nacheinander) ausführen. Eine mögliche Lösung besteht darin, --concurrency=1 in der Befehlszeile zu Sellery anzugeben. Und es funktioniert. Aber es gibt ein Problem: Während die Aufgaben alle folgenden Anfragen zurückgeben None ausgeführt werden:

from celery.task.control import inspect 

# Inspect all nodes. 
i = inspect() 

print(i.scheduled()) # None 
print(i.active()) # None 
print(i.reserved()) # None 
print(i.registered()) # None 

Auch läuft celery inspect ping kehrt Error: No nodes replied within time constraint. Damit ich keine Informationen über den Sellerie Warteschlangenstatus erhalten kann.

Es ist mein Test Python Module:

celeryconfig.py

#BROKER_URL = 'redis://localhost:6379/0' 
BROKER_URL = 'amqp://' 
#CELERY_RESULT_BACKEND = "redis" 
CELERY_RESULT_BACKEND = "amqp://" 
# for php 
CELERY_TASK_RESULT_EXPIRES = None 
CELERY_TASK_SERIALIZER = 'json' 
CELERY_RESULT_SERIALIZER = 'json' 
CELERY_ACKS_LATE = True 

tasks.py

from celery import Celery 
from time import sleep 

app = Celery('hello') 
app.config_from_object('celeryconfig') 

@app.task 
def add(x, y): 
    sleep(30)  
    return x + y 

client.py

from tasks import add 

result=add.delay(4, 4) 
result=add.delay(4, 4) 
result=add.delay(4, 4) 
result=add.delay(4, 4) 
result=add.delay(4, 4) 
result=add.delay(4, 4) 

Also, die Frage ist, wie die Aufgaben nacheinander ausgeführt werden UND in der Lage sein, den Status der Warteschlange zu überprüfen?

Answer 1

Selleries Inspektionen werden durchgeführt, indem eine Abfrage an alles gesendet wird, was zuhört, und dann die Antworten gesammelt werden. Jeder Worker, der nicht innerhalb des Timeouts antwortet (was meiner Meinung nach 1 Sekunde ist), wird ignoriert. Es ist, als ob es nicht existierte.

Die Tatsache, dass Sie --concurrency=1 verwenden, sollte kein Problem sein. Ich habe es gerade ausprobiert und es hat hier gut funktioniert. Selbst mit einer Nebenläufigkeit von 1 hat ein Sellerie-Arbeiter normalerweise einen zusätzlichen Ausführungsstrang für die Kommunikation. (Ich sage "normal", weil ich sicher bin, gibt es Möglichkeiten, Sellerie zu konfigurieren, um sich in den Fuß zu schießen. Was ich sage hält mit den Standardeinstellungen.) Als ich versuchte --concurrency=1 gab es tatsächlich zwei Threads pro Arbeiter. Selbst wenn der Worker damit beschäftigt ist, eine Aufgabe zu berechnen, sollte ein Thread in der Lage sein, auf die Übertragung zu antworten.

Dies gesagt, wenn die Maschine stark belastet wird, kann es zu lange dauern, bis der Arbeiter reagiert. Die Art, wie ich daran gearbeitet habe ist, Anrufe wie i.scheduled() zu wiederholen, bis ich eine Antwort von allen bekomme. In meinen Projekten weiß ich, wie viele Mitarbeiter arbeiten sollten und so habe ich eine Liste, die ich verwenden kann, um zu wissen, ob jeder geantwortet hat.