Rückruf für Sellerie apply_async

Question

Ich verwende celery in meiner Anwendung, um periodische Aufgaben auszuführen. Lassen Sie uns einfach folgende Beispiel

from myqueue import Queue 
@perodic_task(run_every=timedelta(minutes=1)) 
def process_queue(): 
    queue = Queue() 
    uid, questions = queue.pop() 
    if uid is None: 
     return 

    job = group(do_stuff(q) for q in questions) 
    job.apply_async() 

def do_stuff(question): 
    try: 
     ... 
    except: 
     ... 
     raise 

siehe Wie Sie im obigen Beispiel sehen können, verwende ich celery async Aufgabe zu laufen, aber (da es eine Warteschlange ist) muss ich queue.fail(uid) bei Ausnahme in do_stuff oder queue.ack(uid) sonst tun . In dieser Situation wäre es sehr klar und nützlich, in beiden Fällen einen Rückruf von meiner Aufgabe zu erhalten - on_failure und on_success.

Ich sah einige documentation, aber nie Praktiken der Verwendung von Callbacks mit apply_async gesehen. Ist es möglich, das zu tun?

Answer 1

Subclass der Task-Klasse und die on_success und ON_FAILURE Funktionen überlasten:

class CallbackTask(Task): 
    def on_success(self, retval, task_id, args, kwargs): 
     pass 

    def on_failure(self, exc, task_id, args, kwargs, einfo): 
     pass 


@celery.task(base=CallbackTask) # this does the trick 
def add(x, y): 
    return x + y 

Answer 2

Sie können Erfolg und Fehler Rückrufe über den Link und LINK_ERR kwargs angeben, wenn Sie anrufen apply_async. Die Sellerie-Dokumentation enthält ein klares Beispiel: http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks