cy-plugin-queue

串行任务队列。任务按 FIFO 依次执行，内置日志、重试、超时、中断能力。

特性

• 串行执行：同一时刻只跑一个任务
• 重试：支持全局 / 任务级重试次数，指数退避
• 超时：可选；未配置不限制；超时抛出 TimeoutError
• 中断：通过 AbortSignal 取消当前任务
• 回调：全局 + 任务级 onSuccess / onError，可携带 extraData
• 调试日志：debug: true 开启

安装

npm install cy-plugin-queue

快速开始

import Queue from 'cy-plugin-queue';
import type { IQueueOptions, ITaskOptions } from 'cy-plugin-queue';

const queue = new Queue({
  debug: true,
  retryCount: 1,
  timeout: 5000,
  onSuccess: (result, extra) => {
    console.log('全局成功:', result, extra);
  },
  onError: (err, extra) => {
    console.error('全局失败:', err, extra);
  }
});

queue.push({
  taskParams: { title: 'hello' },
  task: async (params) => {
    return `done: ${params?.title}`;
  },
  onSuccess: (res) => {
    console.log(res);
  }
});

API

new Queue(options?)

| 选项 | 类型 | 默认 | 说明 | |------|------|------|------| | debug | boolean | false | 是否打印调试日志 | | retryCount | number | 0 | 默认重试次数（不含首次执行） | | timeout | number | 未设置 | 默认超时（ms）；未设置则不限制 | | retryOnTimeout | boolean | false | 超时错误是否允许重试 | | onSuccess | (res, extraData?) => void | - | 全局成功回调 | | onError | (err, extraData?) => void | - | 全局失败回调 |

queue.push(task)

| 字段 | 类型 | 必填 | 说明 | |------|------|------|------| | task | (params?, ctx?) => any \| Promise<any> | 是 | 任务函数；ctx.signal 可用于监听中断 | | taskParams | Record<string, any> | 否 | 传给 task 的参数 | | extraData | any | 否 | 传给成功/失败回调的额外数据 | | retryCount | number | 否 | 覆盖队列默认重试次数 | | timeout | number | 否 | 覆盖队列默认超时（ms） | | id | string | 否 | 自定义任务 ID；不传则自动生成 | | onSuccess | (res, extraData?) => void | 否 | 任务成功回调 | | onError | (err, extraData?) => void | 否 | 任务失败回调 |

实例方法 / 属性

| API | 说明 | |-----|------| | push(task) | 入队并尝试执行 | | clear(triggerError?, abortCurrent?) | 清空等待队列；见下方示例 | | abort() | 中断当前正在执行的任务 | | length | 等待中的任务数量（不含正在执行的） | | isTaskExecuting | 是否有任务正在执行 |

导出类型 / 工具

import Queue, {
  TimeoutError,
  isTimeoutError
} from 'cy-plugin-queue';

import type {
  IQueueOptions,
  ITaskOptions,
  ILoggerOptions,
  IRetryOptions,
  ITimeoutOptions
} from 'cy-plugin-queue';

使用示例

1. 同步任务

import Queue from 'cy-plugin-queue';

const queue = new Queue();

queue.push({
  task: () => {
    return 'hello world';
  },
  onSuccess: (res) => {
    console.log(res); // hello world
  }
});

2. 异步任务

queue.push({
  taskParams: { id: 1 },
  task: async (params) => {
    await new Promise((r) => setTimeout(r, 500));
    return `task-${params?.id}`;
  },
  onSuccess: (res) => {
    console.log(res);
  }
});

3. 串行执行（FIFO）

任务按入队顺序依次执行，前一个结束后才会跑下一个。

const queue = new Queue();
const order: number[] = [];

queue.push({
  task: async () => {
    await new Promise((r) => setTimeout(r, 100));
    order.push(1);
  }
});

queue.push({
  task: async () => {
    order.push(2);
  }
});

// 最终 order === [1, 2]

4. taskParams 与 extraData

queue.push({
  taskParams: { title: '任务一' },
  extraData: { from: 'ui' },
  task: async (params) => {
    return params?.title;
  },
  onSuccess: (res, extra) => {
    console.log(res, extra); // '任务一' { from: 'ui' }
  }
});

5. 全局回调 + 任务回调

任务成功/失败时，会先触发任务级回调，再触发全局回调。

const queue = new Queue({
  onSuccess: (res, extra) => {
    console.log('全局成功', res, extra);
  },
  onError: (err, extra) => {
    console.error('全局失败', err, extra);
  }
});

queue.push({
  extraData: { a: 1 },
  task: async () => 'ok',
  onSuccess: (res, extra) => {
    console.log('任务成功', res, extra);
  }
});

queue.push({
  extraData: { a: 2 },
  task: async () => {
    throw new Error('fail');
  },
  onError: (err, extra) => {
    console.log('任务失败', err, extra);
  }
});

6. 重试（全局默认）

retryCount 表示额外重试次数（不含首次）。失败后按指数退避等待再试。

const queue = new Queue({
  retryCount: 2 // 最多执行 1 + 2 = 3 次
});

queue.push({
  task: async () => {
    // 前几次失败，最终成功即可被重试救回
    throw new Error('临时失败');
  },
  onError: (err) => {
    console.error('重试耗尽后仍失败', err);
  }
});

7. 任务级重试覆盖全局

const queue = new Queue({ retryCount: 5 });

queue.push({
  retryCount: 0, // 本任务不重试
  task: async () => {
    throw new Error('fail');
  },
  onError: (err) => {
    console.error(err);
  }
});

8. 超时（需显式配置）

未设置 timeout 时不会自动超时。超时后抛出 TimeoutError，并触发当前任务的 AbortSignal。

import Queue, { TimeoutError, isTimeoutError } from 'cy-plugin-queue';

const queue = new Queue({
  timeout: 3000 // 全局默认 3 秒
});

queue.push({
  task: async (_params, ctx) => {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => resolve('done'), 10000);

      ctx?.signal?.addEventListener('abort', () => {
        clearTimeout(timer);
        reject(new Error('aborted'));
      });
    });
  },
  onError: (err) => {
    if (isTimeoutError(err) || err instanceof TimeoutError) {
      console.error('任务超时', err.message); // Task timeout
    }
  }
});

9. 任务级超时覆盖全局

const queue = new Queue({ timeout: 10000 });

queue.push({
  timeout: 1000, // 本任务 1 秒超时
  task: () => new Promise(() => {}),
  onError: (err) => {
    console.error(err); // TimeoutError
  }
});

10. 仅任务设置超时（无全局 timeout）

const queue = new Queue(); // 未配置全局 timeout

queue.push({
  timeout: 2000,
  task: () => new Promise(() => {}),
  onError: (err) => {
    console.error(err); // TimeoutError
  }
});

11. 超时默认不重试 / 开启超时重试

超时默认不重试。需要时设置 retryOnTimeout: true。

// 默认：超时立即失败，不重试
const queue1 = new Queue({
  timeout: 1000,
  retryCount: 2
});

// 开启：超时也走重试
const queue2 = new Queue({
  timeout: 1000,
  retryCount: 2,
  retryOnTimeout: true
});

12. 中断当前任务（abort + signal）

任务内需监听 ctx.signal，才能在 abort() / 超时时正确收尾。

const queue = new Queue();

queue.push({
  task: (_params, ctx) => {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => resolve('done'), 10000);

      ctx?.signal?.addEventListener('abort', () => {
        clearTimeout(timer);
        reject(new Error('aborted'));
      });
    });
  },
  onError: (err) => {
    console.error(err.message); // aborted
  }
});

// 稍后中断
setTimeout(() => {
  queue.abort();
}, 500);

13. 清空等待队列 clear

const queue = new Queue({
  onError: (err) => {
    console.error('全局', err.message);
  }
});

queue.push({
  task: async () => {
    await new Promise((r) => setTimeout(r, 3000));
  }
});

queue.push({
  task: () => 'queued',
  onError: (err) => {
    console.error('排队任务', err.message); // Task cleared
  }
});

// 只清空等待队列，不触发回调；当前任务继续跑
queue.clear();

// 清空等待队列，并对排队任务触发 onError（当前任务不受影响）
queue.clear(true);

// 清空等待队列 + 中断当前正在执行的任务
queue.clear(false, true);

// 清空并回调排队任务，同时中断当前任务
queue.clear(true, true);

| 调用 | 等待队列 | 排队任务 onError | 当前任务 | |------|----------|------------------|----------| | clear() | 清空 | 否 | 继续 | | clear(true) | 清空 | 是（Task cleared） | 继续 | | clear(false, true) | 清空 | 否 | 中断（需任务监听 signal） | | clear(true, true) | 清空 | 是 | 中断 |

14. 查询状态

console.log(queue.length); // 等待中任务数
console.log(queue.isTaskExecuting); // 是否正在执行

15. 自定义任务 ID

queue.push({
  id: 'upload-avatar',
  task: async () => {
    return 'ok';
  }
});

16. 开启调试日志

const queue = new Queue({
  debug: true
});

开启后会输出入队参数、开始/成功/失败、重试、超时等信息。

17. 综合示例

import Queue, { isTimeoutError } from 'cy-plugin-queue';

const queue = new Queue({
  debug: true,
  retryCount: 2,
  timeout: 3000,
  retryOnTimeout: false,
  onSuccess: (result, extra) => {
    console.log('全局成功:', result, extra);
  },
  onError: (err, extra) => {
    if (isTimeoutError(err)) {
      console.error('全局超时:', err, extra);
      return;
    }
    console.error('全局失败:', err, extra);
  }
});

// 带中断感知的异步任务
queue.push({
  taskParams: { title: '任务一' },
  extraData: { a: 1 },
  timeout: 3000,
  retryCount: 2,
  task: async (param, ctx) => {
    return new Promise<string>((resolve, reject) => {
      const timer = setTimeout(() => {
        resolve(JSON.stringify(param));
      }, 1000);

      ctx?.signal?.addEventListener('abort', () => {
        clearTimeout(timer);
        reject(new Error('aborted'));
      });
    });
  },
  onSuccess: (res, extra) => {
    console.log('任务一成功', res, extra);
  }
});

// 普通异步任务
queue.push({
  taskParams: { title: '任务二' },
  extraData: { a: 2 },
  task: async (param) => {
    await new Promise((r) => setTimeout(r, 500));
    return String(param?.title);
  },
  onSuccess: (res, extra) => {
    console.log('任务二成功', res, extra);
  }
});

// 失败 + 重试
queue.push({
  taskParams: { title: '任务三' },
  retryCount: 2,
  task: async () => {
    throw new Error('模拟失败');
  },
  onError: (err) => {
    console.log('任务三失败', err);
  }
});

// 同步直接返回
queue.push({
  taskParams: { title: '任务四' },
  task: () => '任务四成功',
  onSuccess: (res) => {
    console.log(res);
  }
});

// 需要时：
// queue.abort();           // 中断当前任务
// queue.clear();           // 清空等待队列
// queue.clear(true);       // 清空并回调排队任务
// queue.clear(false, true); // 清空并中断当前任务

行为说明

1. 串行：push 后自动调度；已有任务在执行时只入队，不并发。
2. 超时可选：只有配置了全局或任务级 timeout 才会启用超时。
3. 超时错误：统一为 TimeoutError（name: 'TimeoutError'，code: 'TIMEOUT'）；即便任务因 abort 先抛错，对外仍为超时错误。
4. Abort 不重试：识别 AbortError / ABORT_ERR / message 含 abort 的错误，直接失败。
5. 超时默认不重试：需 retryOnTimeout: true 才重试超时。
6. clear 默认不影响当前任务：要停当前任务请用 abort() 或 clear(..., true)。
7. 任务需自行响应 signal：否则 abort() / 超时 abort 可能无法立刻结束业务侧异步逻辑。

License

MIT