?? ?? ??? Celery ?? ?? ?? ????? ? ??? ???????.

? Celery ???? ??? ? ?? "??" ??? ????.

• ?? ?? ??: ??? ???? ? ??? ?????.
• "?? ??": ?? ??? ??? ????? ?? ????? ???? ? ??? ??? ?????.

??? ???? ??? ??? ?? ????? ?? ??? ??? ? ?? ?? ?????.

??? ???? ? ??? ?? ???? ?? ??? ??? ? ??? ??? ???. ??? ???? ??? ????. ????? ??? ?? ???? ?? ????? ?? ??? ???, ??? ?? ??? ???? ??? ??? 6?? ?? ??? ? ??? ??? ?? ??????.

??? ?? ??

Celery ??? ???? ?? ??? ??? ? ????.

? 1: Celery ??? ?? ????? ??? ???????. ?, ?? ?? ??? ??? ???? ???? **kwargs? ?? ? ????. ??? ??? ??????! ?? ??? ??? ?? ???!

? 2: ?? ??? "headers" ??? ?? ?? ? ?? ??? ??? ? ????.

?? ??

Celery ??? ???? ???? ?? ??? ?????.

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()

?? ?? ??

??? ????? ??? ???? ?? ??? ?????.

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())

?? ??? ???????

??? ???? ?? ??? ???? StatsD? ?? ???? ?? ?? ?? ?????.

StatsD? ??????? ?????? ?????? ???? ??? ????? ???? ?? ?? ?? ?? ?????.

• ????: StatsD ?? [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)計數(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)計數(shù)據(jù) ...

?? ??: ? ?? ?? ?? ??

? ??? ????? ???? ??? ? ????.

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

?? ?? ?? ?? ??? ??? ? ?? ?? ??? ???? ?? ?? ??? ?? ???? ??? ? ????.

? ??? Celery ??? ?? ??? ??? ??????? ?? ?????. ??? ??? PHP ??? ????? ?? ?? ??? ?????!