簽名

2.0 版本新特性。

剛剛在calling 這一節中學習了使用 delay 方法呼叫任務，並且通常這就是你所需要的，但是有時候你可能想將一個任務呼叫的簽名傳遞給另外一個程序或者作為另外一個函式的引數。

任務簽名包含了一次任務呼叫的引數、關鍵字引數以及執行選項資訊，它可以傳遞給其他函式，甚至序列化後通過網路傳輸。

• 你使用 add 任務的名稱建立一個簽名，就像這樣：

>>> from celery import signature
>>> signature('tasks.add', args=(2, 2), countdown=10)
tasks.add(2
 
, 2)

這個任務簽名有兩個引數：（2，2），並且 countdown 執行選項被設定成 10。

• 或者你也可以使用任務的 signature 方法建立任務簽名：

>>> add.signature((2, 2), countdown=10)
tasks.add(2, 2)

• 還有一個使用引數展開的快捷方式：

>>> add.s(2, 2)
tasks.add(2, 2)

• 關鍵字引數也是支援的：

>>> add.s(2, 2, debug=True)
tasks.add(2, 2, debug=True)

• 從任何簽名例項，你都可以探查它的不同欄位：

>>> s = add.signature((2, 2), {'debug': True}, countdown=10)
>>> s.args
(2, 2)
>>> s.kwargs
{'debug': True}
>>> s.options
{'countdown': 10}

• 簽名支援 “Calling API”，如 delay、apply_async 等等，包括直接呼叫（__call__）。

直接呼叫簽名會在當前程序中執行任務：

>>> add(2, 2)
4
>>> add
 
.s(2, 2)()
4

delay 是 apply_async 函式的星號引數展開快捷方式：

>>> result = add.delay(2, 2)
>>> result.get()
4

apply_async 方法與 app.Task.apply_async() 方法的引數相同：

>>> add.apply_async(args, kwargs, **options)
>>> add.signature(args, kwargs, **options).apply_async()

>>> add.apply_async((2, 2), countdown=1)
>>> add.signature((2, 2), countdown=1).apply_async()

• 使用 s() 不能定義執行選項，但是提供了一個鏈式的 set 函式來處理這些：

>>> add.s(2, 2).set(countdown=1)
proj.tasks.add(2, 2)

Partials

使用簽名，你可以在工作單元中執行任務：

>>> add.s(2, 2).delay()
>>> add.s(2, 2).apply_async(countdown=1)

或者你可以在當前程序中直接呼叫任務：

>>> add.s(2, 2)()
4

通過 apply_async/delay 宣告額外的引數、關鍵字引數、執行選項來實現偏函式的功能：

• 任何新加的引數都會前置到簽名的當前引數列表

>>> partial = add.s(2)          # incomplete signature
>>> partial.delay(4)            # 4 + 2
>>> partial.apply_async((4,))  # same

• 任何新加的關鍵字引數都會與簽名現有的關鍵字引數合併，優先選擇新的關鍵字引數：

>>> s = add.s(2, 2)
>>> s.delay(debug=True)                    # -> add(2, 2, debug=True)
>>> s.apply_async(kwargs={'debug': True})  # same

• 任何新加的執行選項都會與簽名的執行選項合併，優先選擇新的執行選項：

>>> s = add.signature((2, 2), countdown=10)
>>> s.apply_async(countdown=1)  # countdown is now 1

你還可以通過克隆任務來衍生出新任務：

>>> s = add.s(2)
proj.tasks.add(2)

>>> s.clone(args=(4,), kwargs={'debug': True})
proj.tasks.add(4, 2, debug=True)

Immutability

3.0 版本新特性。

便函式註定要和回撥函式一起使用，任何連結任務、或者 chord 回撥函式都會使用父任務的結果作為函式呼叫引數。有時候，你想宣告一個不帶任何附加引數的回撥函式，此時你可以將簽名設定成不可變的。

>>> add.apply_async((2, 2), link=reset_buffers.signature(immutable=True))

可以使用 .si() 快捷方式建立不可變簽名：

>>> add.apply_async((2, 2), link=reset_buffers.si())

當任務簽名是不可變的時，只有執行選項可以被設定，所以不能以部分引數/關鍵字引數呼叫簽名。

注意：
這個手冊中我在簽名前使用字首操作符~。你最好不要在生產環境這麼使用，但是在 python 控制檯做些實驗是比較方便的：

>>> ~sig

>>> # is the same as
>>> sig.delay().get()

Callbacks

3.0 版本新特性。

回撥函式可以通過使用 apply_async 函式的 link 引數新增到任何任務：

add.apply_async((2, 2), link=other_task.s())

回撥函式只有在任務成功退出才會呼叫，並且它會將父任務的返回結果作為引數。

如我簽名所描述的，你給簽名新增的任何引數，都會置於簽名本身宣告的引數之前！

如果你有如下簽名：

>>> sig = add.s(10)

那麼，sig.delay(result) 變成：

>>> add.apply_async(args=(result, 10))

…

現在，我們呼叫任務 add，設定回撥函式並使用部分引數：

>>> add.apply_async((2, 2), link=add.s(8))

如所預期的，它將首先啟動一個任務計算 2+2，然後啟動另一個任務計算 4+8。

原語

3.0 版本新特性。

概要
- group
組元語是一個簽名，引數是要併發執行的任務的列表

• chain
  鏈元語讓我們可以將簽名連結起來，使得一個任務呼叫後執行另外一個任務，本質上形成一個回撥鏈。

• chord
  弦就像帶有一個回撥函式的組。弦由一個頭部組和絃體組成，而弦體是頭部組中所有任務都完成之後應該執行的任務。

• map
  對映元語就像內建的 map 函式，但是它還建立一個帶有引數列表的臨時任務。例如，task.map([1, 2]) - 將呼叫一個任務，引數被按序傳給任務，所以結果是：

res = [task(1), task(2)]

• starmap
  除了引數被展開，其餘都和 map 相同。例如，add.starmap([(2,2), (4,4)])將會生成如下任務呼叫：

res = [add(2,2), add(4,4)]

• chunks
  分塊將一個長引數列表分解成，例如：

>>>items = zip(xrange(1000), xrange(1000))  #1000 items
>>>add.chunks(items, 10)

會將10個項為一塊，分成100塊，從而產生100個任務（每個任務處理10個項）

這些元語本身也是簽名，所以他們可以任意組合，形成複雜的工組流。

下面是一些例子：

• 簡單的鏈
  這是一個簡單的鏈，第一個任務執行完將它的返回結果傳遞給下一個任務，以此類推。

>>> from celery import chain

>>> # 2 + 2 + 4 + 8
>>> res = chain(add.s(2, 2), add.s(4), add.s(8))()
>>> res.get()
16

這還可以使用管道符寫：

>>> (add.s(2, 2) | add.s(4) | add.s(8))().get()
16

• 不可變簽名
  簽名可以是部分的，所以可以在現有引數基礎上新增引數，但是有時候你可能並不想這樣，例如你不想要鏈中前面任務的返回值。

這種情況下，你可以將簽名標記為不可變的，則引數不能再改變：

>>> add.signature((2, 2), immutable=True)

還有一個快捷方式 si()方法，並且這是建立簽名的推薦方式：

>>>add.si(2,2)

現在可以建立獨立任務組成的鏈：

>>> res = (add.si(2, 2) | add.si(4, 4) | add.si(8, 8))()
>>> res.get()
16

>>> res.parent.get()
8

>>> res.parent.parent.get()
4

• 簡單的組
  你可以很容易建立一個併發執行的任務：

>>> from celery import group
>>> res = group(add.s(i, i) for i in xrange(10))()
>>> res.get(timeout=1)
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

• 簡單的弦
  弦使得我們可以新增一個回撥函式，在組中所有任務都執行完成後，該回調函式將被呼叫。非易並行的演算法常常需要這種方式：

>>> from celery import chord
>>> res = chord((add.s(i, i) for i in xrange(10)), xsum.s())()
>>> res.get()
90

上面這個例子建立了並行執行的10個任務，當所有的任務執行完成，他們的返回值組成一個列表傳遞給 xsum 任務。

弦體也可以是不可變的，此時組任務的返回值不傳遞給回撥函式：

>>> chord((import_contact.s(c) for c in contacts),
...       notify_complete.si(import_id)).apply_async()

注意上面使用的 si；建立了一個不可變的簽名，意味著所有傳遞的新引數都被忽略（包括前面任務的返回值）。

• 組合
  鏈也可以是部分的：

>>> c1 = (add.s(4) | mul.s(8))

# (16 + 4) * 8
>>> res = c1(16)
>>> res.get()
160

著意味著你可以組合鏈：

# ((4 + 16) * 2 + 4) * 8
>>> c2 = (add.s(4, 16) | mul.s(2) | (add.s(4) | mul.s(8)))

>>> res = c2()
>>> res.get()
352

將組和另外一個任務鏈在一起將自動升級成弦:

>>> c3 = (group(add.s(i, i) for i in xrange(10)) | xsum.s())
>>> res = c3()
>>> res.get()
90

組和絃也接收部分引數，所以在一個鏈中，前一個任務的返回值將傳遞給組中所有任務：

>>> new_user_workflow = (create_user.s() | group(
...                      import_contacts.s(),
...                      send_welcome_email.s()))
... new_user_workflow.delay(username='artv',
...                         first='Art',
...                         last='Vandelay',
...                         email='[email protected]')

如果你不想傳遞引數給組，那麼讓組中的簽名不可變：

>>> res = (add.s(4, 4) | group(add.si(i, i) for i in xrange(10)))()
>>> res.get()
<GroupResult: de44df8c-821d-4c84-9a6a-44769c738f98 [
    bc01831b-9486-4e51-b046-480d7c9b78de,
    2650a1b8-32bf-4771-a645-b0a35dcc791b,
    dcbee2a5-e92d-4b03-b6eb-7aec60fd30cf,
    59f92e0a-23ea-41ce-9fad-8645a0e7759c,
    26e1e707-eccf-4bf4-bbd8-1e1729c3cce3,
    2d10a5f4-37f0-41b2-96ac-a973b1df024d,
    e13d3bdb-7ae3-4101-81a4-6f17ee21df2d,
    104b2be0-7b75-44eb-ac8e-f9220bdfa140,
    c5c551a5-0386-4973-aa37-b65cbeb2624b,
    83f72d71-4b71-428e-b604-6f16599a9f37]>

>>> res.parent.get()
8

Chains

3.0版本新特性。

任務可以鏈在一起：如果當前任務成功執行，被連結的任務將會被啟動：

>>> res = add.apply_async((2, 2), link=mul.s(16))
>>> res.get()
4

被連結的任務將會用父任務的返回值作為它的第一個引數。上面這個例子當前任務返回值是4，將會執行 mul(4, 16)。

結果將跟蹤被原任務呼叫的任意子任務，而且這可以從結果例項中訪問到：

>>> res.children
[<AsyncResult: 8c350acf-519d-4553-8a53-4ad3a5c5aeb4>]

>>> res.children[0].get()
64

結果例項還有一個 collect() 方法，它將結果視作圖，使得你可以在結果上迭代：

>>> list(res.collect())
[(<AsyncResult: 7b720856-dc5f-4415-9134-5c89def5664e>, 4),
 (<AsyncResult: 8c350acf-519d-4553-8a53-4ad3a5c5aeb4>, 64)]

預設情況下，如果結果圖沒有完全形成，collect()方法將會丟擲一個 IncompleteStream 異常（至少有一個任務未完成），但是你可以獲取結果圖的一箇中間表示形式：

>>> for result, value in res.collect(intermediate=True)):
....

只要你願意，你可以連結任意多的任務，並且簽名也可以被連結：

>>> s = add.s(2, 2)
>>> s.link(mul.s(4))
>>> s.link(log_result.s())

你還可以使用on_error方法新增錯誤回撥函式：
>>> add.s(2, 2).on_error(log_error.s()).delay()

當任務簽名被應用，將導致下面呼叫發生：

>>> add.apply_async((2, 2), link_error=log_error.s())

工作單元實際上不會將錯誤回撥作為任務執行，而是會直接呼叫回撥函式，這使得原始請求、異常、堆疊回溯物件都可以被傳遞給錯誤回撥。

下面是一個錯誤回撥的例子：

from __future__ import print_function

import os

from proj.celery import app

@app.task
def log_error(request, exc, traceback):
    with open(os.path.join('/var/errors', request.id), 'a') as fh:
        print('--\n\n{0} {1} {2}'.format(
            task_id, exc, traceback), file=fh)

為了使連結任務更簡便，有一個特殊的任務簽名 chain 可以將任務連結到一起：

>>> from celery import chain
>>> from proj.tasks import add, mul

>>> # (4 + 4) * 8 * 10
>>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))
proj.tasks.add(4, 4) | proj.tasks.mul(8) | proj.tasks.mul(10)

呼叫任務鏈將在當前程序呼叫任務，並且返回鏈中最後一個任務的返回值：

>>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))()
>>> res.get()
640

它還將設定 parent 屬性，這使得你可以以你自己的方式獲取中間結果：

>>> res.parent.get()
64

>>> res.parent.parent.get()
8

>>> res.parent.parent
<AsyncResult: eeaad925-6778-4ad1-88c8-b2a63d017933>

鏈還可以通過管道符建立：

>>> (add.s(2, 2) | mul.s(8) | mul.s(10)).apply_async()

圖

除此之外，你還可以將結果圖作為依賴圖使用：

>>> res = chain(add.s(4, 4), mul.s(8), mul.s(10))()

>>> res.parent.parent.graph
285fa253-fcf8-42ef-8b95-0078897e83e6(1)
    463afec2-5ed4-4036-b22d-ba067ec64f52(0)
872c3995-6fa0-46ca-98c2-5a19155afcf0(2)
    285fa253-fcf8-42ef-8b95-0078897e83e6(1)
        463afec2-5ed4-4036-b22d-ba067ec64f52(0)

你甚至可以將圖轉化為dot格式：

>>> with open('graph.dot', 'w') as fh:
...     res.parent.parent.graph.to_dot(fh)

然後建立圖片：

$ dot -Tpng graph.dot -o graph.png

Groups

3.0版本新特性。

組可以用來執行並行任務。

組函式接收多個任務簽名作為引數：

>>> from celery import group
>>> from proj.tasks import add

>>> group(add.s(2, 2), add.s(4, 4))
(proj.tasks.add(2, 2), proj.tasks.add(4, 4))

如果你呼叫組，組中的任務將會在當前程序中一個接一個被啟動，並且返回一個 GroupResult 例項，用來跟蹤結果，或者告訴你有多少給任務已經成功執行等等：

>>> g = group(add.s(2, 2), add.s(4, 4))
>>> res = g()
>>> res.get()
[4, 8]

組也支援迭代：

>>> group(add.s(i, i) for i in xrange(100))()

組是一個簽名，所以也可以和其他簽名組合。

組結果

組任務返回一個特殊的結果，這個結果就像普通任務一樣使用，只不過它將組作為一個整體看待：

>>> from celery import group
>>> from tasks import add

>>> job = group([
...             add.s(2, 2),
...             add.s(4, 4),
...             add.s(8, 8),
...             add.s(16, 16),
...             add.s(32, 32),
... ])

>>> result = job.apply_async()

>>> result.ready()  # have all subtasks completed?
True
>>> result.successful() # were all subtasks successful?
True
>>> result.get()
[4, 8, 16, 32, 64]

GroupResult 包含一個 AsyncResult 例項的列表，並且就像單個任務一樣操作。

組結果包含如下操作：

• successful()
  如果所有的子任務都成功執行，那麼返回 True（例如，沒有丟擲異常）。

• failed()
  如果任意子任務失敗，那麼返回 True。

• waiting()
  如果任意子任務沒有執行完成，那麼返回 True。

• ready()
  如果所有的任務都執行完成，那麼返回 True。

• completed_count()
  返回完成的子任務數。

• revoke()
  取消所有子任務。

• join()
  收集所有子任務的返回值，並按照他們呼叫的順序返回（作為一個列表）。

Chords

2.3 版本新特性。

注意：
在一個弦中應用的任務不能忽略任務的返回值。如果結果後端對弦中任何任務（弦頭部或者弦體）禁用，你應該閱讀“Important Notes”這一節。弦現在還不支援 RPC 儲存後端。

弦是一個任務，只有弦頭任務組中所有任務都執行完成，弦體任務才會執行。

我們來計算表示式 1+1+2+2+3+3+...+n+n 一直到100。

首先，你需要兩個任務，add() 以及 tsum() (sum() 是一個標準函式)：

@app.task
def add(x, y):
    return x + y

@app.task
def tsum(numbers):
    return sum(numbers)

現在，你可以使用弦來並行執行疊加步驟，然後計算所有疊加結果的和：

>>> from celery import chord
>>> from tasks import add, tsum

>>> chord(add.s(i, i)
...       for i in xrange(100))(tsum.s()).get()
9900

這明顯是一個很勉強的例子，訊息傳遞和同步的耗費使得它要比python裡直接如下計算要慢得多：

>>> sum(i + i for i in xrange(100))

同步的耗費非常大，所以你應該儘量避免使用弦。不過話又說回來，因為同步是許多並行演算法所需要的，所以弦仍然是你工具箱裡一個強大的元語。

我們一步步來定義弦：

>>> callback = tsum.s()
>>> header = [add.s(i, i) for i in range(100)]
>>> result = chord(header)(callback)
>>> result.get()
9900

記住，回撥函式只有在弦頭任務組中所有任務都返回後才執行。頭任務組中的每步都作為一個任務執行，可能在不同的節點上執行。回撥函式將應用每個任務的返回值作為引數。chord() 返回的任務ID 是回撥函式的 ID，所以你可以等待它完成並拿到它的返回值（但是記住永遠不要讓一個任務等待其他任務）。

錯誤處理

如果其中一個任務丟擲異常將會發生什麼呢？

弦的回撥函式結果將轉化為failure狀態， 並且錯誤被設定成ChordError異常：

>>> c = chord([add.s(4, 4), raising_task.s(), add.s(8, 8)])
>>> result = c()
>>> result.get()

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "*/celery/result.py", line 120, in get
    interval=interval)
  File "*/celery/backends/amqp.py", line 150, in wait_for
    raise meta['result']
celery.exceptions.ChordError: Dependency 97de6f3f-ea67-4517-a21c-d867c61fcb47
    raised ValueError('something something',)

根據結果儲存後端的不同，堆疊回溯資訊也不一樣，可以檢視錯誤描述，這包括失敗任務的ID和原異常的字串表示。你還可以在result.traceback中找到原異常堆疊回溯資訊。

注意餘下的任務仍然會繼續執行，所以即使中間這個任務失敗，第三個任務（add.s(8,8)）依舊執行。另外，ChordError 只顯示首先失敗的任務（實時）：它不關心任務組中的順序。

當弦失敗時執行一個操作，你可以給弦回撥新增一個錯誤回撥函式：

@app.task
def on_chord_error(request, exc, traceback):
    print('Task {0!r} raised error: {1!r}'.format(request.id, exc))

>>> c = (group(add.s(i, i) for i in range(10)) |
...      xsum.s().on_error(on_chord_error.s()))).delay()

Important Notes
弦中的任務不可以忽略返回值。實際操作中，這意味著為了使用弦你必須啟用result_backend。另外，如果你的配置中task_ignore_result 設定成真，請確保弦中使用的每個任務定義時都設定了 ignore_result=False。應用到任務子類和裝飾類都管用。

任務子類示例：

class MyTask(Task):
    ignore_result = False

裝飾類示例：

@app.task(ignore_result=False)
def another_task(project):
    do_something()

預設情況下，同步步驟是由一個週期性任務實現，它每秒輪訓組的完成狀態，當完成後呼叫回撥函式。

實現示例：

from celery import maybe_signature

@app.task(bind=True)
def unlock_chord(self, group, callback, interval=1, max_retries=None):
    if group.ready():
        return maybe_signature(callback).delay(group.join())
    raise self.retry(countdown=interval, max_retries=max_retries)

這被所有結果儲存後端使用，除了 Redis 和 Memcached：他們定義了一個計數器，每執行完一個任務加1，當計數器超過組中任務數時呼叫回撥函式。

Redis 和 Memcached 方式是更好的選擇，但是在其他儲存後端中不容易被實現（歡迎建議！）。

注意：
弦在Redis 2.2 以下版本不能正常工作；你需要升級到至少redis-server 2.2。

注意：
如果你使用了弦，並且使用Redis 儲存後端，同時又覆蓋了 Task.after_return() 方法，那麼你需要確保呼叫 super 方法，否則弦的回撥函式不會執行。

def after_return(self, *args, **kwargs):
    do_something()
    super(MyTask, self).after_return(*args, **kwargs)

Map & Starmap

map 和 starmap 是內建的任務，他們對序列中每個元素呼叫任務。

他們與任務組不同在於：
- 只有一個任務訊息被髮送
- 操作是按次序的

例如使用 map：

>>> from proj.tasks import add

>>> ~xsum.map([range(10), range(100)])
[45, 4950]

與下列任務等價：

@app.task
def temp():
    return [xsum(range(10)), xsum(range(100))]

使用 starmap:

>>> ~add.starmap(zip(range(10), range(10)))
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

與下面任務等價：

@app.task
def temp():
    return [add(i, i) for i in range(10)]

map 和 starmap 都是簽名物件，所以他們都可以用作其他簽名，或者組合到任務組等等，例如十秒鐘後呼叫 starmap:

>>> add.starmap(zip(range(10), range(10))).apply_async(countdown=10)

Chunks

分塊可以讓你將迭代的工作片段化，使得如果你有100萬個物件處理，那麼你可以建立10個任務，每個任務處理10萬個物件。

有人也許會擔心分塊任務會導致並行度的降低，但是對於一個繁忙的叢集來說一般是不會的，實踐中，因為你避免了訊息的耗費，結果可能還會提高效能。

你可以使用app.Task.chunks() 建立一個分塊簽名:

>>> add.chunks(zip(range(100), range(100)), 10)

在任務組中，為塊傳送任務訊息將在呼叫時在當前程序中進行：

>>> from proj.tasks import add

>>> res = add.chunks(zip(range(100), range(100)), 10)()
>>> res.get()
[[0, 2, 4, 6, 8, 10, 12, 14, 16, 18],
 [20, 22, 24, 26, 28, 30, 32, 34, 36, 38],
 [40, 42, 44, 46, 48, 50, 52, 54, 56, 58],
 [60, 62, 64, 66, 68, 70, 72, 74, 76, 78],
 [80, 82, 84, 86, 88, 90, 92, 94, 96, 98],
 [100, 102, 104, 106, 108, 110, 112, 114, 116, 118],
 [120, 122, 124, 126, 128, 130, 132, 134, 136, 138],
 [140, 142, 144, 146, 148, 150, 152, 154, 156, 158],
 [160, 162, 164, 166, 168, 170, 172, 174, 176, 178],
 [180, 182, 184, 186, 188, 190, 192, 194, 196, 198]]

當呼叫.apply_async方法，將建立一個專門的任務使得獨立子任務在一個工作單元中執行：

>>> add.chunks(zip(range(100), range(100)), 10).apply_async()

你還可以將分塊轉化成組：

>>> group = add.chunks(zip(range(100), range(100)), 10).group()

在任務組中，你可以通過遞增延遲調整每個任務的countdown時間：

>>> group.skew(start=1, stop=10)()

著意味著第一個任務countdown時間為1秒，第二個任務countdown時間為 2 秒，以此類推。