---
# System prepended metadata

title: Celery_Calling Tasks
tags: [celery 5.2, python, celery]

---

# Celery_Calling Tasks
###### tags: `celery` `celery 5.2` `python`
[官方連結_Calling Tasks](https://docs.celeryproject.org/en/stable/userguide/calling.html)

## Calling Tasks
### Basics
這份文件說明task instances與[canvas](https://docs.celeryproject.org/en/stable/userguide/canvas.html#guide-canvas)所使用的"Calling API"。

API定義了一組標準的執行選項以及三種方法：
1. `apply_async(args[, kwargs[, …]])`
    * 派送任務訊息
2. `delay(*args, **kwargs)`
    * 派送任務訊息的快捷方式，但不支援執行選項的設置
3. `calling (__call__)`
    * 應用物件(object)支援Calling API，代表任務並不會被worker執行，而是在當前程序執行(current process)。(沒有訊息被派送)

:::info
**Quick Cheat Sheet**

1. `T.delay(arg, kwarg=value)`
    * Star arguments shortcut to `.apply_async. (.delay(*args, **kwargs) calls .apply_async(args, kwargs))`.
2. `T.apply_async((arg,), {'kwarg': value})`
3. `T.apply_async(countdown=10)`
    * 現在開始10秒內執行
4. `T.apply_async(eta=now + timedelta(seconds=10))`
    * 現在開始10秒內執行，指定使用eta
5. `T.apply_async(countdown=60, expires=120)`
    * 現在開始1分鐘內執行，但2分鐘之後到期
6. `T.apply_async(expires=now + timedelta(days=2))`
    * 使用[datatime](https://docs.python.org/dev/library/datetime.html#datetime.datetime)設置，2天後到期
:::
#### Example
使用[delay()](https://docs.celeryproject.org/en/stable/reference/celery.app.task.html#celery.app.task.Task.delay)很方便，它看起來就像是使用一個再普通不過的函數：
```python
task.delay(arg1, arg2, kwarg1='x', kwarg2='y')
```

使用[apply_async()](https://docs.celeryproject.org/en/stable/reference/celery.app.task.html#celery.app.task.Task.apply_async)來代替的話可以這樣寫：
```python
task.apply_async(args=[arg1, arg2], kwargs={'kwarg1': 'x', 'kwarg2': 'y'})
```

因此，使用`delay`是很方便沒錯，不過如果你要設置額外的執行選項的話，就必需使用`apply_async`。

這文件的其餘部份會直接進入任務執行選項的細部設置。所有的範例均使用一個名為`add`的任務，函數會回傳兩個參數的加總：
```python
@app.task
def add(x, y):
    return x + y
```

:::info
**There’s another way…**

後面你在讀Canvas的時候會有更深的體驗，不過signature(簽章)是用於傳遞任務調用簽章的對象(例如，透過網路派送)，而且它們也支援Calling API：
```python
task.s(arg1, arg2, kwarg1='x', kwargs2='y').apply_async()
```
:::

### Linking (callbacks/errbacks)
Celery支援將任務連結在一起，一個跟著一個這樣。這種callback task(回呼任務)會伴隨著父任務的結果做為局部參數(partial argument)，如下說明：
```python
add.apply_async((2, 2), link=add.s(16))
```

第一個任務(父任務)的回傳值4將會被送去新任務，並將16加到第一個任務的回傳值4，最後得到(2+2)+16=20。

如果任務拋出異常(errbacks)，你也可以觸發回呼函數(callback)。不過worker並不會真正的去呼叫這個errback來做為任務，而是會直接呼叫errback fucntion，這樣就可以把原始的請求(request)、例外(exception)與回溯(trackback)物件傳遞給它。

下面給出`error callback`的範例：
```python
@app.task
def error_handler(request, exc, traceback):
    print('Task {0} raised exception: {1!r}\n{2!r}'.format(
          request.id, exc, traceback))
```

可以使用`link_error`這個執行選項將之加入任務中：
```python
add.apply_async((2, 2), link_error=error_handler.s())
```

此外，`link`與`link_error`都可以使用`list`來表示：
```python
add.apply_async((2, 2), link=[add.s(16), other_task.s()])
```

以`list`來表示的時候，callbacks與errbacks都會被按序執行，而且所有的回呼函數都會使用父任務的回傳值做為局部參數。

:::info
**What’s s?**

上面用的`add.s`就是一種簽章(signature)。如果你不知道這是什麼的話，後續可以讀讀[canvas guide](https://docs.celeryproject.org/en/stable/userguide/canvas.html#guide-canvas)。在那邊你還可以學到一些關於[chain](https://docs.celeryproject.org/en/stable/reference/celery.html#celery.chain)：一種將任務鏈結起來的簡單方法。
:::

### On message
Celery支援使用`on_message` callback來補捉所有狀態變化。

舉例來說，如果你想要一個長時間執行的任務發送它的任務進度的話，可以這麼做：
```python
@app.task(bind=True)
def hello(self, a, b):
    time.sleep(1)
    self.update_state(state="PROGRESS", meta={'progress': 50})
    time.sleep(1)
    self.update_state(state="PROGRESS", meta={'progress': 90})
    time.sleep(1)
    return 'hello world: %i' % (a+b)
```

```python
def on_raw_message(body):
    print(body)

a, b = 1, 1
r = hello.apply_async(args=(a, b))
print(r.get(on_message=on_raw_message, propagate=False))
```

會得到下面的結果：
```jsonld
{'task_id': '5660d3a3-92b8-40df-8ccc-33a5d1d680d7',
 'result': {'progress': 50},
 'children': [],
 'status': 'PROGRESS',
 'traceback': None}
{'task_id': '5660d3a3-92b8-40df-8ccc-33a5d1d680d7',
 'result': {'progress': 90},
 'children': [],
 'status': 'PROGRESS',
 'traceback': None}
{'task_id': '5660d3a3-92b8-40df-8ccc-33a5d1d680d7',
 'result': 'hello world: 10',
 'children': [],
 'status': 'SUCCESS',
 'traceback': None}
hello world: 10
```
### ETA and Countdown
ETA(estimated time of arrival)<sub>(預計到達時間)</sub>允許你可以設置指定的日期與時間，這是任務執行的最早時間。`countdown`是一種以秒為單位設置ETA的快捷方式。
```python
# 下面範例設置3秒後執行
>>> result = add.apply_async((2, 2), countdown=3)
>>> result.get()    # this takes at least 3 seconds to return
20
```

任務保證在指定日期與時間之後的某個時間被執行，但不一定在那個確切的時間。當然有很多的原因會造成你無法在時間內執行，像是隊列中還有許多項目正排隊等待，或網路負載延遲。為了確認你的任務能夠被及時處理，你應該監視隊最是否擁塞。使用Munin或類似的工具，接收警示，再採取適合的調整來減輕工作量。參閱[Mnium](https://docs.celeryproject.org/en/stable/userguide/monitoring.html#monitoring-munin)

雖然`countdown`是一個整數，不過`eta`必需要是[datatime](https://docs.python.org/dev/library/datetime.html#datetime.datetime)，指定確切的日期與時間(包含毫秒精度以及時區)：
```python
>>> from datetime import datetime, timedelta

>>> tomorrow = datetime.utcnow() + timedelta(days=1)
>>> add.apply_async((2, 2), eta=tomorrow)
```

:::warning
Warning:

當你使用RabbitMQ做為message broker的時候，你指定一個`countdown`超過15分鐘，你可能會遇到worker拋出一個[PreconditionFailed](https://docs.celeryproject.org/projects/amqp/en/latest/reference/amqp.exceptions.html#amqp.exceptions.PreconditionFailed)的異常：
```
amqp.exceptions.PreconditionFailed: (0, 0): (406) PRECONDITION_FAILED - consumer ack timed out on channel
```

RabbitMQ在3.8.15之後，其`consumer_timeout`預設為15分鐘。而在3.8.17之後預設為30分鐘。如果consumer沒有確認其交付(delivery)超過timeout value，那就會關閉它的channel，然後拋出一個異常`PRECONDITION_FAILED`。更多細節請參閱[Delivery Acknowledgement Tiemout](https://www.rabbitmq.com/consumers.html#acknowledgement-timeout)。

想要解決這個問題，就是要在RabbitMQ的配置文件`rabbitmq.conf`中指定參數`consumer_timeout`的值要大於等於你的countdown value(倒數計時的值)。舉例來說，你可以指定一個很大大大大的值，`consumer_timeout = 31622400000`(毫秒計算來看，這是1年的值)來避免後續有出問題。
:::
### Expiration
參數`exipres`定義了一個可選的到期時間，可以是任務派送之後幾秒，或是使用[`datatime`](https://docs.python.org/dev/library/datetime.html#datetime.datetime)指定日期與時間
```python
>>> # Task expires after one minute from now.
>>> add.apply_async((10, 10), expires=60)

>>> # Also supports datetime
>>> from datetime import datetime, timedelta
>>> add.apply_async((10, 10), kwargs,
...                 expires=datetime.now() + timedelta(days=1)
```
當一個worker接收到一個過期任務的時候，它會將該任務標記為[REVOKED](https://docs.celeryproject.org/en/stable/userguide/tasks.html#std-state-REVOKED)[TaskRevokedError](https://docs.celeryproject.org/en/stable/reference/celery.exceptions.html#celery.exceptions.TaskRevokedError)
### Message Sending Retry
在連接失敗的時候Celery會自動重新派送訊息，而且你是可以設置這種重新測試的行為，像是多久重新測試，最大重新測試次數，或是通通設置為disable。

要disabled重新測試，你可以設置`retry=False`：
```python
add.apply_async((2, 2), retry=False)
```
:::info
**Related Settings**

* [task_publish_retry](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_publish_retry)
* [task_publish_retry_policy](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_publish_retry_policy)
:::

#### Retry Policy
重新測試的策略是一種著控制重新測試派送任務行為的映射，可以包含下面幾個關鍵：
* max_retries
    * 放棄之前的最大重新測試次數，在這情況下會拋出重新測試異常
    * 設置為`None`則代表會一直重新測試
    * 預設值為3
* interval_start
    * 定義重新測試之間的等待秒數(浮點數或整數)，預設為0(第一次重測是瞬間的)
* interval_step
    * 在每次連續重新測試的時候，這個數值會被加到重新測試延值(點數或整數)
    * 預設值為0.2
* interval_max
    * 重新測試之間的等待最大秒數(浮點數或整數)
    * 預設值為0.2

舉例來說，預設的重新測試策略為：
```python
add.apply_async((2, 2), retry=True, retry_policy={
    'max_retries': 3,
    'interval_start': 0,
    'interval_step': 0.2,
    'interval_max': 0.2,
})
```
重新測試所花的最長時間為0.4秒。預設情況下相對較短，因為如果broker連接中斷，連接失敗可能會導致堆疊效應。舉例來說，許多web server程序等待重新測試，阻止了其它傳入的請求。

### Connection Error Handling
當你派送一個任務，並且該訊息傳輸連接遺失了，或連接無法被啟動，會拋出`OperationalError`異常：
```python
>>> from proj.tasks import add
>>> add.delay(2, 2)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "celery/app/task.py", line 388, in delay
        return self.apply_async(args, kwargs)
  File "celery/app/task.py", line 503, in apply_async
    **options
  File "celery/app/base.py", line 662, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "celery/backends/rpc.py", line 275, in on_task_call
    maybe_declare(self.binding(producer.channel), retry=True)
  File "/opt/celery/kombu/kombu/messaging.py", line 204, in _get_channel
    channel = self._channel = channel()
  File "/opt/celery/py-amqp/amqp/connection.py", line 272, in connect
    self.transport.connect()
  File "/opt/celery/py-amqp/amqp/transport.py", line 100, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/opt/celery/py-amqp/amqp/transport.py", line 141, in _connect
    self.sock.connect(sa)
  kombu.exceptions.OperationalError: [Errno 61] Connection refused
```

假如你啟用重新測試，那就只會在重新測試次數耗盡或立即禁用的時候才會發生。

你也可以這麼處理錯誤：
```python
>>> from celery.utils.log import get_logger
>>> logger = get_logger(__name__)

>>> try:
...     add.delay(2, 2)
... except add.OperationalError as exc:
...     logger.exception('Sending task raised: %r', exc)
```
### Serializers
資料在clients與workers之間的傳輸是需要被序列化的，因此所有在Celery的訊息都有著`content_type`的表頭(header)，它描述著序列化的編碼方式。

預設序列化方式是`JSON`，但是你可以使用[task_serializer](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_serializer)來改變，或是每次任務單獨的設置，甚至可以細到每個訊息的設置。

內建支援json、pickle、YAML、以及msgpack，你可以透過Kombu來自定義序列化方式。

:::info
See also:
[Message Serialization](https://docs.celeryproject.org/projects/kombu/en/master/userguide/serialization.html#guide-serialization)
:::

:::info
**Security**

pickle這個模組允許可以執行任意的函數，請參閱[security guide](https://docs.celeryproject.org/en/stable/userguide/security.html#guide-security)。

Celery自帶一個特殊的序列化方式，可以針對你所派送的訊息做加密簽章的處理。
:::

每個序列化方式都有它的優缺點：
* json
    * JSON支援很多程式語言，是Python的標準庫，編碼相當快速，如[simplejson](https://pypi.org/project/simplejson/)。
    * JSON的主要缺點就是它限制了你使用下列資料類型(Decimals與dates很明顯的是沒有的)
        * strings
        * Unicode
        * floats
        * Boolean
        * dictionaries
        * lists
    * 二進位資料(Binary data)會使用Base64編碼，與所支援的原生二進位類型的編碼格式相比，傳輸數據量增加了34%
    * 如果你的資料符合上述限制，而且你有跨語言支援需求，JSON的預設設置也許是你的最好選擇
    * 更多信息請參閱[http://json.org](https://www.json.org/json-en.html)
    :::info
    **Note:**
    
    從Python[官方文件](https://docs.python.org/3.6/library/json.html))資料來看，其JSON的鍵值結終為字串(str)類型。當你把dictionary轉為JSON的時候，所有的`keys`(鍵值)也都會強制轉為字串。所以你如果轉來轉去的把dict轉JSON，再從JSON轉回dict的話，那它可能就不是原本的dict。簡單來說就是你的dict如果有非字串類型的鍵值，那轉回去就不會是原始的dict。
    :::
* pickle
    * 如果你沒有支援Python以外的語言需求，使用pickle將擁有所有Python內建資料類型的支援(class instance除外)，派送binary files時更小的訊息，以及較JSON快一些的處理速度
    * 更多信息請參閱[pickle](https://docs.python.org/dev/library/pickle.html#module-pickle)
* yaml
    * YAML有著許多與json相同的特性，不過YAML支援更多的資料類型(包含日期、遞歸引用..等)
    * Python的標準庫中，YAML執行效率較JSON慢一些些
    * 如果你需要更豐富的資料類型以及保持跨語言的兼容性，那YAML會較上述更適合。
    * 更多信息請參閱[http://yaml.org](http://yaml.org/)
* msgpack
    * msgpack是一個二進位序列化格式，功能面類似於JSON。還有新，剛出的，所以應該將它視為實驗性的會比較好
    * 更多信息請參閱[http://msgpack.org/](http://msgpack.org/)


使用的編碼方式可以做為訊息的表頭(message header)，這樣worker就知道對任務做怎麼樣的反序列化。如果你使用自定義的序列化方式，這序列化方式必須可以提供worker使用。

下面順序用於決定派送任務的時候所使用的序列化方式：
1. The `serializer execution` option
2. The [`Task.serializer`](https://docs.celeryproject.org/en/stable/reference/celery.app.task.html#celery.app.task.Task.serializer) attribute
3. The [`task_serializer`](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_serializer) setting

下面說明為單個任務設置自定義序列化函式：
```python
>>> add.apply_async((10, 10), serializer='json')
```

### Compression
Celery可以使用下面內建方案來壓縮訊息：
* brotli
    * brotli針對網頁進行了最佳化，尤其是小型文字類型的文件( text documents)。對於提供靜態內容，像是字型與html頁面最有效
    * 使用它，安裝Celery的時候記得如下安裝：
    ```shell
    $ pip install celery[brotli]
    ```
* bzip2
    * bzip2可以建立比gzip更小的檔案，但壓縮與解壓縮的速度明顯比gzip還要慢
    * 要使用它，請確保你的Python支援執行文件是使用bzip2編譯的
    * 如果你得到下面異常[`ImportError`](https://docs.python.org/dev/library/exceptions.html#ImportError)，這意味著你應該重新編譯你的Python支援的bzip2版本
    ```python
    >>> import bz2
    Traceback (most recent call last):
      File "<stdin>", line 1, in <module>
    ImportError: No module named 'bz2'
    ```
* gzip
    * gzip適用於記憶體空間較小的系統，適合用於記憶體有限的系統。它通常被用來產生副檔名為`.tar.gz`的檔案
    * 要使用它，請確保你的Python支援執行文件是使用gzip編譯的
    * 如果你得到下面異常[`ImportError`](https://docs.python.org/dev/library/exceptions.html#ImportError)，這意味著你應該重新編譯你的Python支援的bzip2版本
    ```python
    >>> import gzip
    Traceback (most recent call last):
      File "<stdin>", line 1, in <module>
    ImportError: No module named 'gzip'
    ```
* lzma
    * lzma提供了良好的壓縮比並且有更快的壓縮、解壓縮執行效率，只是這也造成記憶體用量更高
    * 要使用它，請確保你的Python支援執行文件是使用lzma編譯的，而且你的Python是3.3或更新的版本
    * 如果你得到下面異常[`ImportError`](https://docs.python.org/dev/library/exceptions.html#ImportError)，這意味著你應該重新編譯你的Python支援的lzma版本
    ```python
    >>> import lzma
    Traceback (most recent call last):
      File "<stdin>", line 1, in <module>
    ImportError: No module named 'lzma'
    ```
    * 你也可以如下安裝：
    ```shell
    $ pip install celery[lzma]
    ```
* zlib
    * zlib是庫形式的Deflate演算法的抽象，它包括對gzip文件格式和API中的輕量級串流格式的支援。它是許多軟體系統的關鍵組成，像Linux kernel、Git VCS..
    * 要使用它，請確保你的Python支援執行文件是使用zlib編譯的
    * * 如果你得到下面異常[`ImportError`](https://docs.python.org/dev/library/exceptions.html#ImportError)，這意味著你應該重新編譯你的Python支援的lzma版本
    ```python
    >>> import zlib
    Traceback (most recent call last):
      File "<stdin>", line 1, in <module>
    ImportError: No module named 'zlib'
    ```
* zstd
    * zstd以zlib-level的即時壓縮以及更好的壓縮比為目標。由Huff0與FSE兩個libary所提供的非常快速的entropy stage的支持
    * 使用它，安裝Celery的時候記得如下安裝：
    ```shell
    $ pip install celery[zstd]
    ```

你也可以自定義壓縮方案，並且在[kombu.compression.register](https://docs.celeryproject.org/projects/kombu/en/master/reference/kombu.compression.html#kombu.compression.register)中註冊它。

下面順序用於派送任務的時候決定壓縮的方式：
1. The compression execution option
2. The `Task.compression` attribute
3. The [task_compression](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_compression) attribute

下面範例說明在呼叫任務的時候指定壓縮方式：
```python
>>> add.apply_async((2, 2), compression='zlib')
```
### Connections
你可以通過建立publisher來手動處理連接的部份：
```python
results = []
with add.app.pool.acquire(block=True) as connection:
    with add.get_publisher(connection) as publisher:
        try:
            for args in numbers:
                res = add.apply_async((2, 2), publisher=publisher)
                results.append(res)
print([res.get() for res in results])
```

雖然說這個特殊的例子用`group`的話可以有更好的表示方式：
```python
>>> from celery import group

>>> numbers = [(2, 2), (4, 4), (8, 8), (16, 16)]
>>> res = group(add.s(i, j) for i, j in numbers).apply_async()

>>> res.get()
[4, 8, 16, 32]
```

:::info
**Automatic Pool Support**

從版本2.3之後支援`automatic connection pools`，因此你不需要手動處理連接和發行者以重用連接。

`connection pools`從版本2.5之後預設狀態為啟用。

見[broker_pool_limit](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-broker_pool_limit)說明
:::
### Routing options
Celery可以路由任務到不同的隊列。

簡單的路由(name<->name)可以使用選項`queue`來完成：
```python
add.apply_async(queue='priority.high')
```
然後，你可以利用參數`-Q`來指派worker讓它加入相對應設置的隊列名稱(`celery, priority.high`)：
```shell
$ celery -A proj worker -l INFO -Q celery,priority.high
```

:::info
**See also:**

不建議在程式碼程以硬編碼的方式來加入隊列，最好的方法是利用配置路由來設置。([task_routes](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_routes))

更多路由可以參閱[Routing Tasks](https://docs.celeryproject.org/en/stable/userguide/routing.html#guide-routing)
:::
### Results options
你可以利用設置[task_ignore_result](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-task_ignore_result)或選項(`ignore_result`)來決定是否保存任務執行結果

```python
>>> result = add.apply_async((1, 2), ignore_result=True)
>>> result.get()
None

>>> # Do not ignore result (default)
...
>>> result = add.apply_async((1, 2), ignore_result=False)
>>> result.get()
3
```

如果你想要額外保存任務的元資料(metadata)在result backend，那就記得將[result_extended](https://docs.celeryproject.org/en/stable/userguide/configuration.html#std-setting-result_extended)設置為True。

:::info
**See also:**

更多任務說明請參閱[Tasks](https://docs.celeryproject.org/en/stable/userguide/tasks.html#guide-tasks)
:::
#### Advanced Options
這些選項是提供給應用AMQP完整路由功能的進階用戶，有興趣的話可以閱讀[routind guide](https://docs.celeryproject.org/en/stable/userguide/routing.html#guide-routing)
* excnahge
    * 要將訊息派送到的那個交換機的名稱(或[kombu.entity.Exchange](https://docs.celeryproject.org/projects/kombu/en/master/reference/kombu.html#kombu.Exchange))
* routing_key
    * routing key
        * 用來確定的routing key
* priority
    * 0-255的數值，255代表最高優先(RabbitMQ)
    * Redis的話則0為最高優化

## History
20190725_依據4.3版本說明調整
20211210_依據5.2版本說明調整