Sellerie Aufruf andere Funktion und weiterhin die Verkettung

Question

Ich habe Aufgaben in einer Kette mit drei Aufgaben fetch_page, check_source und Store-Seite

def update_page_info(**headers): 
    chain=fetch_page.s(headers['key']) | check_source.s(headers['key_1']) | store_info.s() 
    chain().apply_async() 

fetch_page die Seite abruft und sammelt, was es sammeln muss:

@app.task(bind=True) 
def fetch_page(self,url): 
    #fetch_page here and return a tuple so that it can be unpacked 
    # dosomething 

Nach dem Abrufen der Seite wird nun in den nächsten Aufgaben check_source nach der Quelle gesucht.

so die Verwirrung ist kann ich einige andere Aufgaben von hier aufrufen ?? Wird es Unstimmigkeiten geben oder wird der Arbeiter jemals in eine Sackgasse geraten?

Und schließlich sollte es ausführen store_info(), die von check_source nur speichern Dinge zurückgegeben würden()

@app.task(bind=True) 
def store_info(self,result): 
    print ("store_info ") 
    try: 
     #store the fetched pages 

    except Exception as exc: 
     #dosomething 
    finally: 
     pass 

Ich war nach diesem Ansatz, der nur wenig Änderung erforderlich http://docs.celeryproject.org/en/latest/userguide/tasks.html#avoid-launching-synchronous-subtasks.

Kann jemand mir vorschlagen, wie es gemacht werden sollte und Dinge, um die ich vorsichtiger sein muss?

Answer 1

Dies sollte alles funktionieren, wie Sie es lesen (und kommunizieren) sollte. Die drei Aufgaben werden ohne "Inkonsistenz" ausgeführt.

Wenn Sie einmal aufrufen, werden die drei verketteten Teilaufgaben ausschließlich voneinander ausgeführt. Allerdings besteht bei diesem Setup immer noch das Potenzial für Deadlocks. Wenn Sie update_page_info aufgerufen haben, während vorherige Aufgaben von dem letzten Mal, als Sie es aufgerufen haben, ausgeführt wurden, können Sie mehrere Aufgaben auf einmal ausführen. Dies würde das Potenzial für Deadlocks abhängig davon, wie Ihre Aufgaben Ressourcen gemeinsam nutzen, einführen.

Wenn Ihre Aufgaben Ressourcen teilen, würde ich vorschlagen, etwas wie redis oder memcached als ein Schließsystem über Arbeiter zu verwenden.

Edit: der Code, den ich jetzt sehe, ist völlig in Ordnung, da die Ergebnisse als Parameter an die nächste Aufgabe weitergegeben werden.