基于队列的负载整形

概念定义

被限速或容量脆弱的服务在突发下崩溃。队列是减震器：到达模式对服务不再重要，它看到的是自己能持续应付的平滑排空速率。

工作机制

生产方入队即返回；固定或自动扩缩的消费者池按配置并发排空。队列深度成为首要健康指标——增长能提前数分钟预测饱和。死信队列收容毒消息，单条坏数据不再堵死排水。

使用场景

带配额的第三方 API 调用、邮件与通知发送、以及价值能存活于异步的写路径。交互式流程除非业务接受最终完成，否则保持同步。

常见误区

队列不带来速度带来存活性：每个请求排队等轮，延迟等于队列深度。从用户请求同步入队却没有完成模型，用户就失去反馈；无界队列把过载变成无界延迟加内存——上限与拒绝是设计的一部分。消费者应按可持续速率而非全速排空，比赛式排空会重建队列要防止的过载；毒消息会堵死严格 FIFO——设最大投递数与死信策略，消费侧重试要退避加抖动。

代码示例

export interface Job<T> {
  id: string;
  payload: T;
  attempts: number;
}

export interface DeadLetterEntry<T> {
  job: Job<T>;
  error: string;
  failedAt: number;
}

export class RetryQueue<T> {
  private readonly pending: Job<T>[] = [];
  private readonly deadLetters: DeadLetterEntry<T>[] = [];
  private readonly maxAttempts: number;
  private readonly now: () => number;

  constructor(maxAttempts = 3, now: () => number = () => Date.now()) {
    if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
      throw new RangeError("maxAttempts must be a positive integer");
    }
    if (typeof now !== "function") throw new TypeError("now must be a function");
    this.maxAttempts = maxAttempts;
    this.now = now;
  }

  enqueue(id: string, payload: T): void {
    if (typeof id !== "string" || id.length === 0) {
      throw new TypeError("id must be a non-empty string");
    }
    this.pending.push({ id, payload, attempts: 0 });
  }

  async drain(
    handler: (payload: T, job: Job<T>) => Promise<void>,
  ): Promise<{ processed: number; deadLettered: number }> {
    if (typeof handler !== "function") throw new TypeError("handler must be a function");
    let processed = 0;
    let deadLettered = 0;
    while (this.pending.length > 0) {
      const job = this.pending.shift();
      if (!job) break;
      job.attempts += 1;
      try {
        await handler(job.payload, job);
        processed += 1;
      } catch (error) {
        if (job.attempts < this.maxAttempts) {
          this.pending.push(job);
        } else {
          this.deadLetters.push({
            job,
            error: error instanceof Error ? error.message : String(error),
            failedAt: this.now(),
          });
          deadLettered += 1;
        }
      }
    }
    return { processed, deadLettered };
  }

  get deadLetterCount(): number {
    return this.deadLetters.length;
  }

  listDeadLetters(): readonly DeadLetterEntry<T>[] {
    return [...this.deadLetters];
  }
}

export class SlidingWindowLimiter {
  constructor(limit, windowMs) {
    if (!Number.isInteger(limit) || limit <= 0) throw new RangeError('limit must be positive');
    if (!Number.isInteger(windowMs) || windowMs <= 0) throw new RangeError('windowMs must be positive');
    this.limit = limit;
    this.windowMs = windowMs;
    this.hits = new Map();
  }

  allow(key, now = Date.now()) {
    const cutoff = now - this.windowMs;
    const times = (this.hits.get(key) ?? []).filter((t) => t > cutoff);
    if (times.length >= this.limit) {
      this.hits.set(key, times);
      return { allowed: false, retryAfterMs: times[0] + this.windowMs - now };
    }
    times.push(now);
    this.hits.set(key, times);
    return { allowed: true, remaining: this.limit - times.length };
  }
}