基礎入門_Python-模塊和包.深入Celery之子任務及原語組式/鏈式/回調

2019-11-08 01:54:35

字體：大中小

來源：轉載

供稿：網友

子級任務:

說明: 前面學習的任務調用基本上已夠用,但有時業務場景可能需要任務關聯,且子任務可能需要父任務為其傳遞一些必要參數,此時可通過func.subtask或簡寫func.s實現,子任務對象依然可以調用任務對象的API,delay和apply_async,但和常規調用不同的是子任務調用時傳遞的參數加生成子任務時初始化參數的才是最終任務的參數,所以子任務在調用時就不需要再重復傳遞子任務初始化時的參數了.

1234567 >>> mail = send_mail.subtask(('sub', 'msg'), countdown=1)>>> mail.delay('limanman')<AsyncResult: de491985-5170-43c3-905f-f899f3aef460>或>>> mail = send_mail.s('sub', 'msg', countdown=1)>>> mail.delay('limanman')<AsyncResult: c67194e0-16e1-47a8-a4bb-420f4966b7f1>

注意: send_mail.subtask(('sub', 'msg'), countdown=1)或send_mail.s('sub', 'msg', countdown=1)返回的是Signature對象,此對象攜帶生成子任務時的參數,且可調用任務對象的API,當調用時,傳遞的參數被提前.

原語操作:

# 使組式:

123	`>>>` `from` `celery` `import` `group>>> group(send_mail.s(i, i, i)` `for` `i` `in` `xrange(10))().get()>>> group(send_mail.s('msg_{0}'.format(i))` `for` `i` `in` `xrange(10))('usr',` `'sub').get()`

說明: group內部使用列表生成式生成了10個子任務,生成時且提供了一個參數,當調用group('usr', 'sub')時其實是遍歷group內部的所有子任務調用其.delay('usr', 'sub'),所以是并發異步執行的,上面說過,調用時傳遞的參數會被提前,所以其實最終的參數順序是'usr', 'sub', 'msg_{0}'.format(i),異步返回的是GrouPResult對象,如果要獲取每個子任務的結果,可通過同步阻塞的get方法獲取.

# 使鏈式:

12345 >>> from celery import chain>>> result = chain(add.s(2) | add.s(1))(5)>>> result.get()>>> result.parent.get()>>> result.parent.parent.get()

說明: chain內部兩個鏈式子任務,|表示前一個的輸出作為另一個的輸入,當調用chain(...)(5)時,其實是首先調用add.s(2).delay(5)計算出結果為7,然后|傳遞作為add.s(1)的輸入,其實是調用add.s(1).delay(7)

計算出結果為8,異步返回的是AsyncResult對象,可通過同步阻塞的get方法獲取,鏈式操作還有一個特別的地方可通過result.parent.get()獲取上層結果.

# 使回調:

123 >>> from celery import chain group chord>>> chord((send_mail.s('usr_{0}'.format(i), 'sub_{0}'.format(i), 'msg_{0}'.format(i)) for i in xrange(10)), send_mail.s('sub', 'msg'))().get()>>> chain(group(send_mail.s('usr_{0}'.format(i), 'sub_{0}'.format(i), 'msg_{0}'.format(i)) for i in xrange(10))|send_mail.s('sub', 'msg'))().get()