batch-processing-patterns

# 批量处理与长时任务指南

来自生产级桌面应用的实战经验，覆盖批量文件处理、远程 API 轮询、并发控制和错误恢复。

## 适用场景

- 批量文件处理（转码、压缩、水印）
- 远程 AI 任务轮询（视频生成、语音合成）
- 爬虫/批量 HTTP 请求
- 后台队列任务

---

## 1. 批处理架构

```
任务队列
├── 并发调度器（动态调整并发数）
│   ├── Worker 1 → processItem()
│   ├── Worker 2 → processItem()
│   └── Worker N → processItem()
├── 中止控制器（shouldStop + 子进程清理）
├── 熔断器（连续失败 N 次暂停）
├── 跳过检查（断点续传 / 前置过滤）
└── 进度报告（per-item + overall）
```

### 核心原则

1. **逐项处理，逐项报告** — 不等全部完成
2. **中断即停** — 每个 item 之间检查 abort 信号
3. **失败不中断** — 单项失败标记 `failed`，继续处理其他
4. **熔断保护** — 连续失败超阈值暂停整个队列

---

## 2. 并发调度

### 自适应并发池

根据每个 item 的处理耗时动态调整并发数：

```typescript
class AdaptiveScheduler {
  private concurrency: number;
  private running = 0;
  private queue: (() => void)[] = [];

  constructor(
    private min: number,
    private max: number,
    private slowThresholdMs: number
  ) {
    this.concurrency = Math.ceil((min + max) / 2);
  }

  async run<T>(fn: () => Promise<T>): Promise<T> {
    if (this.running >= this.concurrency) {
      await new Promise<void>((resolve) => this.queue.push(resolve));
    }
    this.running++;
    const start = Date.now();

    try {
      return await fn();
    } finally {
      const elapsed = Date.now() - start;
      this.running--;

      // 自适应调整
      if (elapsed > this.slowThresholdMs && this.concurrency > this.min) {
        this.concurrency--;
      } else if (elapsed < this.slowThresholdMs / 2 && this.concurrency < this.max) {
        this.concurrency++;
      }

      if (this.queue.length > 0) {
        this.queue.shift()!();
      }
    }
  }
}
```

### 预设配置

| 场景 | 初始 | 最小 | 最大 | 慢阈值 |
|------|------|------|------|--------|
| CPU 密集（FFmpeg 转码） | 4 | 1 | CPU 核数 | 3s |
| API 调用（AI 服务） | 3 | 1 | 5 | 8s |
|...