慢活别塞进请求里
原创 · 约 29 分钟阅读 · 阅读 --

慢活别塞进请求里

作者: 字与码

工程实践

古董级程序员，大厂出来后一直在创业公司，现在仍在一线写代码、做产品、和 AI 编程工具较劲。更完整的更新会放在微信公众号「字与码」：工作经历、新技术观察、AI 开发实践，以及这些年踩过的坑，都会陆续写在那里。

《AI时代的程序员修养》上一篇讲数据库。数据库是系统最硬的约束层，但不是所有事情都应该压在数据库和请求链路上。很多功能要跑得稳，靠的是缓存、队列和异步任务。

AI 很容易把慢操作直接写进接口里：请求来了，查库，调外部服务，生成文件，写历史记录，发通知，最后一起返回。代码看起来顺，用户等得久，系统也容易被一个慢下游拖死。

慢活别塞进请求里。但把慢活丢出去，也不等于问题消失。缓存有一致性，队列有堆积，异步任务有重复执行，重试有副作用，日志和历史记录也可能反过来拖慢主流程。

缓存、队列和异步任务各管什么

这三个东西经常一起出现，但职责不同。

组件主要解决什么最容易搞错什么
缓存减少重复计算和重复读取，降低延迟把它当成真实数据源
队列削峰、解耦、顺序化、延迟处理把它当成不会满的垃圾桶
异步任务把慢操作移出请求链路，后台完成忘记幂等、重试、进度和失败恢复

比如“用户上传一个 CSV，系统导入并生成报告”。

请求里只应该做几件事：

  1. 鉴权。
  2. 保存文件。
  3. 创建导入任务。
  4. 返回任务 ID。

解析、校验、批量写入、报告生成、通知用户，都应该进入后台流程。

这时缓存可以保存任务进度或中间计算结果；队列负责把任务交给 worker；异步任务负责执行导入。每个组件都有边界，不是简单地“加个 Redis”“加个 MQ”“开个后台线程”。

让 AI 写这类功能时，如果只说“异步处理”，它可能随手起一个线程。更好的指令是：

HTTP 请求只创建任务并返回 job_id。
文件解析、数据校验、批量写入、报告生成都由后台 worker 执行。
任务状态必须可查询，失败必须有错误码和错误摘要。
worker 重启后 running 超过 30 分钟的任务可以被重新捞起。
每个任务必须通过 job_id 幂等执行。

这段话把执行边界、状态恢复和幂等都写进去了。

缓存不是第二个数据库

缓存的诱惑在于它快。快到很多人会不自觉地把它当数据库用。

一个常见缓存代码：

async def get_user_profile(user_id: str):
    cached = await redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    profile = await db.get_user_profile(user_id)
    await redis.set(f"user:{user_id}", json.dumps(profile), ex=300)
    return profile

这段代码作为读缓存可以，但问题马上会出现：

  • 用户资料更新后，缓存什么时候失效？
  • 如果 DB 查不到，是不是也缓存空值？
  • 很多请求同时 miss，会不会一起打 DB？
  • 序列化字段变了，老缓存还能不能读？
  • Redis 挂了，接口是失败还是降级查 DB？

缓存设计至少要回答四件事：key、value、TTL、失效策略。

问题例子
key 怎么设计user_profile:{user_id}:v2
value 存什么响应模型，不要存 ORM 内部对象
TTL 多久5 分钟、1 小时，或者按业务变化频率
怎么失效写后删除、写后更新、自然过期、版本号

缓存 key 最好带版本。字段结构变化时，可以直接切到 v2，避免旧缓存反序列化失败：

cache_key = f"user_profile:v2:{user_id}"

缓存空值也很重要。比如用户不存在，如果每次都穿透到 DB，恶意请求可以打爆数据库：

if profile is None:
    await redis.set(cache_key, "__null__", ex=60)
    return None

空值 TTL 要短一些，避免刚创建的数据长时间查不到。

缓存击穿和 singleflight

缓存最大的问题不是 miss，而是同一热点 key 同时 miss。

假设首页配置缓存 5 分钟，刚好过期时来了 1000 个请求。如果每个请求都去查 DB 或调用配置服务，缓存反而变成了流量放大器。

本地进程里可以用 singleflight 思路：同一个 key 同一时刻只允许一个加载任务，其他请求等待结果。

class SingleFlightCache:
    def __init__(self):
        self._locks: dict[str, asyncio.Lock] = {}

    async def get_or_load(self, key: str, loader):
        cached = await redis.get(key)
        if cached is not None:
            return decode(cached)

        lock = self._locks.setdefault(key, asyncio.Lock())
        async with lock:
            cached = await redis.get(key)
            if cached is not None:
                return decode(cached)

            value = await loader()
            await redis.set(key, encode(value), ex=300)
            return value

这段代码只是示意。线上还要清理 _locks，处理 loader 异常，控制等待超时。如果是多进程或多机器，还要考虑分布式锁，或者用后台刷新、提前刷新、随机 TTL 降低同时过期概率。

让 AI 写缓存时，可以要求：

热点 key 需要防击穿。
同一进程内使用 singleflight，避免并发 miss 同时访问 DB。
TTL 加 10% 随机抖动，避免大量 key 同时过期。
Redis 失败时降级查 DB，但要记录指标并限制并发。

缓存不是一个 get/set 包装，缓存是另一条读路径。读路径一多，就要考虑一致性和故障模式。

队列不是垃圾桶

队列常被拿来“解耦”。调用方把任务写进队列，worker 慢慢处理，看起来主流程轻了。

但队列也有容量，worker 也有速度，下游也会慢。队列只是把压力从请求侧搬到后台侧。如果没有反压，后台侧迟早会爆。

一个任务消息至少应该包含：

{
  "job_id": "job_123",
  "job_type": "import_csv",
  "dedupe_key": "user_1:file_abc",
  "attempt": 0,
  "created_at": "2026-07-03T10:00:00+08:00",
  "payload": {
    "file_id": "file_abc",
    "user_id": "user_1"
  }
}

job_id 用来追踪，dedupe_key 用来防重复，attempt 用来控制重试。不要只塞业务 payload。

队列设计要先回答：

  • 队列最大能堆多少？
  • worker 每秒能处理多少？
  • 下游服务能承受多少并发？
  • 任务失败后怎么重试？
  • 重试多少次进死信？
  • 同一个用户的任务要不要限速？
  • 同一个业务对象的任务要不要保序？
  • 队列堆积时，前端能不能继续提交？

如果这些没设计，AI 生成的消费者通常会像这样：

async def worker():
    while True:
        message = await queue.get()
        await process(message)
        await queue.ack(message)

这段代码没有超时，没有失败分类，没有重试，没有死信，没有并发上限，也没有退出逻辑。能跑，但不够用。

幂等消费比重试更重要

队列系统通常至少保证“尽量送到”，不保证“只送一次”。消息可能重复，worker 可能处理成功但 ack 失败，消费者重启后可能重新拿到同一条消息。

所以消费者必须幂等。

一个处理记录表可以这样设计：

create table processed_events (
  event_id text primary key,
  status text not null,
  processed_at timestamptz not null default now(),
  error_code text
);

消费者先占位，再执行：

async def consume(event: Event):
    inserted = await processed_events.try_insert(event.event_id, status="running")
    if not inserted:
        return

    try:
        await handle_event(event)
        await processed_events.mark_done(event.event_id)
    except RetryableError as exc:
        await processed_events.mark_retryable(event.event_id, exc.code)
        raise
    except PermanentError as exc:
        await processed_events.mark_failed(event.event_id, exc.code)
        await dead_letters.add(event, reason=exc.code)

如果外部副作用本身不幂等，比如发短信、扣款、创建工单，就必须用业务幂等键约束对方接口，或者把副作用状态也落库。

比如发通知可以记录：

create unique index idx_notifications_dedupe
on notifications (user_id, template_id, dedupe_key);

同一用户、同一模板、同一去重键只能创建一次通知。即使消息重复，也不会重复发。

让 AI 写消费者时，可以加硬约束：

消费者必须按 event_id 幂等。
外部副作用必须有业务 dedupe_key。
重试可能导致同一消息重复执行，代码必须安全。
请写重复投递同一 event_id 的测试。

没有幂等，重试就是放大事故。

重试要分类型

“失败就重试”是一句危险的话。

错误至少要分两类：

类型示例处理
临时错误超时、连接失败、下游 503、限流延迟重试
永久错误参数非法、资源不存在、权限不足、格式错误不重试，进死信或标失败

如果把永久错误也重试，队列会被坏消息卡住。如果把临时错误直接标失败，系统会丢掉本可以恢复的任务。

重试还要退避：

def next_retry_delay(attempt: int) -> int:
    delays = [10, 30, 120, 300, 900]
    return delays[min(attempt, len(delays) - 1)]

也可以加随机抖动，避免大批任务同一时间重试。

任务表里通常需要这些字段：

attempt integer not null default 0,
max_attempts integer not null default 5,
next_run_at timestamptz,
last_error_code text,
last_error_message text

worker 执行失败时更新：

update jobs
set attempt = attempt + 1,
    status = case
      when attempt + 1 >= max_attempts then 'failed'
      else 'pending'
    end,
    next_run_at = :next_run_at,
    last_error_code = :error_code,
    last_error_message = :error_message
where id = :job_id;

这里也要注意事务边界。任务状态更新要可靠，不能因为日志写入失败导致任务状态没更新。

死信不是失败垃圾堆

死信队列经常被建出来，然后没人看。那它就只是另一个垃圾堆。

死信应该是可排查、可重放、可统计的失败入口。

死信记录至少要包含：

  • 原始消息。
  • 业务 key，如 job_id、user_id、order_id。
  • 失败类型和错误码。
  • 最后一次错误摘要。
  • 尝试次数。
  • 首次失败时间和最后失败时间。
  • 是否可人工重放。
  • 处理状态。
create table dead_letters (
  id uuid primary key,
  queue_name text not null,
  message_key text not null,
  payload jsonb not null,
  error_code text not null,
  error_message text,
  attempts integer not null,
  first_failed_at timestamptz not null,
  last_failed_at timestamptz not null,
  status text not null check (status in ('open', 'ignored', 'replayed', 'resolved'))
);

有了表，还要有运营动作：查询、过滤、重放、忽略、批量处理、报警。否则死信只是在数据库里安静地堆积。

让 AI 写死信逻辑时，可以要求：

死信必须可查询、可重放、可标记处理。
死信记录保留原始 payload、错误码、尝试次数和业务 key。
重放时必须生成新的 replay_id，但保留原 event_id 用于幂等判断。

死信的价值不在“失败了放哪里”，而在“失败后怎么恢复”。

批处理比逐条写稳得多

后台任务经常要写日志、写历史、更新统计、同步状态。AI 很容易写成每处理一条就写一次数据库。

如果任务有 20 万条记录，每条都写一条日志，每条都 commit 一次，性能会非常差，还可能让数据库反过来拖慢 worker。

更稳的做法是内存队列 + 后台批量写：

class AsyncLogWriter:
    def __init__(self, batch_size: int = 1000, flush_interval: float = 5.0):
        self.queue = asyncio.Queue(maxsize=50_000)
        self.batch_size = batch_size
        self.flush_interval = flush_interval

    async def write(self, record: dict):
        try:
            self.queue.put_nowait(record)
        except asyncio.QueueFull:
            if record.get("level") == "DEBUG":
                return
            raise

    async def run(self):
        batch = []
        while True:
            try:
                item = await asyncio.wait_for(self.queue.get(), timeout=self.flush_interval)
                batch.append(item)
                if len(batch) < self.batch_size:
                    continue
            except asyncio.TimeoutError:
                pass

            if batch:
                await insert_logs_batch(batch)
                batch.clear()

这里有几个关键点：

  • 执行进程只把日志放进内存队列。
  • 后台 task 批量写数据库。
  • 队列有上限。
  • 队列满时，低价值 DEBUG 日志可以丢弃或采样。
  • 数据库慢不能无限反压主执行流程。

这不是只适用于日志。很多历史记录、遥测明细、低优先级审计、统计事件都可以类似处理。关键是要区分“必须同步落库的业务状态”和“可以异步批量写入的观测数据”。

让 AI 设计这类写入时，可以要求：

不要在主执行循环里逐条同步写数据库。
stdout/stderr reader 只负责把日志放入内存队列。
后台 task 按 1000-5000 条或 5 秒批量写 DB。
队列满时 DEBUG 日志采样或丢弃摘要，不能反压执行进程。
业务状态更新仍然必须可靠落库。

这类约束能明显改变系统吞吐。

反压是系统说“我忙不过来”

队列能削峰，但不能消灭峰值。如果生产速度长期大于消费速度，队列迟早满。

反压就是系统告诉上游：我忙不过来了。

常见反压策略：

场景策略
队列长度超过阈值拒绝新任务，返回稍后再试
用户提交太快按用户限流
低优任务过多降低优先级或延迟执行
下游慢减小 worker 并发
日志队列满丢弃 DEBUG，保留 ERROR 摘要
批处理落后暂停导入入口或降低批次

反压不是失败，而是保护系统。

例如创建任务前先检查用户维度的 pending 数：

select count(*)
from jobs
where user_id = :user_id
  and status in ('pending', 'running');

超过限制就拒绝：

{
  "error": {
    "code": "TOO_MANY_PENDING_JOBS",
    "message": "当前排队任务较多，请稍后再试",
    "retryable": true
  }
}

如果没有反压，系统会表现得很“努力”：接口继续收任务，队列继续堆，worker 继续追，数据库继续慢，最后所有人一起超时。

异步任务也需要状态机

异步任务不是后台随便跑一段代码。它应该有状态机。

一套常见状态：

状态含义
pending已创建，等待执行
runningworker 已抢占，正在执行
success执行成功
failed不可恢复失败或超过重试次数
retrying等待下次重试
cancelled用户或系统取消
timeout执行超过租约

worker 抢任务不要先查再改，要用原子更新或 skip locked：

select *
from jobs
where status = 'pending'
  and run_at <= now()
order by priority desc, run_at asc
limit 100
for update skip locked;

抢到任务后写 locked_by 和 locked_at。如果 worker 崩溃，后台巡检可以把 running 且 locked_at 太久的任务捞回来。

update jobs
set status = 'pending',
    locked_by = null,
    locked_at = null,
    run_at = now()
where status = 'running'
  and locked_at < now() - interval '30 minutes';

这就是可恢复性。没有状态机和租约，worker 重启后任务就可能永远停在 running。

一份给 AI 的缓存队列提示词

让 AI 写缓存、队列或异步任务时，可以先给它这段指令：

先不要写代码。

请为这个功能设计缓存、队列和异步任务方案：
1. 哪些操作必须在请求内完成，哪些必须异步。
2. 缓存 key、value、TTL、失效策略、空值缓存和击穿保护。
3. 队列消息 schema：job_id/event_id、业务 key、attempt、payload、trace_id。
4. worker 并发上限、下游限流、任务抢占方式和租约。
5. 幂等设计：重复消息、重复任务、外部副作用如何处理。
6. 错误分类：临时错误、永久错误、重试次数、退避策略。
7. 死信设计：记录字段、查询、重放、忽略、报警。
8. 批量写入：哪些日志或历史记录可以异步批量写，批量大小和刷新间隔。
9. 反压策略：队列满、用户任务过多、下游变慢时怎么保护系统。
10. 状态查询接口、测试用例和关键指标。

方案确认后，再写代码。

这段 prompt 的目的，是让 AI 先承认系统会慢、会满、会重复、会失败。承认这些，代码才有可能稳。

写之前问自己

动手前，至少问一遍：

  • 这个操作真的要在请求里完成吗？
  • 缓存失效时，会不会同时打爆下游？
  • 缓存里的数据是不是可能比数据库旧？
  • 消息重复投递会不会重复产生副作用？
  • 临时错误和永久错误有没有分开？
  • 死信有没有人能看、能重放、能关闭？
  • 队列满了是拒绝、降级，还是继续堆？
  • 批量写日志或历史记录会不会反压主流程？
  • worker 崩溃后，任务能不能被重新捞起？
  • 有没有指标能看到队列长度、消费延迟、重试和死信？

缓存、队列和异步任务不是为了让代码看起来高级。它们的价值，是让系统在慢、满、错、抖的时候仍然能保持边界。

AI 可以很快写出 Redis 访问、队列消费者和后台 worker。程序员要做的是把“后台跑一下”变成可恢复、可观测、可限流、可解释的系统。

本专栏其他文章

查看专栏「AI时代的程序员修养」全部文章