格式转换流水线

上传一张图要等 12 秒

有损压缩和无损压缩的策略都搞定了。我把代码合并到上传流程里，满怀期待地测试了一下。

结果：用户上传一张 8.7MB 的图片后，要等 12 秒才能看到”上传成功”。

上传耗时分解：
  1. 客户端直传 OSS：                    2s
  2. 回调通知服务器：                    0.1s
  3. 服务器下载 OSS 原图到本地：          1.5s
  4. 智能分析内容类型：                   0.3s
  5. 生成 3 种尺寸 × 2 种格式 = 6 个文件： 6s
  6. 上传 6 个文件到 OSS：               2s
  总计：                                  12s

用户上传图片时盯着一个 loading 动画等 12 秒？不可接受。

问题出在哪？第 3~6 步都是同步执行的，阻塞了上传接口的返回。

解法很明确：上传归上传，处理归处理。异步化。

异步处理架构

核心思路：

之前（同步）：
  用户上传 → OSS → 回调 → 下载 → 压缩 × 6 → 上传 OSS → 返回（12秒）

之后（异步）：
  用户上传 → OSS → 回调 → 丢消息队列 → 立即返回（0.5秒）
                                      ↓
                               Worker 消费 → 下载 → 压缩 × 6 → 上传 OSS → 通知完成

修改上传回调接口，把处理任务丢到队列后立即返回：

效果：上传回调接口从 12 秒降到了 0.3 秒。

Worker：并行生成多格式多尺寸

Worker 从队列中消费任务，并行处理：

                    ┌──────────────────────┐
   RabbitMQ        │  image.process 队列   │
   image.process ──┤  prefetch_count = 4   │
                    └──────────┬───────────┘
                               │ 每次投递 1 个任务
                               ▼
                    ┌──────────────────────┐
                    │  Worker 进程          │
                    │  下载原图 → 生成变体   │
                    │  → 6 路并发上传 OSS   │
                    └──────────┬───────────┘
                               │ 全部成功 → ack
                               │ 失败 → nack 或进死信队列
                               ▼
                    ┌──────────────────────┐
                    │  guangying-img 桶     │
                    │  originals/ + derived/│
                    └──────────────────────┘

Worker 的核心是三件事：从队列取任务、把任务内部的处理并行化、正确地确认与重试。

先看派生规则。3 种尺寸 × 2 种格式不是随便定的——尺寸对应三档使用场景，格式对应一条降级链（WebP 优先，JPEG 兜底）：

# 派生图规格：(尺寸, 格式)
#   200  → 列表页缩略图
#   800  → 详情页主图
#   2000 → 原图浏览页
VARIANTS = [
    (200, "webp"),
    (200, "jpeg"),
    (800, "webp"),
    (800, "jpeg"),
    (2000, "webp"),
    (2000, "jpeg"),
]

并行化的关键在”任务内部的并行”。如果 6 个衍生图串行处理，一张图的耗时就接近 6 倍单图耗时；并发处理则受限于 CPU 而不是串行等待。整个 Worker 写成这样：

import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from io import BytesIO

import oss2
import pika
from PIL import Image

LOG = logging.getLogger("image-worker")

QUEUE_NAME = "image.process"
BUCKET_NAME = "guangying-img"
ENDPOINT = "https://oss-cn-beijing.aliyuncs.com"

# 单个任务内部的最大并行度：3 尺寸 × 2 格式 = 6
MAX_WORKERS = 6


def build_bucket() -> oss2.Bucket:
    auth = oss2.Auth(os.environ["OSS_ACCESS_KEY_ID"], os.environ["OSS_ACCESS_KEY_SECRET"])
    return oss2.Bucket(auth, ENDPOINT, BUCKET_NAME, connect_timeout=10)


def target_key(source_key: str, size: int, fmt: str) -> str:
    stem = os.path.splitext(os.path.basename(source_key))[0]
    return f"derived/{stem}_{size}.{fmt}"


def make_variant(source: bytes, size: int, fmt: str, quality: int = 80) -> bytes:
    """按目标尺寸和格式生成一张衍生图。"""
    with Image.open(BytesIO(source)) as img:
        img = img.convert("RGB")
        img.thumbnail((size, size), Image.LANCZOS)
        buf = BytesIO()
        if fmt == "webp":
            img.save(buf, format="WEBP", quality=quality, method=4)
        else:
            img.save(buf, format="JPEG", quality=quality, optimize=True)
        return buf.getvalue()


def handle_task(body: bytes, bucket: oss2.Bucket) -> None:
    task = json.loads(body)
    source_key = task["source_key"]

    # 幂等：目标对象已经存在就跳过，避免重复消费把 CPU 白烧
    pending = []
    for size, fmt in VARIANTS:
        key = target_key(source_key, size, fmt)
        if bucket.object_exists(key):
            continue
        pending.append((size, fmt, key))

    if not pending:
        return

    # 原图只下载一次，6 个变体共用同一份内存数据
    source = bucket.get_object(source_key).read()

    def produce(item: tuple[int, str, str]) -> None:
        size, fmt, key = item
        data = make_variant(source, size, fmt)
        bucket.put_object(key, data)

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        list(pool.map(produce, pending))

    LOG.info("task done, %d variants for %s", len(pending), source_key)


def on_message(channel, method, properties, body: bytes) -> None:
    try:
        handle_task(body, bucket)
    except Exception:
        LOG.exception("task failed, delivery_tag=%s", method.delivery_tag)
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        return
    channel.basic_ack(delivery_tag=method.delivery_tag)


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    bucket = build_bucket()
    connection = pika.BlockingConnection(pika.URLParameters(os.environ["AMQP_URL"]))
    channel = connection.channel()
    # 预取数量决定单个 Worker 同时处理几个任务，也是背压的抓手
    channel.basic_qos(prefetch_count=4)
    channel.basic_consume(queue=QUEUE_NAME, on_message_callback=on_message)
    LOG.info("worker started")
    channel.start_consuming()

三个参数直接决定了 Worker 的行为：

1. prefetch_count=4 决定单个 Worker 同时拿着几个任务。设小了，队列积压消不掉；设大了，一个 Worker 手里压着太多任务，任务延迟会变长，而且内存里会同时占着好几份原图。
2. MAX_WORKERS=6 决定单个任务内部的并发度。它不应该超过变体数量，否则线程在空转；也不应该低于变体数量，否则串行等待白白拖长。
3. 实例数 决定总吞吐。三者的关系是：总并发能力约等于实例数乘以 prefetch_count。

所以扩容策略可以很简单——按队列积压量扩缩容：

三条容易踩的坑：

1. 不要在每个变体里重新下载一次原图。原图可能有好几 MB，下载 6 次就是 6 倍流量和 6 倍等待。下载一次、内存里共享，是这个 Worker 里最值钱的一个优化。
2. 变体列表和上传流水线要一致。上传流水线产出的派生图命名必须和 Worker 写的对象 key 完全对得上，否则会出现”处理完的图找不到”或者”CDN 拿到半成品”。
3. 单张图的耗时不代表吞吐。串行时 6 个变体的耗时是 6 倍单图时间，并发之后接近 1 倍；但 CPU 占用会明显上升。压测要看的是”每分钟处理多少张图”，不是”每张图多久”。

前端进度通知

用户上传后立即返回”处理中”，前端通过轮询或 WebSocket 获取处理进度：

后端轮询接口：

处理进度优化：分阶段通知

一张图要生成 6~10 个衍生图，每完成一个都可以通知前端：

前端展示更细粒度的进度：

错误处理和重试

异步处理带来了新的挑战——Worker 可能崩溃、任务可能失败。需要健壮的重试机制：

任务失败
   │
   ▼
是不是可重试的错误？
   ├─ 否（图片损坏、格式不支持、鉴权失败）
   │     └─→ 记录失败原因，直接进死信队列，不再重试
   └─ 是（网络超时、5xx、限流）
         └─→ 指数退避后重试
               ├─ 重试次数未到上限 ──→ 重新入队
               └─ 重试次数用完 ─────→ 死信队列 + 告警

第一步是给错误分类。不是所有失败都值得重试——重试一个”文件已经损坏”的任务，只会让同一个错误重复三次，还白白占用 CPU。

import logging
import random

LOG = logging.getLogger("image-worker")

# 不可重试的错误：重试再多次也不会成功
PERMANENT_ERRORS = {
    "UnsupportedImageFormat",
    "CorruptedImage",
    "AccessDenied",
    "ImageTooLarge",
}

MAX_ATTEMPTS = 5
BASE_DELAY_SECONDS = 5


def is_retryable(exc: Exception) -> bool:
    """判断这个错误值不值得重试。"""
    code = getattr(exc, "code", None)
    if code in PERMANENT_ERRORS:
        return False
    if isinstance(exc, (TimeoutError, ConnectionError)):
        return True
    # OSS 返回 5xx 说明是服务端临时故障，可以重试
    status = getattr(exc, "status", None)
    if isinstance(status, int) and status >= 500:
        return True
    return False


def backoff_delay(attempt: int) -> float:
    """指数退避 + 随机抖动，避免大量任务在同一时刻集体重试。"""
    base = BASE_DELAY_SECONDS * (2 ** (attempt - 1))
    return base + random.uniform(0, base * 0.3)

第二步是重试的落点。有三种实现方式，我选的是”延迟重投”：

用 RabbitMQ 的 TTL 加死信交换机就能实现”延迟重投”：消息先进一个带 TTL 的延迟队列，过期后被死信交换机路由回主队列。

QUEUE_NAME = "image.process"
DLQ_NAME = "image.process.dlq"
DELAY_EXCHANGE = "image.dlx"


def dead_letter_args(ttl_ms: int) -> dict:
    return {
        "x-message-ttl": ttl_ms,
        "x-dead-letter-exchange": DELAY_EXCHANGE,
        "x-dead-letter-routing-key": QUEUE_NAME,
    }


def setup_topology(channel) -> None:
    channel.exchange_declare(exchange=DELAY_EXCHANGE, exchange_type="direct")
    channel.queue_declare(queue=QUEUE_NAME, durable=True)
    channel.queue_declare(queue=DLQ_NAME, durable=True)
    channel.queue_bind(queue=DLQ_NAME, exchange=DELAY_EXCHANGE, routing_key=DLQ_NAME)

    # 四个延迟档位：5 秒、30 秒、2 分钟、10 分钟
    for ttl in (5_000, 30_000, 120_000, 600_000):
        name = f"{QUEUE_NAME}.delay.{ttl}"
        channel.queue_declare(queue=name, durable=True, arguments=dead_letter_args(ttl))
        channel.queue_bind(queue=name, exchange=DELAY_EXCHANGE, routing_key=name)

第三步是幂等。重试的前提是重复执行不会造成副作用，所以任务必须能安全地重跑：

• 目标对象已存在就跳过。上传前先 bucket.object_exists(key) 判断。
• 上传用临时对象加完成后改名。避免消费者读到写了一半的文件。
• 计数用幂等的写入。任务处理次数用”事件去重表加唯一索引”来记，不要用”读出来加一再写回去”。

def put_object_atomic(bucket: oss2.Bucket, key: str, data: bytes) -> None:
    """先写临时对象，再改名到目标 key，避免消费者读到不完整文件。"""
    tmp_key = f"tmp/{key}"
    bucket.put_object(tmp_key, data)
    bucket.copy_object(tmp_key, key)
    bucket.delete_object(tmp_key)

第四步是失败之后要有人知道。死信队列里的任务如果没人管，就是静默的数据丢失。我给死信队列配了两类告警：

最后一条纪律：重试上限要设，但要留出人工重放的口子。死信队列不是垃圾桶——里面的任务要能被一键重放，所以任务体里必须带上完整的上下文（图片 ID、原图路径、失败原因、重试次数），而不是只留一个对象路径。

处理性能优化

一张图生成 6~10 个衍生图，如何更快？

上一章节

无损压缩

下一章节

内容审核