批量生成不能简单地把所有请求交给 Promise.all，因为应用会在瞬间把压力传递给模型服务、连接池和自身内存。本文用一个可直接运行的 Node.js 调度器，把并发预算、有界队列、超时取消与租户公平性放进同一条请求链路。

Promise.all 为什么会放大故障

假设一个内容平台允许用户一次生成 100 条商品描述。最直接的实现通常是：

const results = await Promise.all(
  products.map(product => generateDescription(product))
);

Promise.all 负责聚合结果，但不负责限制任务启动速度。只要 generateDescription 在调用时立即发出网络请求，这 100 个任务就会几乎同时进入模型 SDK、HTTP 连接池或上游网关。

在多租户服务中，问题会进一步放大：一个租户提交大批量任务，可能占满全部连接；其他租户的少量交互请求只能排在后面。上游开始返回 429 后，若客户端立即重试，又会形成新的流量尖峰。

问题直接影响常见连锁反应
无并发上限同时请求数快速增长连接耗尽、429、超时
无队列上限待处理 Promise 持续堆积内存上涨、GC 压力增加
不区分模型慢模型占用全部执行槽快模型请求也被拖慢
不区分租户大批量租户长期占队小请求出现高尾延迟
只设置网络超时请求在队列中无限等待用户已离开，任务仍然执行

关键点不是彻底消除排队，而是让排队发生在可观察、可拒绝、可取消的应用层，而不是失控地转移到连接池和上游服务中。

把容量拆成四个控制面

一个实用的 LLM 请求调度器至少需要四类约束。

第一是分模型并发预算。不同模型的延迟、成本和上游配额可能不同，例如普通生成模型允许 3 个并发，推理模型只允许 1 个。预算应按实际部署、账号或区域继续细分，而不是假设所有模型共享同一容量。

第二是有界等待队列。当执行槽已满时，请求可以短暂等待；达到队列上限后必须快速失败。对 HTTP 服务，通常可以返回 503 并附带 Retry-After，也可以在明确表示租户配额不足时返回 429。具体状态码要与网关和客户端重试策略保持一致。

第三是覆盖排队时间的总超时。如果只在真正调用模型时启动超时，一个已经等待 20 秒的请求还会继续消耗上游资源。更合理的做法是在进入调度器时启动计时，并通过 AbortSignal 把取消传递给实际请求。

第四是租户公平调度。简单 FIFO 会让先提交 1000 个任务的租户挡住后来的交互请求。轮询租户队列可以让每次调度机会在活跃租户之间流转。它不能抢占已经运行的任务，但能避免单个租户长期垄断后续执行槽。

可运行的 Node.js 调度器

下面的示例只使用 Node.js 内置 API，可保存为 scheduler.mjs，然后使用 Node.js 18 或更高版本运行 node scheduler.mjs。示例中的 fakeLlm 模拟可取消的模型调用；接入真实 SDK 时，应把同一个 signal 传给该 SDK 支持的取消参数。

import { setTimeout as sleep } from 'node:timers/promises';

class BusyError extends Error {
  constructor(message = 'queue is full') {
    super(message);
    this.name = 'BusyError';
  }
}

class FairScheduler {
  constructor({ modelLimits, maxQueue, worker }) {
    this.limits = new Map(Object.entries(modelLimits));
    this.running = new Map(
      [...this.limits.keys()].map(model => [model, 0])
    );
    this.maxQueue = maxQueue;
    this.worker = worker;

    this.queues = new Map();
    this.tenantRing = [];
    this.activeTenants = new Set();
    this.queued = 0;
  }

  run(job, { timeoutMs = 5000 } = {}) {
    if (!this.limits.has(job.model)) {
      return Promise.reject(new Error(`unknown model: ${job.model}`));
    }
    if (this.queued >= this.maxQueue) {
      return Promise.reject(new BusyError());
    }

    const controller = new AbortController();
    const timer = setTimeout(() => {
      controller.abort(new Error('request deadline exceeded'));
    }, timeoutMs);

    const promise = new Promise((resolve, reject) => {
      const item = {
        job,
        signal: controller.signal,
        resolve,
        reject,
        state: 'queued',
        onAbort: null
      };

      let queue = this.queues.get(job.tenantId);
      if (!queue) {
        queue = [];
        this.queues.set(job.tenantId, queue);
      }
      queue.push(item);
      this.queued += 1;

      if (!this.activeTenants.has(job.tenantId)) {
        this.activeTenants.add(job.tenantId);
        this.tenantRing.push(job.tenantId);
      }

      item.onAbort = () => {
        if (item.state !== 'queued') return;
        const index = queue.indexOf(item);
        if (index !== -1) {
          queue.splice(index, 1);
          this.queued -= 1;
        }
        item.state = 'done';
        reject(item.signal.reason ?? new Error('aborted'));
        this.pump();
      };

      item.signal.addEventListener('abort', item.onAbort, { once: true });
      this.pump();
    });

    return promise.finally(() => clearTimeout(timer));
  }

  nextRunnable() {
    const turns = this.tenantRing.length;

    for (let i = 0; i < turns; i += 1) {
      const tenantId = this.tenantRing.shift();
      this.activeTenants.delete(tenantId);
      const queue = this.queues.get(tenantId) ?? [];

      const index = queue.findIndex(item => {
        const model = item.job.model;
        return this.running.get(model) < this.limits.get(model);
      });

      let item = null;
      if (index !== -1) {
        item = queue.splice(index, 1)[0];
        this.queued -= 1;
      }

      if (queue.length > 0) {
        this.tenantRing.push(tenantId);
        this.activeTenants.add(tenantId);
      } else {
        this.queues.delete(tenantId);
      }

      if (item) return item;
    }

    return null;
  }

  pump() {
    while (true) {
      const item = this.nextRunnable();
      if (!item) return;

      const model = item.job.model;
      item.state = 'running';
      this.running.set(model, this.running.get(model) + 1);

      Promise.resolve()
        .then(() => this.worker(item.job, item.signal))
        .then(item.resolve, item.reject)
        .finally(() => {
          item.state = 'done';
          item.signal.removeEventListener('abort', item.onAbort);
          this.running.set(model, this.running.get(model) - 1);
          this.pump();
        });
    }
  }
}

async function fakeLlm(job, signal) {
  const latency = 300 + Math.floor(Math.random() * 700);
  await sleep(latency, undefined, { signal });
  return `${job.model}: ${job.prompt}`;
}

const scheduler = new FairScheduler({
  modelLimits: {
    chat: 3,
    reasoning: 1
  },
  maxQueue: 8,
  worker: fakeLlm
});

const tasks = [];
for (let i = 0; i < 6; i += 1) {
  for (const tenantId of ['tenant-a', 'tenant-b', 'tenant-c']) {
    tasks.push(
      scheduler.run(
        {
          tenantId,
          model: i % 3 === 0 ? 'reasoning' : 'chat',
          prompt: `task-${i}`
        },
        { timeoutMs: 1800 }
      )
    );
  }
}

const results = await Promise.allSettled(tasks);
for (const result of results) {
  if (result.status === 'fulfilled') {
    console.log('OK ', result.value);
  } else {
    console.log('ERR', result.reason.name, result.reason.message);
  }
}

这里仍然使用了 Promise.allSettled，但它只负责等待结果；真正的启动节奏由调度器控制。队列满时，新任务立即得到 BusyError。超时从进入调度器开始计算：任务尚在队列时会被移除，已经运行时则由 AbortSignal 中断可取消的工作。

调度器会在租户之间轮询，并在某个模型没有容量时寻找该租户队列中其他可运行模型，因此它提供的是公平性优先的弱顺序，而不是严格 FIFO。如果业务要求同一会话严格有序，应单独为会话增加串行约束。

接入生产服务时还要补什么

调度器解决的是进程内负载控制，不等于完整的分布式配额系统。多实例部署时，每个进程都持有一份预算：如果 10 个实例各允许 3 个并发，上游实际可能看到 30 个并发。可以根据实例数静态切分预算，或使用集中式令牌服务；不要把进程内计数误认为全局上限。

HTTP 层还应把客户端断开连接转换为取消信号。需要注意，取消本地等待不一定代表上游已经停止计费，最终行为取决于模型提供方是否接受并及时处理取消。因此，超时既是资源保护手段，也是调用方停止等待的边界，不能承诺一定撤销远端计算。

建议至少记录以下指标：

  • 按模型统计运行中任务数、队列长度和执行耗时；
  • 按租户统计接收、完成、超时与队列拒绝数量；
  • 分开记录排队耗时和上游耗时，避免把两者混成一个延迟；
  • 记录 429、网络错误和取消，但不要在日志中写入完整提示词；
  • 为队列使用率设置告警，持续接近上限通常意味着容量不足或上游变慢。

重试也必须重新进入调度器，并使用指数退避和随机抖动。若在 worker 内无限重试，一个任务会长期占用执行槽；更稳妥的方式是限制尝试次数，同时确保总请求截止时间覆盖全部重试。

预算值不要凭感觉一次定死。可以从上游明确配额、单请求资源消耗和可接受排队时间出发设置保守初值，再根据真实指标逐步调整。队列越长并不代表吞吐越高，它通常只会容纳更多已经来不及完成的请求。

总结

把 LLM 原型改造成负载可控的服务，核心不是替换 Promise.all，而是明确每个请求何时可以执行、最多等待多久，以及系统过载时如何失败。

  • 用分模型并发预算隔离不同延迟和配额的模型；
  • 用有界队列限制内存占用，并在过载时快速拒绝；
  • 让超时覆盖排队与执行阶段，并向下游传递取消信号；
  • 用租户轮询减少大批量任务对其他租户的阻塞；
  • 分别观测排队、执行、拒绝和取消，再据此调整容量。

这些机制不会增加上游模型的真实吞吐，但能阻止流量尖峰被无条件放大，让服务在容量不足时以可预期、可恢复的方式退化。