Gọi lại cho cần tây apply_async

Question

Tôi sử dụng celery trong ứng dụng của mình để chạy các tác vụ định kỳ. Hãy xem ví dụ đơn giản dưới đây

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 

Như bạn có thể thấy trong ví dụ trên, tôi sử dụng celery để chạy nhiệm vụ async, nhưng (vì nó là một hàng đợi) tôi cần phải làm queue.fail(uid) trong trường hợp ngoại lệ trong do_stuff hoặc queue.ack(uid) khác . Trong tình huống này, sẽ rất rõ ràng và hữu ích khi có một số cuộc gọi lại từ nhiệm vụ của tôi trong cả hai trường hợp - on_failure và on_success.

Tôi đã xem một số documentation, nhưng chưa bao giờ thấy thực tiễn sử dụng gọi lại với apply_async. Có thể làm điều đó không?

Answer 1

Subclass lớp công tác và quá tải chức năng on_success và on_failure:

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

Bạn có thể chỉ định thành công và lỗi callbacks qua liên kết và link_err kwargs khi bạn gọi apply_async. Tài liệu cần tây bao gồm một ví dụ rõ ràng: http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks