python-background-jobs

Python 后台作业与任务队列

将长时间运行或不可靠的工作与请求/响应周期解耦。立即向用户返回，同时由后台 Worker 异步处理繁重的工作。

何时使用本技能

• 处理耗时超过几秒的任务
• 发送电子邮件、通知或 Webhook
• 生成报告或导出数据
• 处理上传内容或媒体转换
• 与不可靠的外部服务集成
• 构建事件驱动架构

核心概念

1. 任务队列模式

API 接受请求，将作业加入队列，并立即返回作业 ID。Worker 异步处理作业。

2. 幂等性

任务在失败时可能会被重试。设计时应确保可安全地重复执行。

3. 作业状态机

作业会经历状态转换：pending → running → succeeded/failed。

4. 至少一次投递

大多数队列都保证至少一次投递。你的代码必须处理重复项。

快速开始

本技能使用 Celery 作为示例，它是一个被广泛采用的任务队列。RQ、Dramatiq 和云原生解决方案（AWS SQS、GCP Tasks）等替代方案同样是有效选择。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

@app.task
def send_email(to: str, subject: str, body: str) -> None:
    # This runs in a background worker
    email_client.send(to, subject, body)

# In your API handler
send_email.delay("[email protected]", "Welcome!", "Thanks for signing up")

基本模式

模式 1：立即返回作业 ID

对于超过几秒的操作，返回作业 ID 并异步处理。

from uuid import uuid4
from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

@dataclass
class Job:
    id: str
    status: JobStatus
    created_at: datetime
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: dict | None = None
    error: str | None = None

# API endpoint
async def start_export(request: ExportRequest) -> JobResponse:
    """Start export job and return job ID."""
    job_id = str(uuid4())

    # Persist job record
    await jobs_repo.create(Job(
        id=job_id,
        status=JobStatus.PENDING,
        created_at=datetime.utcnow(),
    ))

    # Enqueue task for background processing
    await task_queue.enqueue(
        "export_data",
        job_id=job_id,
        params=request.model_dump(),
    )

    # Return immediately with job ID
    return JobResponse(
        job_id=job_id,
        status="pending",
        poll_url=f"/jobs/{job_id}",
    )

模式 2：Celery 任务配置

使用适当的重试和超时设置配置 Celery 任务。

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

# Global configuration
app.conf.update(
    task_time_limit=3600,          # Hard limit: 1 hour
    task_soft_time_limit=3000,      # Soft limit: 50 minutes
    task_acks_late=True,            # Acknowledge after completion
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,   # Don't prefetch too many tasks
)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
)
def process_payment(self, payment_id: str) -> dict:
    """Process payment with automatic retry on transient errors."""
    try:
        result = payment_gateway.charge(payment_id)
        return {"status": "success", "transaction_id": result.id}
    except PaymentDeclinedError as e:
        # Don't retry permanent failures
        return {"status": "declined", "reason": str(e)}
    except TransientError as e:
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

模式 3：让任务具备幂等性

Worker 可能在崩溃或超时时重试。设计时应确保可安全地重复执行。

@app.task(bind=True)
def process_order(self, order_id: str) -> None:
    """Process order idempotently."""
    order = orders_repo.get(order_id)

    # Already processed? Return early
    if order.status == OrderStatus.COMPLETED:
        logger.info("Order already processed", order_id=order_id)
        return

    # Already in progress? Check if we should continue
    if order.status == OrderStatus.PROCESSING:
        # Use idempotency key to avoid double-charging
        pass

    # Process with idempotency key
    result = payment_provider.charge(
        amount=order.total,
        idempotency_key=f"order-{order_id}",  # Critical!
    )

    orders_repo.update(order_id, status=OrderStatus.COMPLETED)

幂等性策略：

1. 写入前检查：在操作前验证状态
2. 幂等键：对外部服务使用唯一令牌
3. Upsert 模式：INSERT ... ON CONFLICT UPDATE
4. 去重窗口：跟踪已处理的 ID N 小时

模式 4：作业状态管理

持久化作业状态转换，以便于可见性和调试。

class JobRepository:
    """Repository for managing job state."""

    async def create(self, job: Job) -> Job:
        """Create new job record."""
        await self._db.execute(
            """INSERT INTO jobs (id, status, created_at)
               VALUES ($1, $2, $3)""",
            job.id, job.status.value, job.created_at,
        )
        return job

    async def update_status(
        self,
        job_id: str,
        status: JobStatus,
        **fields,
    ) -> None:
        """Update job status with timestamp."""
        updates = {"status": status.value, **fields}

        if status == JobStatus.RUNNING:
            updates["started_at"] = datetime.utcnow()
        elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED):
            updates["completed_at"] = datetime.utcnow()

        await self._db.execute(
            "UPDATE jobs SET status = $1, ... WHERE id = $2",
            updates, job_id,
        )

        logger.info(
            "Job status updated",
            job_id=job_id,
            status=status.value,
        )

详细示例与模式

详细章节（以 ## Advanced Patterns 开头）位于 references/details.md。当上面的导航摘要不足时，请阅读该文件。

最佳实践摘要

1. 立即返回 - 不要为长时间操作阻塞请求
2. 持久化作业状态 - 支持状态轮询和调试
3. 让任务具备幂等性 - 可在任何失败时安全重试
4. 使用幂等键 - 用于外部服务调用
5. 设置超时 - 包括软限制和硬限制
6. 实现 DLQ - 捕获永久失败的任务
7. 记录状态转换 - 跟踪作业状态变化
8. 适当重试 - 对瞬时错误使用指数退避
9. 不要重试永久性失败 - 验证错误、无效凭据
10. 监控队列深度 - 在积压增长时告警