本文首發于公眾號：Hunter后端
原文鏈接：celery筆記三之task和task的呼叫
這一篇筆記介紹 task 和 task 的呼叫，
以下是本篇筆記目錄：
- 基礎的 task 定義方式
- 日志處理
- 任務重試
- 忽略任務運行結果
- task 的呼叫
1、基礎的 task 定義方式
前面兩篇筆記中介紹了最簡單的定義方式，使用 @app.task 作為裝飾器：
@app.task
def add(x, y):
return x + y
如果是在 Django 系統中使用 celery，需要定義一個延時任務或者周期定時任務，可以使用 @shared_task 來修飾
from celery import shared_task
@shared_task
def add(x, y):
return x + y
在 Django 系統中使用 celery 的方式會在接下來的幾篇筆記中介紹道，
多個裝飾器
如果是 celery 的任務和其他裝飾器一起聯用，記得將 celery 的裝飾器放在最后使用，也就是串列的最前面：
@app.task
@decorator1
@decorator2
def add(x, y):
return x + y
task名稱
每個 task 都有一個唯一的名稱用來標識這個 task，如果我們在定義的時候不指定，系統會為我們默認一個名稱，這些名稱會在 celery 的 worker 啟動的時候被系統掃描然后輸出一個串列展示，
還是上一篇筆記中我們定義的兩個 task，我們給其中一個指定 name：
#tasks1.py
from .celery import app
@app.task(name="tasks1.add")
def add(x, y):
return x + y
可以觀察在 celery 的 worker 啟動的時候，會有一個輸出：
[tasks]
. proj.tasks2.mul
. tasks1.add
可以看到這個地方，系統就會使用我們定義的 name 了，
2、日志處理
我們可以在啟動 worker 的時候指定日志的輸出，定義格式如下：
celery -A proj worker -l INFO --logfile=/Users/hunter/python/celery_log/celery.log
在 task 中的定義可以使用 celery 中方法：
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)
也可以直接使用 logging 模塊：
import logging
logger1 = logging.getLogger(__name__)
直接在 task 中輸出：
@app.task(name="tasks1.add")
def add(x, y):
logger.info("this is from logger")
return x + y
然后在 worker 啟動時指定的日志檔案就會有我們列印出的日志內容：
[2022-07-24 16:28:33,210: INFO/ForkPoolWorker-7] tasks1.add[4db4b0fc-c6ca-472a-8847-ae42e0a7959a]: this is from logger
[2022-07-24 16:28:33,224: INFO/ForkPoolWorker-7] Task tasks1.add[4db4b0fc-c6ca-472a-8847-ae42e0a7959a] succeeded in 0.016244667931459844s: 3
3、任務重試
對于一個 task，我們可以對其設定 retry 引數來指定其在任務執行失敗后會重試幾次，以及隔多長時間重試，
比如對于下面的 div() 函式，我們來輸入除數為 0 的情況查看重試的功能，
當然，這里我們是故意輸入引數錯誤，在實際的專案中可能會是其他的原因造成任務失敗，比如資料庫連接失敗等
任務重試的引數也都在 @app.task() 中定義：
# tasks1.py
@app.task(autoretry_for=(Exception, ), default_retry_delay=10, retry_kwargs={'max_retries': 5})
def div(x, y):
return x / y
在這里，autoretry_for 表示的是某種報錯情況下重試，我們定義的 Exception 表示任何錯誤都重試，
如果只是想在某種特定的 exception 情況下重試，將那種 exception 的值替換 Exception 即可，
default_retry_delay 表示重試間隔時長，默認值是 3 * 60s，即三分鐘，是以秒為單位，這里我們設定的是 10s，
retry_kwargs 是一個 dict，其中有一個 max_retries 引數，表示的是最大重試次數，我們定為 5
然后可以嘗試呼叫這個延時任務：
from proj.tasks1 import div
div.delay(1, 0)
然后可以看到在日志檔案會有如下輸出：
[2022-07-24 16:59:35,653: INFO/ForkPoolWorker-7] Task proj.tasks1.div[1f65c410-1b2a-4127-9d83-a84b1ad9dd2c] retry: Retry in 10s: ZeroDivisionError('division by zero',)
且每隔 10s 執行一次，一共執行 5 次，5次之后還是不成功則會報錯，
retry_backoff 和 retry_backoff_max
還有一個 retry_backoff 和 retry_backoff_max 引數，這兩個引數是用于這種情況：如果你的 task 依賴另一個 service 服務，比如會呼叫其他系統的 API，然后這兩個引數可以用于避免請求過多的占用服務，
retry_backoff 引數可以設定成一個 布爾型資料，為 True 的話，自動重試的時間間隔會成倍的增長
第一次重試是 1 s后
第二次是 2s 后
第三次是 4s 后
第四次是 8s 后
...
如果 retry_backoff 引數是一個數字，比如是 3，那么后續的間隔時間則是 3 的倍數增長
第一次重試 3s 后
第二次是 6s 后
第三次是 12s 后
第四次是 24s 后
retry_backoff_max 是重試的最大的間隔時間，比如重試次數設定的很大，retry_backoff 的間隔時間重復達到了這個值之后就不再增大了，
這個值默認是 600s，也就是 10分鐘，
我們看一下下面這個例子：
# tasks1.py
@app.task(autoretry_for=(Exception, ), retry_backoff=2, retry_backoff_max=40, retry_kwargs={'max_retries': 8})
def div(x, y):
return x / y
關于重試的機制，理論上應該是按照我們前面列出來的重試時間間隔進行重試，但是如果我們這樣直接運行 div.delay()，得出的間隔時間是不定的，是在 0 到 最大值之間得出的一個隨機值，
這樣產生的原因是因為還有一個 retry_jitter 引數，這個引數默認是 True，所以時間間隔會是一個隨機值，
如果需要任務延時的間隔值是按照 retry_backoff 和 retry_backoff_max 兩個設定值來運行，那么則需要將 retry_jitter 值設為 False，
# tasks1.py
@app.task(autoretry_for=(Exception, ), retry_backoff=2, retry_backoff_max=40, retry_jitter=False, retry_kwargs={'max_retries': 8})
def div(x, y):
return x / y
然后運行 div 的延時任務，就可以看到延時任務按照規律的間隔時間重試了，以下是日志：
[2022-07-24 19:00:38,588: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 2s: ZeroDivisionError('division by zero',)
[2022-07-24 19:00:40,662: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:00:40,664: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 4s: ZeroDivisionError('division by zero',)
[2022-07-24 19:00:44,744: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:00:44,746: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 8s: ZeroDivisionError('division by zero',)
[2022-07-24 19:00:52,870: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:00:52,872: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 16s: ZeroDivisionError('division by zero',)
[2022-07-24 19:01:09,338: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:01:09,340: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 32s: ZeroDivisionError('division by zero',)
[2022-07-24 19:01:41,843: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:01:41,845: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)
[2022-07-24 19:02:21,923: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:02:21,925: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)
[2022-07-24 19:03:02,001: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
[2022-07-24 19:03:02,003: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)
因為我們設定的重試間隔時間最大為 40s，所以這個地方延時間隔時間到了 40 之后，就不再往上繼續增長了，
4、忽略任務運行結果
有時候延時任務的結果我們并不想保存，但是我們配置了 result_backend 引數，這個時候我們有三種方式不保存運行結果，
1.ignore_result=True 不保存任務運行的結果
@app.task(ignore_result=True)
def add(x, y):
return x + y
2.app.conf 配置
也可以通過 app.conf 的配置來禁用結果的保存：
app.conf.update(
task_ignore_result=True
)
3.執行單個任務的時候禁用
from proj.tasks1 import add
add.apply_async((1, 2), ignore_result=True)
apply_async() 函式的作用相當于是帶引數的 delay()，或者 delay() 是簡化版的 apply_async()，這個我們下面會介紹，
5、task 的呼叫
前面簡單兩個簡單的呼叫方法，一個是 apply_async()，一個是 delay()，
簡單來說就是 delay() 是不帶引數執行的 apply_async()，
以下用 add() 函式為例介紹一下他們的用法：
delay()
純粹的延時任務，只能如下操作：
add.delay(1, 2)
apply_async()
帶引數的用法，add() 函式的引數用 () 包起來：
add.apply_async((1, 2))
也可以帶其他引數，比如上面介紹的不保存運行結果：
add.apply_async((1, 2), ignore_result=True)
這個函式還可以指定延時的時間：
countdown引數
現在開始 10s 后開始運行：
add.apply_async((1, 2), countdown=10)
eta引數
也可以用 eta 引數來指定 10s 后運行：
from datetime import datetime, timedelta
now = datetime.now()
add.apply_async((1, 2), eta=now + timedelta(seconds=10))
expires引數
這個是用來設定過期的引數：
add.apply_async((1, 2), countdown=60, expires=120)
上面的引數表示，距現在60秒后開始執行，兩分鐘后過期
如果想獲取更多后端相關文章，可掃碼關注閱讀：
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/555074.html
標籤：其他
上一篇：現代C++學習指南-型別系統
下一篇：返回列表