任务队列的中断和恢复

在 Web 前端开发中，我们经常需要处理一系列异步任务：批量上传图片、逐条导入数据、分片处理大文件、后台轮询接口……这些场景都涉及任务队列的管理。而其中最难处理的问题之一就是：如何优雅地中断一个正在执行的任务队列，并在合适的时机恢复它？

一、为什么需要中断和恢复？

先看几个真实场景：

场景	中断原因	恢复条件
批量图片上传	用户点击"取消"	重新点击"开始"
数据导入	页面切换/网络断开	回到页面/网络恢复
后台计算	CPU 占用过高	用户主动要求继续
轮询请求	组件卸载	组件重新挂载

如果没有中断和恢复机制，只能粗暴地 clearTimeout / ignore promise，但这样会丢失进度和中间状态。

二、基础版：使用标志位控制

最简单的中断方式是用一个布尔标志位。

class TaskQueue {
  constructor() {
    this._running = false;
    this._paused = false;
    this._tasks = [];
  }

  add(task) {
    this._tasks.push(task);
  }

  async run() {
    this._running = true;
    this._paused = false;
    while (this._tasks.length > 0 && this._running) {
      if (this._paused) {
        // 暂停等待
        await new Promise(resolve => {
          this._resume = resolve;
        });
      }
      const task = this._tasks.shift();
      await task();
    }
  }

  pause() {
    this._paused = true;
  }

  resume() {
    this._paused = false;
    this._resume?.();
  }

  cancel() {
    this._running = false;
    this._resume?.();
    this._tasks = [];
  }
}

使用示例：

const queue = new TaskQueue();

queue.add(() => fetch('/api/step/1'));
queue.add(() => fetch('/api/step/2'));
queue.add(() => fetch('/api/step/3'));

queue.run();

// 5 秒后暂停
setTimeout(() => queue.pause(), 5000);
// 10 秒后继续
setTimeout(() => queue.resume(), 10000);

优点：实现简单，零依赖。
缺点：无法取消正在执行中的单个任务，退出时需要等待当前 task 完成。

三、升级版：用 AbortController 实现取消

AbortController 是现代 JavaScript 中标准的中断方案，可以让正在进行的异步操作响应取消信号。

class AbortableTaskQueue {
  constructor() {
    this._tasks = [];
    this._abortController = null;
    this._currentTask = null;
  }

  add(task) {
    this._tasks.push(task);
  }

  async run() {
    this._abortController = new AbortController();
    const signal = this._abortController.signal;

    for (const task of this._tasks) {
      if (signal.aborted) break;
      this._currentTask = task;
      try {
        await task(signal);
      } catch (err) {
        if (err.name === 'AbortError') {
          console.log('任务被取消');
          break;
        }
        throw err;
      }
    }
    this._currentTask = null;
  }

  cancel() {
    this._abortController?.abort();
    this._tasks = [];
  }

  cancelCurrent() {
    this._abortController?.abort();
    this._currentTask = null;
    // 重新创建 AbortController，剩余任务可以继续
    this._abortController = new AbortController();
  }
}

配合支持 signal 的 API 使用效果最佳：

const queue = new AbortableTaskQueue();

queue.add(async (signal) => {
  const resp = await fetch('/api/big-file', { signal });
  return resp.json();
});

queue.add(async (signal) => {
  // 手动检查中断信号
  for (let i = 0; i < 1000; i++) {
    if (signal.aborted) throw new DOMException('Aborted', 'AbortError');
    await processChunk(i);
  }
});

queue.run();

// 随时取消
setTimeout(() => queue.cancel(), 3000);

fetch、axios、fs.readFile（Deno/Node）等主流异步 API 都原生支持 AbortSignal。

四、带进度的可恢复队列

中断后恢复的核心难点是保存进度。我们需要记录每个任务的状态和已完成的部分。

class ResumableTaskQueue {
  constructor() {
    this._tasks = new Map();   // id -> task descriptor
    this._order = [];
    this._abortController = null;
    this._state = 'idle';     // idle | running | paused
    this._progress = new Map(); // id -> progress
  }

  add(id, task) {
    if (!this._tasks.has(id)) {
      this._order.push(id);
    }
    this._tasks.set(id, {
      run: task,
      status: 'pending',   // pending | running | done | failed
    });
  }

  async run() {
    if (this._state === 'running') return;
    this._state = 'running';
    this._abortController = new AbortController();
    const signal = this._abortController.signal;

    for (const id of this._order) {
      const desc = this._tasks.get(id);
      if (!desc || desc.status === 'done') continue;
      if (signal.aborted) break;

      desc.status = 'running';
      try {
        await desc.run({
          signal,
          progress: this._progress.get(id) ?? 0,
          onProgress: (p) => this._progress.set(id, p),
        });
        desc.status = 'done';
      } catch (err) {
        if (err.name === 'AbortError') {
          desc.status = 'pending'; // 恢复后可重试
          break;
        }
        desc.status = 'failed';
      }
    }

    if (this._state !== 'cancelled') {
      this._state = 'idle';
    }
  }

  pause() {
    this._state = 'paused';
    this._abortController?.abort();
  }

  resume() {
    if (this._state === 'paused') {
      this._state = 'idle';
      this.run();
    }
  }

  cancel() {
    this._state = 'cancelled';
    this._abortController?.abort();
    this._progress.clear();
  }

  getProgress() {
    const total = this._order.length;
    const done = this._order.filter(id => this._tasks.get(id)?.status === 'done').length;
    return { total, done, percent: total > 0 ? Math.round(done / total * 100) : 0 };
  }
}

使用示例——分片上传场景：

const uploadQueue = new ResumableTaskQueue();

['file1.zip', 'file2.zip', 'file3.zip'].forEach((name, i) => {
  uploadQueue.add(`chunk-${i}`, async ({ signal, progress, onProgress }) => {
    const totalSize = 1024 * 1024; // 1MB
    let uploaded = progress;

    while (uploaded < totalSize) {
      if (signal.aborted) throw new DOMException('Aborted', 'AbortError');

      const chunk = Math.min(64 * 1024, totalSize - uploaded);
      await fetch('/api/upload', {
        method: 'POST',
        body: JSON.stringify({ name, offset: uploaded, size: chunk }),
        signal,
      });
      uploaded += chunk;
      onProgress(uploaded / totalSize);
    }
  });
});

await uploadQueue.run();

// 用户暂停
uploadQueue.pause();
// 保存进度到 localStorage
localStorage.setItem('upload-progress', JSON.stringify({
  tasks: [...uploadQueue._progress.entries()],
}));

// 恢复时
uploadQueue.resume();

五、用 Generator 实现优雅的中断点

Generator 函数天然支持"暂停-恢复"，非常适合任务队列场景。

function* taskSequence() {
  yield* processFiles();
  yield* validateData();
  yield* uploadResults();
}

function* processFiles() {
  for (const file of files) {
    // 处理每个文件后都可以中断
    yield { type: 'progress', file: file.name };
    const data = readFile(file);
    yield { type: 'processed', file: file.name, data };
  }
}

配合一个通用的执行器：

class GeneratorTaskRunner {
  constructor(gen) {
    this._gen = gen;
    this._paused = false;
    this._done = false;
  }

  async run(onYield) {
    const iterator = this._gen();
    let result = iterator.next();

    while (!result.done && !this._done) {
      if (this._paused) {
        await new Promise(r => { this._resume = r; });
      }
      if (onYield) onYield(result.value);
      result = iterator.next();
    }
  }

  pause() {
    this._paused = true;
  }

  resume() {
    this._paused = false;
    this._resume?.();
  }

  cancel() {
    this._done = true;
    this._resume?.();
  }
}

这种模式的好处是中断点是显式的——你在 yield 处自然暂停，任务状态完全由 Generator 的调用栈保存，恢复时从 yield 后面继续执行。

六、在 Web Worker 中中断和恢复

对于 CPU 密集型任务，Web Worker 是必经之路。中断和恢复通过 postMessage 传递控制信号。

worker.js：

let paused = false;
let cancelled = false;

self.onmessage = (e) => {
  const { type, data } = e.data;

  if (type === 'start') {
    cancelled = false;
    paused = false;
    runTasks(data.tasks);
  } else if (type === 'pause') {
    paused = true;
  } else if (type === 'resume') {
    paused = false;
    self.postMessage({ type: 'resumed' });
  } else if (type === 'cancel') {
    cancelled = true;
    paused = false;
  }
};

async function runTasks(tasks) {
  for (let i = 0; i < tasks.length; i++) {
    if (cancelled) {
      self.postMessage({ type: 'cancelled' });
      return;
    }

    while (paused) {
      await new Promise(r => setTimeout(r, 100));
    }

    const result = heavyComputation(tasks[i]);
    self.postMessage({
      type: 'progress',
      index: i,
      total: tasks.length,
      result,
    });
  }

  self.postMessage({ type: 'done' });
}

主线程：

const worker = new Worker('worker.js');

worker.postMessage({
  type: 'start',
  tasks: [data1, data2, data3],
});

// 暂停
worker.postMessage({ type: 'pause' });

// 恢复
worker.postMessage({ type: 'resume' });

// 取消
worker.postMessage({ type: 'cancel' });

这种方式的好处是 Worker 线程的计算完全不影响主线程响应，用户交互不会被阻塞。

七、React 中的任务队列恢复

React 组件卸载时自动中断，重新挂载时恢复，是 Hooks 的经典用例。

function useTaskQueue() {
  const queueRef = useRef(new ResumableTaskQueue());

  useEffect(() => {
    return () => {
      queueRef.current.pause();
    };
  }, []);

  const run = useCallback(async (tasks) => {
    tasks.forEach((t, i) => queueRef.current.add(i, t));
    await queueRef.current.run();
  }, []);

  return {
    run,
    pause: () => queueRef.current.pause(),
    resume: () => queueRef.current.resume(),
    cancel: () => queueRef.current.cancel(),
    progress: () => queueRef.current.getProgress(),
  };
}

使用：

function UploadPage() {
  const { run, pause, resume, progress } = useTaskQueue();
  const [pct, setPct] = useState(0);

  const handleUpload = () => {
    run(files.map(f => async (signal) => {
      await uploadFile(f, signal);
      setPct(progress().percent);
    }));
  };

  return (
    <div>
      <progress value={pct} max={100} />
      <button onClick={pause}>暂停</button>
      <button onClick={resume}>继续</button>
    </div>
  );
}

八、总结

方案	适用场景	中断粒度	可恢复性
标志位控制	简单队列	任务级	手动实现
AbortController	fetch / 回调式异步	任意点	需配合进度保存
Generator	纯计算 / 遍历	yield 处	天然恢复
Web Worker	CPU 密集型	循环检查点	消息控制
React Hooks	组件生命周期	自动卸载中断	配合 ref

核心设计原则：

中断点要显式——不要在深层回调里偷偷中断，容易遗漏清理
保存足够状态——恢复时最少能回到上一个已完成的任务
区分暂停和取消——暂停可恢复，取消要释放资源
AbortSignal 是第一选择——它是 Web 标准，原生 API 广泛支持

掌握这些模式后，再也不怕长任务阻塞页面了。无论是批量上传、数据导入还是复杂计算，都能做到想停就停，续上就走。

目录