重複程式碼的集合中又添新成員：追蹤 Celery 任務(wù)的執(zhí)行時(shí)間。

每個(gè) Celery 任務(wù)其實(shí)有兩個(gè)不同的「執(zhí)行」時(shí)間：

• 實(shí)際執(zhí)行時(shí)間: 程式碼運(yùn)作的時(shí)間。
• 「完成時(shí)間」: 包含在佇列中等待可用工作進(jìn)程的時(shí)間。

兩者都很重要，因?yàn)槲覀兊淖罱K目標(biāo)是了解任務(wù)何時(shí)完成。

觸發(fā)任務(wù)後，我們需要知道任務(wù)何時(shí)完成以及何時(shí)可以預(yù)期結(jié)果。這就像項(xiàng)目估算一樣。管理者真正想知道的是專案何時(shí)完成，而不是它在一週內(nèi)就能完成，但沒(méi)有人有空在接下來(lái)的六個(gè)月內(nèi)去做。

使用 Celery 訊號(hào)

我們可以使用 Celery 訊號(hào)來(lái)計(jì)時(shí)任務(wù)。

提示 1: Celery 訊號(hào)的所有參數(shù)都是關(guān)鍵字參數(shù)。這意味著我們可以只列出我們感興趣的關(guān)鍵字參數(shù)，並將其餘參數(shù)打包到 **kwargs 中。這是個(gè)非常棒的設(shè)計(jì)！所有訊號(hào)都應(yīng)該採(cǎi)用這種方式！

提示 2: 我們可以將執(zhí)行開(kāi)始和結(jié)束時(shí)間儲(chǔ)存在任務(wù)物件的「headers」屬性中。

任務(wù)入隊(duì)

當(dāng) Celery 任務(wù)進(jìn)入佇列時(shí)，記錄目前時(shí)間：

from celery import signals
from dateutil.parser import isoparse
from datetime import datetime, timezone

@signals.before_task_publish.connect
def before_task_publish(*, headers: dict, **kwargs):
    raw_eta = headers.get("eta")
    publish_time = isoparse(raw_eta) if raw_eta else datetime.now(tz=timezone.utc)
    headers["__publish_time"] = publish_time.isoformat()

任務(wù)開(kāi)始執(zhí)行

當(dāng)工作流程接收到任務(wù)時(shí)，記錄當(dāng)前時(shí)間：

from celery import signals
from datetime import datetime, timezone

@signals.task_prerun.connect
def task_prerun(*, task: Task, **kwargs):
    setattr(task.request, "__prerun_time", datetime.now(tz=timezone.utc).isoformat())

任務(wù)執(zhí)行結(jié)束

任務(wù)完成後，計(jì)算執(zhí)行時(shí)間並將其儲(chǔ)存到某個(gè)地方，例如 StatsD 或其他監(jiān)控工具。

StatsD 是用於監(jiān)控應(yīng)用程式和偵測(cè)任何軟體以提供自訂指標(biāo)的業(yè)界標(biāo)準(zhǔn)技術(shù)堆疊。

• Netdata: StatsD 簡(jiǎn)介 [1]

from celery import signals, Task
from dateutil.parser import isoparse
from datetime import datetime, timezone, timedelta

def to_milliseconds(td: timedelta) -> int:
    return int(td.total_seconds() * 1000)

@signals.task_postrun.connect
def task_postrun(*, task: Task, **kwargs):
    now = datetime.now(tz=timezone.utc)
    publish_time = isoparse(getattr(task.request, "__publish_time", ""))
    prerun_time = isoparse(getattr(task.request, "__prerun_time", ""))

    exec_time = now - prerun_time if prerun_time else timedelta(0)
    waiting_time = prerun_time - publish_time if publish_time and prerun_time else timedelta(0)
    waiting_and_exec_time = now - publish_time if publish_time else timedelta(0)

    stats = {
        "exec_time_ms": to_milliseconds(exec_time),
        "waiting_time_ms": to_milliseconds(waiting_time),
        "waiting_and_exec_time_ms": to_milliseconds(waiting_and_exec_time),
    }
    # TODO: 將統(tǒng)計(jì)數(shù)據(jù)發(fā)送到 StatsD 或其他監(jiān)控工具
    statsd.timing(f"celery.task.exec_time_ms", stats["exec_time_ms"], tags=[f"task:{task.name}"])
    # ... 發(fā)送其他統(tǒng)計(jì)數(shù)據(jù) ...

額外功能：設(shè)定執(zhí)行時(shí)間過(guò)長(zhǎng)警告

可以在上述函數(shù)中加入硬編碼閾值：

if exec_time > timedelta(hours=1):
    logger.error(f"任務(wù) {task.name} 執(zhí)行時(shí)間過(guò)長(zhǎng): {exec_time}。請(qǐng)檢查！")

或者，可以根據(jù)任務(wù)定義設(shè)定多層閾值或閾值，或任何可以在程式碼中表達(dá)的內(nèi)容。

以上是如何測(cè)量 Celery 任務(wù)的執(zhí)行時(shí)間？的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！