D9 Concurrency and Quotas: Starting Multiple Episodes at Once Without Blowing Through Any Provider's Limits

今日目标

能用队列让多集并行生产，并按 provider 分别限制并发与速率
能给任务排优先级，让插队的急稿不至于饿死排队的长任务
能在超限、限流、排队超时之间做出正确的退避与降级决策

昨天的引擎已经能把一集稳稳当当地跑完，也能断点续跑。今天把它从「一次一集」推到「五集同时开机」，你会立刻撞上一个新问题：机器有的是，额度只有那么多。读完回到页面顶部把这三条勾掉。

小白版讲解

同时开三个组，但只有一台摇臂

剧组扩产的时候，制片主任第一反应是「多开几个组同时拍」。第二反应通常在半天之内到来：三个组同时要用那台摇臂，而摇臂只有一台。人可以多招，灯可以多租，但那台摇臂就是一台。

生成式流水线里的摇臂就是厂商配额。你的进程可以开一百条，云主机可以加到十六核，但 MiniMax 给你的速率上限不会因此变大。这些数字是公开的，我们照着官方速率限制页抄下来，按充值用户那一档算：

接口	充值用户 RPM
语音 T2A v2	20
图像生成	10
视频生成 v1	20
文本 MiniMax-M3	200

先把这几个数字换算成你熟悉的单位。一集短剧四十个分镜，每个分镜至少一句台词，那就是四十次语音合成。语音是每分钟二十次，光配音这一步就要排两分钟，而这还是一集。五集并行就是两百次，十分钟。首帧更紧：图像每分钟十次，五集两百张首帧要二十分钟，比配音还慢一倍。

这两个数字会颠覆一个很自然的直觉。多数人以为并行的瓶颈在「视频生成太慢」，于是拼命去优化视频那一步；但真按配额算下来，卡住整条线的往往是那个看起来最便宜、最快的环节。图像一张两分半钱、几秒钟就返回，可它一分钟只让你调十次。

所以今天要做的不是「把并发调大」，而是给每一家厂商的每一类接口装一道闸门，让你的进程主动把自己限制在配额之内。听起来是自我设限，实际上是相反的：只有主动限住，你才能一直跑在配额的上沿；不限的那种写法会在冲过头之后被厂商拒绝，退避、重试、再冲过头，平均吞吐反而更低。

并行的单位：按集还是按镜头

装闸门之前先定一件事：并行的单位是什么。

按集并行是最省事的：五个进程，各自跑完一集的六个环节。好处是隔离干净，一集崩了不影响另外四集；坏处是失败半径大。一集跑到第五个环节炸了，这一集从头到尾两个多小时的产物全悬在那儿，而你的账已经花出去了。

按镜头并行把粒度切细：所有集的所有镜头进同一个池子，谁的依赖满足了就跑谁。好处是资源利用率高——某一集在等剧本，另一集的四十个镜头正好把语音配额吃满；坏处是编排复杂，你必须自己维护依赖关系，还要能说清楚「第三集现在到底完成了百分之多少」。

本课的选择是按镜头并行，按集统计。执行的粒度是镜头，因为它才能把配额用满；但对外汇报、成本台账、失败重跑都按集来，因为读者关心的是「第三集能不能上线」，不是「第三集的第十七镜怎么样了」。这个组合的代价是要多写一层：任务带着自己属于哪一集的标签，统计的时候按标签聚合。多写的这几十行是值得的。

顺便说一个反直觉的判断：并行度不该由你有几核 CPU 决定。这条线上几乎没有本地计算，全是等网络。真正决定并行度的是配额，以及你愿意为「一次失败要重跑多少东西」付多少钱。

每家厂商一道闸门

闸门有两个互不相同的部件，很多人会把它们混成一个，然后调出一堆莫名其妙的现象。

第一个是并发上限，管的是「同一时刻挂着几个请求」。这个数字是你自己定的，用来保护你自己的进程、内存和钱包。第二个是令牌桶，管的是「一分钟发几次」，这个数字是厂商定的，就是上面那张 RPM 表。

为什么不能只要一个？只有并发上限的话，如果每个请求都很快返回，一分钟能发出去几百次，照样撞限流。只有令牌桶的话，一分钟发二十次没错，但这二十次可能同时在飞，你的内存里同时挂着二十个待下载的视频。两个都要。

令牌桶的形状值得说一句。最偷懒的限速写法是每次请求之前睡一会儿，比如二十 RPM 就每次睡三秒。这么写限住了速率，却把突发能力也一起限掉了：真实的配额允许你在一分钟的开头连着冲二十次，然后安静五十秒。令牌桶保留了这个突发能力，代价是你要维护一个窗口和一个计数。

limiter.js

// 令牌桶：每个窗口发 capacity 个令牌。关键是 tryTake 不阻塞——
// 拿不到就立刻返回 false，让调度器去干别的，而不是在这里干等。
class TokenBucket {
  constructor(capacity, windowMs = 60_000) {
    this.capacity = capacity
    this.windowMs = windowMs
    this.tokens = capacity
    this.windowStart = Date.now()
  }
 
  tryTake() {
    const now = Date.now()
    if (now - this.windowStart >= this.windowMs) {
      this.windowStart = now
      this.tokens = this.capacity
    }
    if (this.tokens <= 0) return false
    this.tokens -= 1
    return true
  }
}
 
// RPM 来自官方速率限制页；并发上限是你自己定的，两者是两回事。
const QUOTA = {
  image: { rpm: 10, concurrency: 2 },
  tts: { rpm: 20, concurrency: 3 },
  video: { rpm: 20, concurrency: 2 },
}

limiter.py

import time
 
 
class TokenBucket:
    """每个窗口发 capacity 个令牌。try_take 不阻塞：拿不到就让调度器去干别的。"""
 
    def __init__(self, capacity: int, window_s: float = 60.0) -> None:
        self.capacity = capacity
        self.window_s = window_s
        self.tokens = capacity
        self.window_start = time.monotonic()
 
    def try_take(self) -> bool:
        now = time.monotonic()
        if now - self.window_start >= self.window_s:
            self.window_start = now
            self.tokens = self.capacity
        if self.tokens <= 0:
            return False
        self.tokens -= 1
        return True
 
 
# RPM 来自官方速率限制页；并发上限是你自己定的，两者是两回事。
QUOTA = {
    "image": {"rpm": 10, "concurrency": 2},
    "tts": {"rpm": 20, "concurrency": 3},
    "video": {"rpm": 20, "concurrency": 2},
}

注意 tryTake 是非阻塞的。这一个细节决定了整套调度好不好用，下面第四节会看到它的后果。

限流不是错误，是信号

MiniMax 的响应体里有一个 base_resp.status_code，0 是成功，其中和今天有关的是 1002 限流与 1039（TPM 维度的限流）。很多人把它们和 1004 鉴权失败、1008 余额不足、2013 参数无效一起塞进 catch 里打一行日志了事，这是浪费。

把这几个码分成两类就清楚了：重试有可能变好的（限流、服务端错误）和重试一定不会变好的（鉴权、余额、参数、内容审核）。前者退避重试，后者一次都不要重试——重试只会让你在一分钟里把同一个错误犯五遍，还占着配额。

对可重试的那一类，退避要做三件事，少一件都不算做完：

第一，指数退避加抖动。五集同时撞上限流，如果都按固定的一秒重试，一秒后它们会一起醒来再撞一次。抖动就是给每个退避时间乘一个零点七到一点三之间的随机数，把它们错开。

第二，退避期间不要占着执行流。这是下一节的主题。

第三，收到限流之后主动降速。这一条最容易漏。限流说明你的发送速率超过了厂商此刻愿意接受的速率，那么退避完了以后还用原来的速度冲，只会再撞一次。正确做法是把令牌桶罚一档：接下来一两个窗口只发一半令牌，等确认不再撞了再恢复。

backoff.js

// 指数退避 + 抖动。抖动是为了避免五集同时醒来、再一起撞一次。
const backoffMs = (attempt) => Math.round(300 * 2 ** (attempt - 1) * (0.7 + Math.random() * 0.6))
 
function onFailure(task, err, gate) {
  const rateLimited = err.statusCode === 1002 || err.statusCode === 1039
  if (rateLimited) {
    gate.stats.rateLimited += 1
    gate.bucket.penalize() // 接下来两个窗口只发一半令牌
  }
  // 鉴权 1004、余额 1008、参数 2013、内容审核 1026/1027 一次都不重试
  if (!err.retryable || task.attempt + 1 >= 5) return scheduler.fail(task, err)
  scheduler.requeue(task, backoffMs(task.attempt + 1))
}

backoff.py

import random
 
 
def backoff_s(attempt: int) -> float:
    """指数退避 + 抖动。抖动是为了避免五集同时醒来、再一起撞一次。"""
    return 0.3 * 2 ** (attempt - 1) * random.uniform(0.7, 1.3)
 
 
def on_failure(task, err, gate, scheduler) -> None:
    if err.status_code in (1002, 1039):
        gate.stats.rate_limited += 1
        gate.bucket.penalize()  # 接下来两个窗口只发一半令牌
    # 鉴权 1004、余额 1008、参数 2013、内容审核 1026/1027 一次都不重试
    if not err.retryable or task.attempt + 1 >= 5:
        scheduler.fail(task, err)
        return
    scheduler.requeue(task, backoff_s(task.attempt + 1))

单个异步任务本身该怎么提交、怎么轮询、超时怎么放弃，是第四天讲过的事，今天不重复；今天只管「很多个任务一起来的时候，闸门该怎么开」。

优先级与公平：三档队列，以及怎么不饿死

急稿总会有。运营半夜说「第六集明天上午要用」，这一集就得插到前面去。做法很直接：给任务分三档，急稿、普通、低优先，调度的时候先看档位，同档按入队时间先来先服务。

真正的坑在下一步：低优先级的任务可能永远排不上。只要急稿源源不断地进来，最早入队的那批低优先任务就一直是队尾。这就是饿死。

防饿死最省事的办法叫老化：一个任务在队列里等得越久，它的有效优先级就越高。等过一档的时间就升一档。这样即使急稿不断，那些等了很久的任务也会慢慢浮上来。

老化有一条必须写死的约束：升档要有上限，不许升进急稿那一档。否则跑上半小时之后队列里全是急稿，这一档就名存实亡了。本课的做法是低优先最多升到普通，急稿这一档只留给人工插队。

还有一个更隐蔽的坑，它会让你的优先级完全失效，而且日志上看不出来。如果任务是先被调度出去、再在里面等配额，那么当三个工作槽都被低优先任务占着、而它们全都在等令牌时，急稿连被调度的机会都没有——队列里排第一有什么用，没人来取。

这就是上一节强调 tryTake 必须非阻塞的原因。正确的结构是调度器在派活之前先问闸门要许可：拿到了才把任务交给工作槽，拿不到就跳过它去看下一个候选。这样配额紧张的时候，令牌会按优先级发给最该拿的那个任务。

长任务不该占着工作槽

视频生成是这条线上唯一的长任务。提交完拿到一个 task_id，然后就是等——官方没有公布典型耗时，你只能按实测来，而且随排队波动。

问题在于，如果你的代码写成一句 await video.generate(...)，这条执行流就被一个「除了等什么也不干」的任务占死了。三个工作槽、十八个视频任务，前三个视频一提交，整条线就停摆了：剧本能跑、配音能跑、时间轴能跑，但没有槽位去跑它们。

解决办法是把长任务的生命周期拆成两段：提交占用工作槽（它是一次很短的 HTTP 请求），等待不占用工作槽。等待期间闸门的票还得继续占着——厂商那边确实还挂着一个任务，这一点不能骗自己；但你的执行流应该立刻回到队列去接别的活。

worker.js

// start() 返回 undefined 表示做完了；返回一个函数表示「已经交给厂商，接下来只是等」。
const task = {
  id: 'ep6-clip-s01',
  gate: 'video',
  start: async () => {
    // 注意没有 await：把 promise 包成函数交回给调度器，工作槽当场还回去
    const inflight = providers.video.generate({ prompt, outPath, durationSec: 6 })
    return async () => {
      await inflight
      run.record({ nodeId: 'ep6-clip-s01', kind: 'video', units: 6 })
    }
  },
}
 
const pending = await task.start()
if (pending) {
  log('已提交给厂商，释放工作槽去接别的活')
  background.push(pending().then(() => scheduler.complete(task)))
} else {
  scheduler.complete(task)
}

worker.py

import asyncio
 
 
async def start_clip(providers, run, prompt, out_path):
    """返回 None 表示做完了；返回一个协程表示「已交给厂商，接下来只是等」。"""
    # 用 create_task 把它挂到事件循环上，当前这条执行流立刻返回
    inflight = asyncio.create_task(
        providers.video.generate(prompt=prompt, out_path=out_path, duration_s=6)
    )
 
    async def wait() -> None:
        await inflight
        run.record(node_id="ep6-clip-s01", kind="video", units=6)
 
    return wait
 
 
pending = await task.start()
if pending is not None:
    log("已提交给厂商，释放工作槽去接别的活")
    background.append(asyncio.create_task(finish(pending, task)))
else:
    scheduler.complete(task)

同样的道理适用于任何等待：等令牌、等退避、等下游依赖，都不该占着工作槽。判断标准很好记——如果这段时间里你的代码什么都不做，它就不该拿着一个槽位。

这一段和消息队列那门课的关系

看到这里，学过分布式的读者会觉得眼熟：这不就是一个带优先级的工作队列加上限流器吗？是的。今天写的这套东西，在生产环境里通常是 Redis Streams 的消费组、或者一个现成的任务队列，加上一层网关侧的限流。

那为什么要自己写一遍？因为本课要教的判断不在队列本身，而在**「闸门按什么维度分桶」**。现成的队列会给你并发控制，但它不知道你的图像和语音是两家不同的配额池，也不知道你的视频任务有八成时间在等而不在算。这两件事只有你知道，也只有你能配。

如果你在做真正的生产系统，建议的边界是：队列的持久化、消费组、失败重投用现成的，而闸门与优先级策略自己写——它只有一两百行，却是最需要按你的账单调整的部分。平台上「30 天从前端工程师到 Agent 工程师」那门课有一天专门讲 Redis Streams 的消费组与待处理列表，想把今天的内存队列换成跨进程的版本，从那儿接上去最省事。

源码导读

动手实验

🧪 D9 实验：按 provider 限流、支持优先级的并行生产队列

Code location: labs/ai-drama-pipeline/day-09-parallel-episodes

验收标准：

闸门统计里每个 provider 的峰值在飞数都不超过它的并发上限，且「令牌拦下」不为零。
用 INJECT=ratelimit 跑一遍，限流响应被识别、退避后重新入队，最终全部成功，没有任务失败。
第六集作为急稿中途插队，日志里能看到它插到已就绪的低优先级任务前面。
低优先级任务被老化升档，日志里有「饿死保护」这一行。
视频任务提交后打印「释放工作槽去接别的活」，释放次数等于视频任务数。

实验默认把速率窗口压缩成六秒，好让整个流程十几秒跑完，配额数字一个都没改；把 RATE_WINDOW_MS 设成 60000 就是真实行为。starter 挖了五个练习点，原样跑会看到六项核对里有五项是叉，做完一个变一个勾。

故障注入用全课统一的 INJECT 环境变量，本天只有一个取值：

BashBash

MOCK=1 pnpm start                    # 正常路径
INJECT=ratelimit MOCK=1 pnpm start   # 前 1 次图像、前 2 次语音返回 status_code=1002

注入只作用在 provider 出口，闸门、调度、优先级、老化这些逻辑照常执行——这是它和「跳过这一步」的区别。

实现令牌桶与并发信号量的非阻塞准入，跑一遍确认「令牌拦下」不再是零，且峰值在飞数落在上限之内。
让五集同时进队列，看闸门统计里图像那一行——它会是第一个被卡住的，跟正文算的账对得上。
换成 INJECT=ratelimit 再跑一遍，观察退避日志，确认退避带抖动、桶被罚了一档、且任务最终成功。
补上优先级排序与老化，观察急稿插队的次数与「饿死保护」日志同时出现。
把视频任务改成提交后归还工作槽，对比修改前后整体耗时的差别。

面试题

今天三道题在下方题库区，侧重并行粒度与失败半径、按 provider 的限流设计、限流响应的正确处理。展开后先看分析过程再看要点——照着推导练，比背要点管用。标注了国内高频与海外高频，方便按目标市场取舍。

检查清单与明日预告

能用队列让多集并行生产，并按 provider 分别限制并发与速率
能给任务排优先级，让插队的急稿不至于饿死排队的长任务
能在超限、限流、排队超时之间做出正确的退避与降级决策
能说清并发上限与令牌桶分别在管什么，以及为什么两个都要
能解释为什么配额等待不能占着工作槽，否则优先级会静默失效
实验的 5 条验收标准全部通过
3 道面试题不看要点也能答出至少 2 道

明天 D10 我们做审片室：一个能按镜头看图看片、改台词、重生成单个镜头的网页后台。顺序是有意的——今天刚把产能放大到五集并行，产量一上去，人工检查就成了新的瓶颈；而人要插手，第一件事是得看得见。更关键的是，今天你已经知道每一次重新生成都要从配额里扣，所以明天算「改一句台词到底要重跑哪些节点」才有意义：算不准，省下的时间全赔在重复的调用上。

Concurrency and Quotas: Starting Multiple Episodes at Once Without Blowing Through Any Provider's Limits