高级并发与调度流控制

JavaScript 进阶

Site Owner

发布于 2026-05-21

深入讲解信号量与互斥锁实现、读写锁并发控制、优先级任务队列、背压机制及生产级调度器完整架构

高级并发与调度流控制

异步任务的并发边界控制；队列缓冲机制；动态任务挂起与自动出队；取消机制（AbortController 的集成）；带指数退避（Exponential Backoff）的自动重试请求器；可取消的 Promise 包装器。

并发控制的核心问题

无控制并发的危害

// ❌ 一次性启动全部请求（无并发控制）
const urls = Array.from({ length: 1000 }, (_, i) => `/api/item/${i}`);

// 同时发出 1000 个请求！
 results =  .(urls.( (url)));

同时运行中的任务数 ≤ concurrency 限制
            ↑
         任务队列（等待中）

class AsyncScheduler {
  #queue     = [];       // 待执行任务队列
  #running   = 0;        // 当前正在执行的任务数
  #concurrency;          // 最大并发数
  #paused    = false;    // 是否暂停
  
  // 统计信息
  #stats = { completed: 0, failed: 0, queued: 0 };

  constructor(concurrency = 5) {
    this.#concurrency = concurrency;
  }

  /**
   * 添加任务，返回 Promise（任务完成时 resolve/reject）
   * @param {Function} taskFn - 返回 Promise 的函数
   * @param {Object} [options]
   * @param {number} [options.priority=0] - 优先级（越大越优先）
   */
  add(taskFn, { priority = 0 } = {}) {
    return new Promise((resolve, reject) => {
      const task = { taskFn, resolve, reject, priority, addedAt: Date.now() };
      
      // 按优先级插入（优先级相同则按入队顺序）
      const insertIdx = this.#queue.findIndex(t => t.priority < priority);
      if (insertIdx === -1) {
        this.#queue.push(task);
      } else {
        this.#queue.splice(insertIdx, 0, task);
      }
      this.#stats.queued++;
      this.#dispatch();
    });
  }

  #dispatch() {
    if (this.#paused) return;
    
    while (this.#running < this.#concurrency && this.#queue.length > 0) {
      const task = this.#queue.shift();
      this.#stats.queued--;
      this.#run(task);
    }
  }

  async #run(task) {
    this.#running++;
    try {
      const result = await task.taskFn();
      task.resolve(result);
      this.#stats.completed++;
    } catch (err) {
      task.reject(err);
      this.#stats.failed++;
    } finally {
      this.#running--;
      this.#dispatch(); // 任务完成，尝试调度下一个
    }
  }

  // 暂停调度（正在运行的任务继续完成）
  pause()  { this.#paused = true; }
  resume() { this.#paused = false; this.#dispatch(); }

  // 清空待执行队列（正在运行的不受影响）
  clear() {
    const count = this.#queue.length;
    this.#queue.forEach(t => t.reject(new Error('队列已清空')));
    this.#queue = [];
    this.#stats.queued = 0;
    return count;
  }

  get stats() {
    return {
      ...this.#stats,
      running: this.#running,
      queued:  this.#queue.length,
    };
  }

  // 等待所有任务完成
  async drain() {
    if (this.#running === 0 && this.#queue.length === 0) return;
    await new Promise(resolve => {
      const check = () => {
        if (this.#running === 0 && this.#queue.length === 0) {
          resolve();
        } else {
          setTimeout(check, 10);
        }
      };
      check();
    });
  }
}

// 使用
const scheduler = new AsyncScheduler(3); // 最多同时 3 个

const urls = Array.from({ length: 20 }, (_, i) => `/api/item/${i}`);

const results = await Promise.allSettled(
  urls.map(url => scheduler.add(() => fetch(url).then(r => r.json())))
);

console.log(scheduler.stats);
// { completed: 18, failed: 2, running: 0, queued: 0 }

/**
 * 将数组分批处理，控制并发
 * @param {Array} items - 待处理数组
 * @param {Function} handler - 异步处理函数
 * @param {number} concurrency - 并发数
 */
async function batchProcess(items, handler, concurrency = 5) {
  const results = new Array(items.length);
  const errors  = [];
  let index = 0;

  async function worker() {
    while (index < items.length) {
      const i = index++;
      try {
        results[i] = await handler(items[i], i);
      } catch (e) {
        errors.push({ index: i, item: items[i], error: e });
        results[i] = null;
      }
    }
  }

  // 启动 concurrency 个 worker
  await Promise.all(
    Array.from({ length: Math.min(concurrency, items.length) }, worker)
  );

  return { results, errors };
}

// 使用
const { results, errors } = await batchProcess(
  Array.from({ length: 100 }, (_, i) => i),
  async (id) => {
    const resp = await fetch(`/api/users/${id}`);
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
    return resp.json();
  },
  10  // 10 个并发
);

console.log(`成功: ${results.filter(Boolean).length}, 失败: ${errors.length}`);

// 最小堆实现的优先级队列（priority 越小越先执行）
class PriorityQueue {
  #heap = [];
  #compare;

  constructor(compare = (a, b) => a.priority - b.priority) {
    this.#compare = compare;
  }

  push(item) {
    this.#heap.push(item);
    this.#siftUp(this.#heap.length - 1);
  }

  pop() {
    const top = this.#heap[0];
    const last = this.#heap.pop();
    if (this.#heap.length > 0) {
      this.#heap[0] = last;
      this.#siftDown(0);
    }
    return top;
  }

  peek() { return this.#heap[0]; }
  get size() { return this.#heap.length; }

  #siftUp(i) {
    while (i > 0) {
      const parent = (i - 1) >> 1;
      if (this.#compare(this.#heap[i], this.#heap[parent]) < 0) {
        [this.#heap[i], this.#heap[parent]] = [this.#heap[parent], this.#heap[i]];
        i = parent;
      } else break;
    }
  }

  #siftDown(i) {
    const n = this.#heap.length;
    while (true) {
      let smallest = i;
      const l = 2 * i + 1, r = 2 * i + 2;
      if (l < n && this.#compare(this.#heap[l], this.#heap[smallest]) < 0) smallest = l;
      if (r < n && this.#compare(this.#heap[r], this.#heap[smallest]) < 0) smallest = r;
      if (smallest !== i) {
        [this.#heap[i], this.#heap[smallest]] = [this.#heap[smallest], this.#heap[i]];
        i = smallest;
      } else break;
    }
  }
}

// React Scheduler 风格：基于截止时间的调度
class ReactStyleScheduler {
  static Priority = {
    IMMEDIATE:  1,  // 同步，立刻执行（如用户输入响应）
    USER_BLOCK: 2,  // 250ms 超时（如点击响应）
    NORMAL:     3,  // 5000ms 超时
    LOW:        4,  // 10000ms 超时
    IDLE:       5,  // 空闲时执行
  };

  static #timeouts = {
    1: -1,
    2: 250,
    3: 5000,
    4: 10000,
    5: Infinity,
  };

  #queue = new PriorityQueue((a, b) => a.expirationTime - b.expirationTime);

  schedule(callback, priority = ReactStyleScheduler.Priority.NORMAL) {
    const timeout = ReactStyleScheduler.#timeouts[priority];
    const expirationTime = performance.now() + timeout;

    const task = { callback, priority, expirationTime };
    this.#queue.push(task);
    this.#requestFlush();

    return task;
  }

  #flushRequested = false;
  #requestFlush() {
    if (this.#flushRequested) return;
    this.#flushRequested = true;
    MessageChannel && new MessageChannel().port1.postMessage(null);
    // 或 queueMicrotask / setTimeout(fn, 0)
  }

  flush(deadline = Infinity) {
    const start = performance.now();
    this.#flushRequested = false;

    while (this.#queue.size > 0) {
      const now = performance.now();
      const task = this.#queue.peek();

      // 超时或截止时间已到
      if (task.expirationTime <= now || now - start < deadline) {
        this.#queue.pop();
        task.callback();
      } else {
        // 时间片用完，让出主线程
        this.#requestFlush();
        break;
      }
    }
  }
}

const controller = new AbortController();
const { signal } = controller;

// fetch 集成
try {
  const response = await fetch('/api/data', { signal });
  const data = await response.json();
} catch (e) {
  if (e.name === 'AbortError') {
    console.log('请求被取消');
  } else {
    throw e; // 重新抛出非取消错误
  }
}

// 触发取消
controller.abort();             // 取消，reason 为 undefined
controller.abort(new Error('用户取消操作')); // 带原因取消

// ES2022：AbortSignal.timeout
const signal = AbortSignal.timeout(5000); // 5 秒超时

try {
  const resp = await fetch('/api/slow', { signal });
} catch (e) {
  if (e.name === 'TimeoutError') {
    console.log('请求超时');
  }
}

// 组合多个取消信号（ES2023：AbortSignal.any）
const userCancel  = new AbortController();
const timeout     = AbortSignal.timeout(5000);
const combinedSignal = AbortSignal.any([userCancel.signal, timeout]);
// 任一信号中止 → combinedSignal 中止

class CancellableTask {
  #controller = new AbortController();

  get signal() { return this.#controller.signal; }

  cancel(reason) {
    this.#controller.abort(reason);
  }

  async run(asyncFn) {
    // 监听取消信号
    if (this.#controller.signal.aborted) {
      throw this.#controller.signal.reason ?? new Error('已取消');
    }

    return new Promise((resolve, reject) => {
      // 信号中止时 reject
      this.#controller.signal.addEventListener('abort', () => {
        reject(this.#controller.signal.reason ?? new Error('操作被取消'));
      }, { once: true });

      asyncFn(this.#controller.signal).then(resolve).catch(reject);
    });
  }
}

// 使用
const task = new CancellableTask();

const resultPromise = task.run(async (signal) => {
  const resp = await fetch('/api/data', { signal });
  // 手动检查取消（对于纯计算任务）
  if (signal.aborted) throw new Error('计算被取消');
  return resp.json();
});

// 2 秒后取消
setTimeout(() => task.cancel(new Error('用户点击取消')), 2000);

try {
  const result = await resultPromise;
} catch (e) {
  console.log('任务失败：', e.message);
}

/**
 * 带指数退避的自动重试请求器
 */
class RetryFetcher {
  #options;

  constructor({
    maxRetries    = 3,
    baseDelay     = 1000,   // 初始延迟 1s
    maxDelay      = 30000,  // 最大延迟 30s
    factor        = 2,      // 退避因子（每次乘以 2）
    jitter        = true,   // 是否添加随机抖动（防止雷同重试）
    retryOn       = (e) => true, // 哪些错误需要重试
  } = {}) {
    this.#options = { maxRetries, baseDelay, maxDelay, factor, jitter, retryOn };
  }

  async fetch(url, fetchOptions = {}) {
    const { maxRetries, baseDelay, maxDelay, factor, jitter, retryOn } = this.#options;
    let attempt = 0;
    let lastError;

    while (attempt <= maxRetries) {
      if (fetchOptions.signal?.aborted) {
        throw fetchOptions.signal.reason ?? new Error('请求被取消');
      }

      try {
        const response = await fetch(url, fetchOptions);
        
        // HTTP 错误是否重试
        if (!response.ok) {
          // 4xx 客户端错误不重试（除 429 Too Many Requests）
          if (response.status < 500 && response.status !== 429) {
            throw new Error(`HTTP ${response.status}`);
          }
          throw new Error(`HTTP ${response.status} (重试中)`);
        }
        
        return response;
      } catch (e) {
        lastError = e;

        if (attempt >= maxRetries || !retryOn(e)) {
          break;
        }

        // 计算退避延迟
        const delay = this.#calcDelay(attempt, { baseDelay, maxDelay, factor, jitter });
        console.log(`第 ${attempt + 1} 次重试，等待 ${delay}ms...`);
        
        await this.#sleep(delay, fetchOptions.signal);
        attempt++;
      }
    }

    throw lastError;
  }

  #calcDelay(attempt, { baseDelay, maxDelay, factor, jitter }) {
    // 指数退避：baseDelay * factor^attempt
    let delay = Math.min(baseDelay * Math.pow(factor, attempt), maxDelay);
    
    // 添加随机抖动：± 25%
    if (jitter) {
      delay = delay * (0.75 + Math.random() * 0.5);
    }
    
    return Math.floor(delay);
  }

  #sleep(ms, signal) {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) return reject(signal.reason);
      
      const id = setTimeout(resolve, ms);
      signal?.addEventListener('abort', () => {
        clearTimeout(id);
        reject(signal.reason ?? new Error('等待被取消'));
      }, { once: true });
    });
  }
}

// 使用
const fetcher = new RetryFetcher({
  maxRetries: 3,
  baseDelay:  500,
  retryOn: (e) => !e.message.startsWith('HTTP 4'), // 4xx 不重试
});

const controller = new AbortController();
const resp = await fetcher.fetch('/api/data', { signal: controller.signal });

/**
 * 可取消的 Promise 包装器
 * 支持 .cancel() 方法中止 Promise
 */
function createCancellablePromise(executor) {
  let cancel;
  
  const innerPromise = new Promise(executor);
  
  const cancellablePromise = new Promise((resolve, reject) => {
    cancel = (reason = new Error('Promise 被取消')) => {
      reject(reason);
    };
    innerPromise.then(resolve, reject);
  });
  
  cancellablePromise.cancel = cancel;
  return cancellablePromise;
}

// 通用 Promise 取消包装
function makeCancellable(promise) {
  let isCancelled = false;
  
  const wrappedPromise = new Promise((resolve, reject) => {
    promise.then(
      value  => isCancelled ? reject({ isCancelled: true }) : resolve(value),
      error  => isCancelled ? reject({ isCancelled: true }) : reject(error)
    );
  });
  
  return {
    promise: wrappedPromise,
    cancel() { isCancelled = true; },
  };
}

// React 风格：在 useEffect 中防止"在卸载后更新状态"
function fetchData(id, signal) {
  const { promise, cancel } = makeCancellable(
    fetch(`/api/data/${id}`, { signal }).then(r => r.json())
  );
  
  promise.then(data => {
    // 组件卸载后这里不会执行
    setData(data);
  }).catch(e => {
    if (!e.isCancelled) setError(e);
  });
  
  return cancel; // useEffect 返回清理函数
}

/**
 * 完整的异步调度系统
 * 支持：并发控制 + 优先级 + 取消 + 重试 + 进度上报
 */
class FullScheduler {
  #concurrency;
  #running   = new Set();
  #queue     = []; // 按优先级排序
  #abortMap  = new Map(); // taskId → AbortController
  #taskId    = 0;

  constructor(concurrency = 5) {
    this.#concurrency = concurrency;
  }

  submit(taskFn, {
    priority   = 0,
    maxRetries = 0,
    timeout    = 0,
    onProgress = null,
  } = {}) {
    const id = this.#taskId++;
    const outerController = new AbortController();
    this.#abortMap.set(id, outerController);

    const promise = new Promise((resolve, reject) => {
      const task = {
        id,
        priority,
        retries: 0,
        maxRetries,
        timeout,
        onProgress,
        execute: async () => {
          // 组合超时 signal
          const signals = [outerController.signal];
          if (timeout > 0) signals.push(AbortSignal.timeout(timeout));
          const signal = AbortSignal.any(signals);

          let attempt = 0;
          while (true) {
            try {
              const result = await taskFn(signal, (progress) => onProgress?.(id, progress));
              resolve(result);
              break;
            } catch (e) {
              if (e.name === 'AbortError' || e.name === 'TimeoutError') {
                reject(e);
                break;
              }
              if (attempt >= maxRetries) {
                reject(e);
                break;
              }
              const delay = 500 * Math.pow(2, attempt++) * (0.5 + Math.random() * 0.5);
              await new Promise(r => setTimeout(r, delay));
            }
          }
          this.#abortMap.delete(id);
        },
      };

      // 按优先级插入队列
      const idx = this.#queue.findIndex(t => t.priority < priority);
      this.#queue.splice(idx === -1 ? this.#queue.length : idx, 0, task);
      this.#tick();
    });

    promise.taskId = id;
    promise.cancel = (reason) => this.cancel(id, reason);
    return promise;
  }

  cancel(taskId, reason = new Error('任务被取消')) {
    const ctrl = this.#abortMap.get(taskId);
    if (ctrl) {
      ctrl.abort(reason);
      this.#abortMap.delete(taskId);
    }
    // 从队列中移除（如果还未执行）
    this.#queue = this.#queue.filter(t => t.id !== taskId);
  }

  cancelAll() {
    for (const [id] of this.#abortMap) this.cancel(id);
  }

  #tick() {
    while (this.#running.size < this.#concurrency && this.#queue.length > 0) {
      const task = this.#queue.shift();
      this.#running.add(task.id);
      task.execute().finally(() => {
        this.#running.delete(task.id);
        this.#tick();
      });
    }
  }

  get status() {
    return {
      running: this.#running.size,
      queued:  this.#queue.length,
    };
  }
}

// 完整使用示例
const scheduler = new FullScheduler(3);

// 提交 10 个任务，2 个高优先级
const tasks = Array.from({ length: 10 }, (_, i) => {
  const isHigh = i < 2;
  return scheduler.submit(
    async (signal, reportProgress) => {
      for (let step = 0; step < 5; step++) {
        if (signal.aborted) throw new Error('cancelled');
        await new Promise(r => setTimeout(r, 100));
        reportProgress((step + 1) / 5);
      }
      return `任务 ${i} 完成`;
    },
    {
      priority:   isHigh ? 10 : 0,
      maxRetries: 2,
      timeout:    3000,
      onProgress: (id, p) => console.log(`任务 ${i} 进度: ${(p * 100).toFixed(0)}%`),
    }
  );
});

// 2 秒后取消部分任务
setTimeout(() => {
  tasks[5].cancel(new Error('用户取消'));
  tasks[6].cancel(new Error('用户取消'));
}, 2000);

const results = await Promise.allSettled(tasks);
results.forEach((r, i) => {
  if (r.status === 'fulfilled') console.log(r.value);
  else console.log(`任务 ${i} 失败：`, r.reason.message);
});

高级并发与调度流控制

高级并发与调度流控制

目录

并发控制的核心问题

无控制并发的危害

并发控制的目标

限速并发调度器（Scheduler）

核心实现

批量处理工具函数

任务队列与优先级调度

优先级堆队列

取消机制（AbortController）

AbortController 基础

超时自动取消

链式取消传播

指数退避重试器

可取消 Promise

手写验证：完整调度系统

深度追问