13

celeryアプリケーションで定期的なタスクを実行するために使用します。以下の簡単な例を見てみましょう

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

上記の例でわかるように、非同期タスクを実行するために使用しますが、(キューであるため)例外が発生した場合やその他の場合celeryに実行する必要があります。この状況では、 と の両方の場合にタスクからのコールバックがあると非常に明確で便利です。queue.fail(uid)do_stuffqueue.ack(uid)on_failureon_success

いくつかのドキュメントを見ましたが、でコールバックを使用する方法は見たことがありませんapply_async。それは可能ですか？

4

2 に答える 2

45

Task クラスをサブクラス化し、on_success および 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
于 2012-11-21T13:27:27.073 に答える
8

apply_async を呼び出すときに、link および link_err kwargs を介して成功およびエラーのコールバックを指定できます。セロリのドキュメントには明確な例が含まれています: http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks

于 2013-01-04T16:41:17.213 に答える