大客户一忙，小客户就排队
「回声」把同一组推理容量卖给了多个业务方。某个大客户做活动，请求量突然涨了十倍；几分钟后，几个小客户陆续投诉：平时几十毫秒的接口，现在要等好几秒。
从网关整体看，系统没有故障：容量跑满，错误率不高，平均延迟只是「有所上升」。但从小客户视角看，他们的请求几乎排在所有大客户请求之后。FIFO 只认到达顺序，不认租户；大客户请求多，自然占满队列。
共享容量必须先回答两个问题：每个租户应该拿到多少，以及某个租户最少能拿到多少。前者是权重，后者是饥饿防护。配额则回答第三个问题：某个租户最多能消耗多少。本节按这个顺序展开，文中数字来自 ai-lab 的确定性公平调度模拟，标注为示例样本上的运行结果。
加权公平队列
最简单的多租户做法是按租户轮询：每个租户轮流取一个请求。它保证每个租户机会均等，但无法表达「大客户买了四倍容量」。加权公平队列（WFQ）把轮询推广成比例分配。
每个租户维护一个虚拟完成时间。请求入队时计算：
finish = max(该租户上次 finish, 全局虚拟时间) + 成本 / 权重
每次出队选择 finish 最小的请求。权重越高，同样的成本让 finish 增长越慢，长期获得的服务量就越多。取 max 的作用是防止空闲租户攒下大量信用：一个租户停了一阵再回来，不能凭「以前没用满」一次性插到所有人前面。
WFQ 的公平性是长期的、按成本的。它不保证每个请求的延迟，只保证在持续积压时，各租户的服务量收敛到权重比例。
饥饿防护不是增加容量
权重 8:2:1 意味着低权重租户长期只拿到约 9% 的容量。如果它的请求还很少，单个请求可能要等很多轮才被服务。等待超过阈值后，调度器可以显式覆盖 WFQ 顺序，让「被相对亏待最久」的请求先走。
选择哪个超时请求也有讲究。按到达先后挑，高权重租户永远有更老的请求挡在前面，低权重租户仍然排不上。这里用「等待时长 / 权重」排序：权重 1 的租户等 200ms，和权重 50 的租户等 200ms，前者的相对等待是后者的 50 倍，应当先走。
必须说清楚它的代价：饥饿防护不创造容量，只重新分配等待。下面的实验里，开启防护后低权重租户的最坏等待显著下降，但高权重租户的最坏等待上升，总平均等待几乎不变。防护阈值设得越低，比例公平被破坏得越多。
配额与突发额度
权重解决「怎么分」，配额解决「最多用多少」。配额用令牌桶实现：桶容量是突发额度，补充速率是持续配额。请求入队前先扣令牌，扣不动就拒绝，并返回补满所需的时间。
只有突发没有补充，额度用完就永久拒绝；只有补充没有突发，任何正常峰值都会被误杀。两个参数要一起定：突发额度覆盖短促高峰，补充速率对齐合同里的长期配额。
配额和公平是两件事。配额超限时请求根本不该进队列；公平只在已经进队的请求之间分配顺序。把两者混成一个「限流」开关，会导致超额租户仍然占着队列位置。
实现
下面的模块包含令牌桶、加权公平队列和饥饿防护。它不访问存储，配额状态在进程内；多实例部署时，配额需要换成共享计数器，但调度算法本身不变。
// file: src/perf/fair-scheduling.ts
// tier: L1
// verify: node --test test/perf/scheduling.test.ts
import { AiError } from "../core/errors.ts";
export interface TenantPolicy {
tenantId: string;
/** 相对权重，正整数。权重 8:2:1 表示长期服务量按此比例分配。 */
weight: number;
/** 突发额度上限，单位微元；允许短时间超过持续速率。 */
burstMicros: number;
/** 持续补充速率，单位微元/毫秒。 */
refillMicrosPerMs: number;
}
export interface FairRequest {
id: string;
tenantId: string;
enqueuedAtMs: number;
serviceMs: number;
costMicros: number;
}
interface TenantState {
policy: TenantPolicy;
tokens: number;
lastRefillMs: number;
virtualFinish: number;
pending: number;
}
interface FairEntry {
request: FairRequest;
finish: number;
sequence: number;
}
function requireCount(value: number, name: string): number {
if (!Number.isInteger(value) || value < 0) {
throw new AiError("invalid-request", `fair-scheduling: ${name} 必须是非负整数`);
}
return value;
}
/**
* 令牌桶：burst 决定能扛多长的突发，refill 决定长期可持续速率。
* 只有 burst 没有 refill，突发额度用完就永久拒绝；只有 refill 没有 burst，
* 任何一个正常峰值都会被误杀。
*/
export class TokenBucket {
private readonly burst: number;
private readonly refillPerMs: number;
private tokens: number;
private lastRefillMs: number;
constructor(burstMicros: number, refillMicrosPerMs: number, startMs = 0) {
if (!Number.isFinite(burstMicros) || burstMicros < 0) {
throw new AiError("invalid-request", "fair-scheduling: burstMicros 必须是非负有限数");
}
if (!Number.isFinite(refillMicrosPerMs) || refillMicrosPerMs < 0) {
throw new AiError("invalid-request", "fair-scheduling: refillMicrosPerMs 必须是非负有限数");
}
this.burst = burstMicros;
this.refillPerMs = refillMicrosPerMs;
this.tokens = burstMicros;
this.lastRefillMs = startMs;
}
available(nowMs: number): number {
if (nowMs > this.lastRefillMs) {
const elapsed = nowMs - this.lastRefillMs;
this.tokens = Math.min(this.burst, this.tokens + elapsed * this.refillPerMs);
this.lastRefillMs = nowMs;
}
return this.tokens;
}
tryConsume(costMicros: number, nowMs: number): boolean {
requireCount(costMicros, "costMicros");
const available = this.available(nowMs);
if (costMicros > available) return false;
this.tokens = available - costMicros;
return true;
}
/** 补满到目标成本还差多少毫秒；没有补充能力时返回 Infinity。 */
retryAfterMs(costMicros: number, nowMs: number): number {
const deficit = costMicros - this.available(nowMs);
if (deficit <= 0) return 0;
if (this.refillPerMs <= 0) return Number.POSITIVE_INFINITY;
return Math.ceil(deficit / this.refillPerMs);
}
}
export interface FairSubmitResult {
ok: boolean;
reason: "ok" | "unknown-tenant" | "quota-exhausted";
retryAfterMs: number;
remainingMicros: number;
}
export interface FairSnapshot {
pending: number;
virtualTime: number;
tenants: Record<string, { pending: number; tokens: number; virtualFinish: number }>;
}
/**
* 加权公平队列（WFQ）。每个租户维护一个虚拟完成时间：
* finish = max(该租户上次 finish, 全局虚拟时间) + 成本 / 权重。
* 每次取 finish 最小者，长期服务量就收敛到权重比例。
*
* 饥饿防护是显式覆盖：等待超过 maxWaitMs 的请求优先于 WFQ 顺序。
* 它牺牲一点点比例公平，换取低权重租户有界的最坏等待。
*/
export class WeightedFairQueue {
private readonly tenants = new Map<string, TenantState>();
private readonly entries: FairEntry[] = [];
private readonly maxWaitMs: number;
private sequence = 0;
private virtualTime = 0;
private guardPromotions = 0;
constructor(maxWaitMs = Number.POSITIVE_INFINITY) {
if (!(maxWaitMs > 0)) {
throw new AiError("invalid-request", "fair-scheduling: maxWaitMs 必须大于 0");
}
this.maxWaitMs = maxWaitMs;
}
register(policy: TenantPolicy): void {
if (policy.tenantId.trim() === "") {
throw new AiError("invalid-request", "fair-scheduling: tenantId 不能为空");
}
if (!Number.isInteger(policy.weight) || policy.weight <= 0) {
throw new AiError("invalid-request", "fair-scheduling: weight 必须是正整数");
}
const existing = this.tenants.get(policy.tenantId);
if (existing !== undefined) {
existing.policy = { ...policy };
return;
}
this.tenants.set(policy.tenantId, {
policy: { ...policy },
tokens: policy.burstMicros,
lastRefillMs: 0,
virtualFinish: 0,
pending: 0,
});
}
get depth(): number {
return this.entries.length;
}
get promotions(): number {
return this.guardPromotions;
}
/** 先扣配额，再进入 WFQ；配额不足时连队列都不进。 */
submit(request: FairRequest, nowMs: number): FairSubmitResult {
const state = this.tenants.get(request.tenantId);
if (state === undefined) {
return { ok: false, reason: "unknown-tenant", retryAfterMs: 0, remainingMicros: 0 };
}
requireCount(request.costMicros, "costMicros");
if (nowMs > state.lastRefillMs) {
const elapsed = nowMs - state.lastRefillMs;
state.tokens = Math.min(
state.policy.burstMicros,
state.tokens + elapsed * state.policy.refillMicrosPerMs,
);
state.lastRefillMs = nowMs;
}
if (state.tokens < request.costMicros) {
const deficit = request.costMicros - state.tokens;
return {
ok: false,
reason: "quota-exhausted",
retryAfterMs:
state.policy.refillMicrosPerMs <= 0
? Number.POSITIVE_INFINITY
: Math.ceil(deficit / state.policy.refillMicrosPerMs),
remainingMicros: state.tokens,
};
}
state.tokens -= request.costMicros;
const start = Math.max(state.virtualFinish, this.virtualTime);
const costUnits = Math.max(1, request.costMicros);
const finish = start + costUnits / state.policy.weight;
state.virtualFinish = finish;
state.pending += 1;
this.entries.push({ request: { ...request }, finish, sequence: this.sequence });
this.sequence += 1;
return { ok: true, reason: "ok", retryAfterMs: 0, remainingMicros: state.tokens };
}
dequeue(nowMs: number): FairRequest | undefined {
if (this.entries.length === 0) return undefined;
let bestIndex = 0;
const overdue = this.entries.findIndex((entry) => nowMs - entry.request.enqueuedAtMs >= this.maxWaitMs);
if (overdue >= 0) {
this.guardPromotions += 1;
// 超时请求之间不能按到达先后挑，否则权重高的租户永远有更老的请求挡在前面。
// 按“等待时长 / 权重”排序，被相对亏待最久的租户先走。
const score = (entry: FairEntry): number => {
const weight = this.tenants.get(entry.request.tenantId)?.policy.weight ?? 1;
return (nowMs - entry.request.enqueuedAtMs) / weight;
};
bestIndex = overdue;
let bestScore = score(this.entries[overdue]);
for (let index = overdue + 1; index < this.entries.length; index += 1) {
const candidate = this.entries[index];
if (nowMs - candidate.request.enqueuedAtMs < this.maxWaitMs) continue;
const candidateScore = score(candidate);
if (
candidateScore > bestScore ||
(candidateScore === bestScore && candidate.sequence < this.entries[bestIndex].sequence)
) {
bestScore = candidateScore;
bestIndex = index;
}
}
} else {
for (let index = 1; index < this.entries.length; index += 1) {
const candidate = this.entries[index];
const current = this.entries[bestIndex];
if (
candidate.finish < current.finish ||
(candidate.finish === current.finish && candidate.sequence < current.sequence)
) {
bestIndex = index;
}
}
}
const [entry] = this.entries.splice(bestIndex, 1);
const state = this.tenants.get(entry.request.tenantId);
if (state !== undefined) {
state.pending = Math.max(0, state.pending - 1);
state.virtualFinish = Math.max(state.virtualFinish, entry.finish);
}
this.virtualTime = Math.max(this.virtualTime, entry.finish);
return entry.request;
}
snapshot(nowMs: number): FairSnapshot {
const tenants: FairSnapshot["tenants"] = {};
for (const [tenantId, state] of this.tenants) {
if (nowMs > state.lastRefillMs) {
const elapsed = nowMs - state.lastRefillMs;
state.tokens = Math.min(
state.policy.burstMicros,
state.tokens + elapsed * state.policy.refillMicrosPerMs,
);
state.lastRefillMs = nowMs;
}
tenants[tenantId] = {
pending: state.pending,
tokens: state.tokens,
virtualFinish: state.virtualFinish,
};
}
return { pending: this.entries.length, virtualTime: this.virtualTime, tenants };
}
}
WeightedFairQueue.submit 先扣配额再入队，配额不足时连队列都不进。dequeue 先检查是否有请求超过 maxWaitMs，有则按「等待/权重」选择，否则按 finish 最小选择。两个分支都保持确定性，测试可以逐条复现。
测试与运行结果
测试覆盖 WFQ 份额、饥饿防护、令牌桶突发与补充、未知租户与配额拒绝。执行：
node --test test/perf/scheduling.test.ts
本机一次运行的真实输出摘要如下，耗时随机器变化：
ℹ tests 14
ℹ pass 14
ℹ fail 0
公平性样本有三个租户，权重 8:2:1；A 每 10ms 到达、B 每 30ms、C 每 150ms，单请求服务 10ms，单槽位容量 100 请求/秒。统计窗口取前 3000ms，避免尾部排空把份额拉平。结果如下，均为示例样本上的运行结果：
| 配置 | A 窗口份额 | B 窗口份额 | C 窗口份额 | B 最坏等待 | A 最坏等待 | 防护提升次数 |
|---|---|---|---|---|---|---|
| 无防护 | 74.67% | 18.67% | 6.67% | 1720ms | 960ms | 0 |
| 200ms 防护 | 63.33% | 30.33% | 6.33% | 310ms | 1210ms | 363 |
无防护时 A 与 B 的窗口服务量是 224 与 56，比值正好 4.00，对应 8:2 的权重。C 的份额低于 9.1%，是因为它在窗口内只提交了 20 条请求，需求本身不足，不是被挤占。
开启 200ms 防护后，B 的最坏等待从 1720ms 降到 310ms，代价是 B 的份额从 18.67% 升到 30.33%，A 从 74.67% 降到 63.33%，A 与 B 的比值降到约 2.09。总平均等待仍是 615.7ms，没有变化。这再次说明防护只重新分配等待，不增加容量。
配额的行为可以直接用小样本验证：突发额度 3000 微元、补充速率 10 微元/毫秒时，连续三条 1000 微元的请求被接受，第四条被拒绝并返回 100ms 后重试；等待 100ms 后，同样的请求被接受。
小结
多租户调度要同时回答三个问题：按什么比例分、最少给多少、最多能用多少。WFQ 用虚拟完成时间实现比例公平；饥饿防护用有界覆盖换取低权重租户的最坏等待；令牌桶把长期配额与短时突发分开。三者都只改变「谁先走」，不改变总容量。
下一章从调度转向上下文：同样一次调用，输入越长，成本与延迟如何变化，哪些 token 可以省，哪些省了会伤质量。
阅读导航：上一节：2.2 并发上限与背压 · 下一节：3.1 上下文窗口的成本结构 。
常见错误与边界
- 用 FIFO 服务多租户。请求多的租户自然占满队列，小租户被饿死。
- 认为轮询就是公平。轮询忽略权重，也忽略请求成本差异。
- WFQ 不设虚拟时间下限。空闲租户攒信用，回来后一次性抢占。
- 按请求数而不是成本计权。长请求会让实际资源分配偏离权重。
- 饥饿防护阈值设得过低。比例公平被破坏，高权重租户的最坏等待上升。
- 以为防护能降低总等待。它只重新分配等待，不增加容量。
- 配额只有突发没有补充。额度用完后的请求永久失败。
- 配额只有补充没有突发。正常峰值会被误判为超额。
- 超额请求仍进入队列。队列位置被占，公平调度无从谈起。
- 单实例内存配额直接上多实例。每个实例各算各的，实际配额被放大数倍。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。