《Python编程实战》8.2 任务队列与重试

from celery import Celery

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

app.conf.update(
    task_acks_late=True,                  # 任务执行完才 ack，崩溃可重新投递
    task_reject_on_worker_lost=True,      # worker 被杀时拒绝任务，而非静默丢失
    task_default_retry_delay=10,
    broker_transport_options={"visibility_timeout": 3600},
)

@app.task(bind=True, max_retries=3,
          autoretry_for=(ConnectionError,),
          retry_backoff=True, retry_jitter=True)
def send_email(self, to):
    ...

from arq import cron
from arq.connections import RedisSettings

async def send_email(ctx, to):
    ...

async def startup(ctx):
    ctx["pool"] = ...          # 建连接池

class WorkerSettings:
    functions = [send_email]
    cron_jobs = [cron(send_email, hour=9, minute=0)]
    redis_settings = RedisSettings(host="localhost", port=6379)
    max_tries = 3

import logging
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log,
)

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("demo")

class Transient(Exception):
    pass

attempts = 0

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=0.05, min=0.05, max=0.4),
    retry=retry_if_exception_type(Transient),
    before_sleep=before_sleep_log(log, logging.WARNING),
    reraise=True,
)
def flaky():
    global attempts
    attempts += 1
    if attempts < 3:
        raise Transient(f"第 {attempts} 次失败")
    return f"第 {attempts} 次成功"

print("结果:", flaky())
print("总尝试次数:", attempts)

Retrying __main__.flaky in 0.05 seconds as it raised Transient: 第 1 次失败.
Retrying __main__.flaky in 0.1 seconds as it raised Transient: 第 2 次失败.
结果: 第 3 次成功
总尝试次数: 3

import asyncio
import json
import uuid
from dataclasses import dataclass, asdict
import fakeredis

QUEUE = "app:q:default"
PROCESSING = "app:q:default:processing"
DLQ = "app:q:default:dead"
MAX_ATTEMPTS = 3

@dataclass
class Task:
    id: str
    name: str
    payload: dict
    attempts: int = 0

    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)

    @classmethod
    def from_json(cls, s: str) -> "Task":
        return cls(**json.loads(s))

class Broker:
    def __init__(self, redis):
        self.redis = redis

    def submit(self, name, payload):
        task = Task(id=uuid.uuid4().hex[:8], name=name, payload=payload)
        self.redis.lpush(QUEUE, task.to_json())
        return task.id

    def fetch(self):
        """RPOPLPUSH：从待处理队列取出并放进 processing，保证不丢"""
        raw = self.redis.rpoplpush(QUEUE, PROCESSING)
        return Task.from_json(raw) if raw else None

    def ack(self, task):
        self.redis.lrem(PROCESSING, 1, task.to_json())

    def retry(self, task):
        self.redis.lrem(PROCESSING, 1, task.to_json())
        task.attempts += 1
        self.redis.lpush(QUEUE, task.to_json())

    def dead_letter(self, task, reason):
        self.redis.lrem(PROCESSING, 1, task.to_json())
        self.redis.lpush(DLQ, json.dumps(
            {"task": json.loads(task.to_json()), "reason": reason}, ensure_ascii=False))

HANDLERS = {}

def handler(name):
    def deco(fn):
        HANDLERS[name] = fn
        return fn
    return deco

@handler("send_email")
def send_email(payload):
    if payload.get("to") == "bad@example.com":
        raise ValueError("SMTP 拒收")
    return f"sent to {payload['to']}"

async def worker(broker, stop: asyncio.Event):
    while not stop.is_set():
        task = broker.fetch()
        if task is None:
            await asyncio.sleep(0.01)
            continue
        try:
            result = HANDLERS[task.name](task.payload)
            print(f"  [{task.name}] 成功 -> {result}")
            broker.ack(task)
        except Exception as e:
            if task.attempts + 1 >= MAX_ATTEMPTS:
                print(f"  [{task.name}] 连续失败 {MAX_ATTEMPTS} 次 -> 进死信")
                broker.dead_letter(task, f"{type(e).__name__}: {e}")
            else:
                print(f"  [{task.name}] 第 {task.attempts + 1} 次失败，重试")
                broker.retry(task)

async def main():
    redis = fakeredis.FakeRedis(decode_responses=True)
    broker = Broker(redis)
    broker.submit("send_email", {"to": "ok@example.com"})
    broker.submit("send_email", {"to": "bad@example.com"})

    stop = asyncio.Event()
    task = asyncio.create_task(worker(broker, stop))
    await asyncio.sleep(0.3)          # 留出 worker 处理的时间
    stop.set()
    await task

    print("待处理:", redis.llen(QUEUE), " 处理中:", redis.llen(PROCESSING),
          " 死信:", redis.llen(DLQ))
    dead = json.loads(redis.lindex(DLQ, 0))
    print("死信内容:", dead["task"]["name"], "attempts=", dead["task"]["attempts"],
          "reason=", dead["reason"])

asyncio.run(main())

  [send_email] 成功 -> sent to ok@example.com
  [send_email] 第 1 次失败，重试
  [send_email] 第 2 次失败，重试
  [send_email] 连续失败 3 次 -> 进死信
待处理: 0  处理中: 0  死信: 1
死信内容: send_email attempts= 2 reason= ValueError: SMTP 拒收