《AI时代的程序员修养》上一篇讲数据库。数据库是系统最硬的约束层但不是所有事情都应该压在数据库和请求链路上。很多功能要跑得稳靠的是缓存、队列和异步任务。AI 很容易把慢操作直接写进接口里请求来了查库调外部服务生成文件写历史记录发通知最后一起返回。代码看起来顺用户等得久系统也容易被一个慢下游拖死。慢活别塞进请求里。但把慢活丢出去也不等于问题消失。缓存有一致性队列有堆积异步任务有重复执行重试有副作用日志和历史记录也可能反过来拖慢主流程。缓存、队列和异步任务各管什么这三个东西经常一起出现但职责不同。组件主要解决什么最容易搞错什么缓存减少重复计算和重复读取降低延迟把它当成真实数据源队列削峰、解耦、顺序化、延迟处理把它当成不会满的垃圾桶异步任务把慢操作移出请求链路后台完成忘记幂等、重试、进度和失败恢复比如“用户上传一个 CSV系统导入并生成报告”。请求里只应该做几件事鉴权。保存文件。创建导入任务。返回任务 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(fuser:{user_id}) if cached: return json.loads(cached) profile await db.get_user_profile(user_id) await redis.set(fuser:{user_id}, json.dumps(profile), ex300) return profile这段代码作为读缓存可以但问题马上会出现用户资料更新后缓存什么时候失效如果 DB 查不到是不是也缓存空值很多请求同时 miss会不会一起打 DB序列化字段变了老缓存还能不能读Redis 挂了接口是失败还是降级查 DB缓存设计至少要回答四件事key、value、TTL、失效策略。问题例子key 怎么设计user_profile:{user_id}:v2value 存什么响应模型不要存 ORM 内部对象TTL 多久5 分钟、1 小时或者按业务变化频率怎么失效写后删除、写后更新、自然过期、版本号缓存 key 最好带版本。字段结构变化时可以直接切到v2避免旧缓存反序列化失败cache_key fuser_profile:v2:{user_id}缓存空值也很重要。比如用户不存在如果每次都穿透到 DB恶意请求可以打爆数据库if profile is None: await redis.set(cache_key, __null__, ex60) 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), ex300) 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:0008: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, statusrunning) 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, reasonexc.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 textworker 执行失败时更新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(maxsize50_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(), timeoutself.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 lockedselect * 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. 队列消息 schemajob_id/event_id、业务 key、attempt、payload、trace_id。 4. worker 并发上限、下游限流、任务抢占方式和租约。 5. 幂等设计重复消息、重复任务、外部副作用如何处理。 6. 错误分类临时错误、永久错误、重试次数、退避策略。 7. 死信设计记录字段、查询、重放、忽略、报警。 8. 批量写入哪些日志或历史记录可以异步批量写批量大小和刷新间隔。 9. 反压策略队列满、用户任务过多、下游变慢时怎么保护系统。 10. 状态查询接口、测试用例和关键指标。 方案确认后再写代码。这段 prompt 的目的是让 AI 先承认系统会慢、会满、会重复、会失败。承认这些代码才有可能稳。写之前问自己动手前至少问一遍这个操作真的要在请求里完成吗缓存失效时会不会同时打爆下游缓存里的数据是不是可能比数据库旧消息重复投递会不会重复产生副作用临时错误和永久错误有没有分开死信有没有人能看、能重放、能关闭队列满了是拒绝、降级还是继续堆批量写日志或历史记录会不会反压主流程worker 崩溃后任务能不能被重新捞起有没有指标能看到队列长度、消费延迟、重试和死信缓存、队列和异步任务不是为了让代码看起来高级。它们的价值是让系统在慢、满、错、抖的时候仍然能保持边界。AI 可以很快写出 Redis 访问、队列消费者和后台 worker。程序员要做的是把“后台跑一下”变成可恢复、可观测、可限流、可解释的系统。
