Flows 与任务依赖

有些工作不是一个任务，而是一棵依赖树。例如生成报告需要先并行读取销售、库存和退款数据，所有子任务成功后，父任务才能汇总。BullMQ 的 Flow 使用父子任务表达这种依赖。

Flow 的基本语义

父任务初始处于 waiting-children，所有必需子任务成功后才进入 waiting 并被 Worker 获取。父子任务可以位于不同队列，整个 Flow 由 FlowProducer 原子添加。

import { FlowProducer } from "bullmq";

interface ReportFlowData {
  reportId: string;
  source?: "sales" | "inventory" | "refunds";
}

const flowProducer = new FlowProducer({
  connection: { host: "127.0.0.1", port: 6379 },
});

const flow = await flowProducer.add({
  name: "aggregate-report",
  queueName: "report-parent",
  data: { reportId: "report-42" } satisfies ReportFlowData,
  children: [
    {
      name: "load-source",
      queueName: "report-source",
      data: {
        reportId: "report-42",
        source: "sales",
      } satisfies ReportFlowData,
    },
    {
      name: "load-source",
      queueName: "report-source",
      data: {
        reportId: "report-42",
        source: "inventory",
      } satisfies ReportFlowData,
    },
  ],
});

console.log(flow.job.id);

Flow 中的自定义 jobId 不能包含冒号 :，因为冒号被 BullMQ 用作内部键分隔符。

子任务返回结果，父任务汇总

import { Job, Worker } from "bullmq";

interface SourceJobData {
  reportId: string;
  source: "sales" | "inventory" | "refunds";
}

interface SourceResult {
  count: number;
}

const sourceWorker = new Worker<SourceJobData, SourceResult>(
  "report-source",
  async (job): Promise<SourceResult> => {
    return { count: await loadCount(job.data.source) };
  },
  { connection: { host: "127.0.0.1", port: 6379 } },
);

const parentWorker = new Worker<ReportFlowData, number>(
  "report-parent",
  async (job: Job<ReportFlowData, number>): Promise<number> => {
    const childrenValues =
      await job.getChildrenValues<SourceResult>();

    const total = Object.values(childrenValues).reduce(
      (sum, result) => sum + result.count,
      0,
    );

    return total;
  },
  { connection: { host: "127.0.0.1", port: 6379 } },
);

async function loadCount(
  source: SourceJobData["source"],
): Promise<number> {
  console.log(`读取 ${source}`);
  return 100;
}

getChildrenValues() 返回以子任务完整键为 key 的结果对象。不要依赖对象遍历顺序来表达业务顺序。

更深层的依赖树

Flow 可嵌套任意深度，处理顺序从叶子向根部推进。例如 engine -> wheels -> chassis 的嵌套定义，实际会先处理最深的 chassis，再处理 wheels，最后处理 engine。

虽然 API 允许很深的树，但深度和节点数会增加排查、Redis 数据量和失败恢复的复杂度。初学阶段建议保持 2 到 3 层，把业务流程图画清楚后再编码。

查询依赖状态

父任务可查询依赖及数量：

const state = await flow.job.getState();
const counts = await flow.job.getDependenciesCount();

const firstPage = await flow.job.getDependencies({
  processed: { count: 20, cursor: 0 },
  unprocessed: { count: 20, cursor: 0 },
  failed: { count: 20, cursor: 0 },
  ignored: { count: 20, cursor: 0 },
});

console.log({ state, counts, firstPage });

大型 Flow 必须分页获取依赖，不要一次拉取全部子任务到内存。

子任务失败时的策略

默认情况下，父任务会等待依赖满足。实际业务应明确选择失败语义，并把选项配置在对应的子任务上。

1. 关键子任务失败，让父任务失败

{
  name: "load-payment-data",
  queueName: "report-source",
  data: { reportId: "report-42", source: "sales" },
  opts: { failParentOnFailure: true },
}

failParentOnFailure: true 会让父任务失败，并可沿树向上传播。官方说明该失败是惰性完成的：父任务需要被 Worker 处理后才转换到 failed，错误表现为 UnrecoverableError。

2. 可选子任务失败，忽略该依赖

{
  name: "load-optional-data",
  queueName: "report-source",
  data: { reportId: "report-42", source: "refunds" },
  opts: { ignoreDependencyOnFailure: true },
}

父任务可以继续，并通过 getIgnoredChildrenFailures() 查看被忽略的失败。适合非关键推荐、可选统计等允许降级的步骤。

3. 子任务失败后立即唤醒父任务

BullMQ 5.58.0 起提供 continueParentOnFailure: true。父任务可立即开始补偿或清理，不必等其他子任务结束：

const failedChildren = await job.getFailedChildrenValues();

if (Object.keys(failedChildren).length > 0) {
  await job.removeUnprocessedChildren();
  await recordFlowFailure(job.id ?? "unknown", failedChildren);
  return;
}

removeUnprocessedChildren() 只移除尚未开始的 waiting 或 delayed 子任务，不会中止已经 active 的任务。

三种策略表达不同业务含义，不要同时随意叠加：

策略 含义
failParentOnFailure 关键依赖失败，整条流程失败
ignoreDependencyOnFailure 可选依赖失败，父任务继续
continueParentOnFailure 失败后立刻让父任务执行补偿逻辑

删除 Flow 的注意事项

因此，生产环境删除 Flow 前应查询状态并记录操作，不要把删除当作随意的“取消按钮”。

复杂度控制与最佳实践

本章小结

Flow 将任务组织成依赖树，父任务等待子任务成功后再处理。使用 FlowProducer 原子创建，父任务通过 getChildrenValues() 聚合结果。真正困难的是失败语义：关键依赖、可选依赖和补偿流程必须分别选择合适策略，同时控制树的规模并保证每个节点幂等。

官方文档