构建带工作池与重试的进程内任务队列

项目目标

本项目实现一个持久化到磁盘的任务队列，作业可通过有界工作池并发执行，失败按指数退避重试，重启后仍能恢复待处理作业。所有时间相关的决策都通过注入实现，因此测试是确定性的而非依赖真实等待。

核心特性

• 按名称入队并返回稳定 id
• 有界工作池控制并发度
• 指数退避加抖动的重试策略
• 状态持久化到磁盘，重启恢复
• 命令行查看各状态计数

架构与分层

队列存储以原子写入方式序列化作业；调度器按到期时间拉取；工作池在并发上限内运行处理器；重试策略把失败作业按计算出的延迟放回队列。处理器按名称注册，队列核心不引用任何业务逻辑。

实施阶段

六个阶段为：持久化作业存储、重试策略、有界工作池、处理器与失败处理、队列 CLI、运行循环与测试。注入式时钟让一小时的回退在微秒内完成验证。

测试与验收

用 vitest 运行。队列测试用内存存储断言入队 id 稳定、nextDue 尊重 runAt；重试测试注入假时钟断言退避增长并封顶；持久化测试从磁盘重载并确认待处理作业仍在。

延伸方向

优先级字段、死信列表与可插拔存储分别对应拉取、收集与替换的清晰接缝。处理器按名称注册，新增作业类型无需改动队列核心。

本译文由 AI 依据英文原文完整重写，覆盖目标、特性、架构、分阶段实现、测试与扩展方向，供中文读者对照英文原稿阅读。

代码示例

export function createQueue({ store, handlers, clock = Date, now = () => clock.now() }) {
  async function enqueue(name, payload = {}, options = {}) {
    const job = {
      id: crypto.randomUUID(),
      name,
      payload,
      status: 'pending',
      attempts: 0,
      runAt: now(),
      maxAttempts: options.maxAttempts ?? 3,
    };
    await store.add(job);
    return job.id;
  }
  async function runOnce() {
    const job = await store.nextDue(now());
    if (!job) return null;
    return handlers[job.name](job, { enqueue });
  }
  return { enqueue, runOnce };
}

mkdir job-queue && cd job-queue
npm init -y
npm install -D vitest@^2.1.0

{
  "name": "job-queue",
  "type": "module",
  "scripts": {
    "queue": "node src/cli.js",
    "test": "vitest run"
  }
}

import { readFile, writeFile, rename } from 'node:fs/promises';

export function createFileStore(path) {
  let jobs = [];
  let loaded = false;
  async function ensure() {
    if (loaded) return;
    try { jobs = JSON.parse(await readFile(path, 'utf8')); }
    catch (err) { if (err.code !== 'ENOENT') throw err; jobs = []; }
    loaded = true;
  }
  async function flush() {
    const tmp = path + '.tmp';
    await writeFile(tmp, JSON.stringify(jobs, null, 2));
    await rename(tmp, path);
  }
  return {
    async add(job) { await ensure(); jobs.push(job); await flush(); },
    async nextDue(at) {
      await ensure();
      const due = jobs.filter((j) => j.status === 'pending' && j.runAt <= at);
      due.sort((a, b) => a.runAt - b.runAt);
      return due[0] ?? null;
    },
    async update(id, patch) {
      const i = jobs.findIndex((j) => j.id === id);
      if (i >= 0) jobs[i] = { ...jobs[i], ...patch };
      await flush();
    },
  };
}

export function backoffDelay(attempt, { base = 100, cap = 30000, jitter = true } = {}) {
  const raw = Math.min(cap, base * 2 ** Math.max(0, attempt - 1));
  if (!jitter) return raw;
  return Math.round(raw / 2 + Math.random() * (raw / 2));
}

export function planRetry(job, now) {
  const attempts = job.attempts + 1;
  if (attempts >= job.maxAttempts) {
    return { ...job, attempts, status: 'failed' };
  }
  return { ...job, attempts, status: 'pending', runAt: now + backoffDelay(attempts) };
}

export async function runPool({ queue, handlers, concurrency = 4, untilEmpty = true }) {
  let running = 0;
  let processed = 0;
  async function attempt() {
    if (running >= concurrency) return;
    const job = await queue.claim();
    if (!job) return;
    running += 1;
    try {
      await handlers[job.name](job);
      await queue.complete(job.id);
      processed += 1;
    } catch (err) {
      await queue.fail(job.id, err);
    } finally {
      running -= 1;
    }
  }
  while (untilEmpty && (await queue.hasPending())) {
    await Promise.all(Array.from({ length: concurrency }, attempt));
  }
  return { processed };
}

export const handlers = {
  async sendEmail(job, { log }) {
    if (!job.payload.to) throw new Error('missing recipient');
    log('sent to ' + job.payload.to);
    return { ok: true };
  },
  async resizeImage(job) {
    if (job.payload.width <= 0) throw new Error('invalid width');
    return { width: job.payload.width };
  },
};

export function wrapHandlers(raw, { log = () => {} } = {}) {
  return Object.fromEntries(
    Object.entries(raw).map(([name, fn]) => [
      name,
      async (job) => fn(job, { log: (m) => log(name + ': ' + m) }),
    ]),
  );
}

import { createFileStore } from './store.js';
import { createQueue } from './queue.js';
import { handlers } from '../jobs/handlers.js';

const [command, name, payloadJson] = process.argv.slice(2);
const store = createFileStore(process.env.QUEUE_FILE ?? './queue.json');
const queue = createQueue({ store, handlers });

if (command === 'add') {
  if (!name) { console.error('usage: queue add <name> [json]'); process.exit(1); }
  const id = await queue.enqueue(name, payloadJson ? JSON.parse(payloadJson) : {});
  console.log(id);
} else if (command === 'status') {
  const counts = await store.counts();
  console.log(JSON.stringify(counts));
} else {
  console.error('unknown command');
  process.exit(1);
}

import { createQueue } from '../src/queue.js';

test('retry moves a failed job back to pending with a delay', async () => {
  const store = memoryStore();
  const clock = { now: 1000 };
  const handlers = { boom: async () => { throw new Error('fail'); } };
  const queue = createQueue({ store, handlers, clock, now: () => clock.now });
  const id = await queue.enqueue('boom', {}, { maxAttempts: 2 });
  await queue.claim();
  await queue.fail(id, new Error('fail'));
  const job = await store.get(id);
  expect(job.status).toBe('pending');
  expect(job.runAt).toBeGreaterThan(clock.now);
});

function memoryStore() {
  const map = new Map();
  return {
    async add(j) { map.set(j.id, j); },
    async get(id) { return map.get(id); },
  };
}