BullMQ Pro 分组（Groups）中的沙盒处理器（Sandboxed Processors）：gid 与隔离执行实战指南

• 后端
• 消息队列
• 任务调度

在 BullMQ Pro 中，你可以把沙盒处理器（Sandboxed Processors）直接用于**分组（Groups）**队列，让每个分组的任务在独立进程/线程中隔离执行，同时通过 job 对象上的gid属性感知任务所属分组。本文以 BullMQ Pro 的官方文档为核心，结合开源仓库中沙盒机制的源码实现，完整讲解分组沙盒处理器的启用方式、SandboxedJobPro的用法、底层通信原理以及使用边界。

为什么要在分组队列上使用沙盒处理器

在进入具体用法之前，先回顾两个核心概念，它们共同构成了本主题的上下文：

• 分组（Groups）：BullMQ Pro 允许在单个队列内通过group.id把任务划分为多个“虚拟队列”，并以**轮询（round-robin）**方式从各分组中取任务，保证每个分组的任务按顺序处理、互不饥饿。详细说明见 Groups 总览。
• 沙盒处理器（Sandboxed Processors）：BullMQ 标准版即支持把处理器放在独立文件中，由 Worker 以独立子进程（child process）或 Worker Threads 方式加载执行。这样 CPU 密集型任务不会阻塞主进程的 Node.js 事件循环，避免因无法及时续租锁而导致任务被判定为 stalled。基础机制见 沙盒处理器指南。

把两者结合，你就能得到“分组级别的隔离执行”：既享受分组带来的公平调度，又享受沙盒带来的进程隔离。其工作方式与标准 BullMQ 中的沙盒处理器基本一致，唯一显著区别是：传入处理器的 job 对象中会多出一个gid（group id）属性，供你按分组区分处理逻辑。

为分组任务启用沙盒处理器

启用方式与标准沙盒处理器完全一致——把处理器定义在独立文件中，并将该文件路径（或 URL）传给 Worker 构造函数。

// my_group_processor.ts import { SandboxedJobPro } from '@taskforcesh/bullmq-pro'; module.exports = function (job: SandboxedJobPro) { // 处理分组任务 };

// worker.ts import { WorkerPro } from '@taskforcesh/bullmq-pro'; import * as path from 'path'; const processorFile = path.join(__dirname, 'my_group_processor.js'); const worker = new WorkerPro(queueName, processorFile, { // 可选：使用 Worker Threads 而非子进程 useWorkerThreads: true, });

标准 BullMQ 沙盒处理器使用的类型是SandboxedJob（见 sandboxed-job.ts）；而在 BullMQ Pro 中，处理器应当使用SandboxedJobPro，它是在SandboxedJob基础上扩展了 Pro 特性的任务对象。分组场景下，处理器收到的 job 具备完整的组信息。

在处理器内部访问 gid 与分组配置

分组沙盒处理器的核心价值在于：处理器能够知道“当前这个任务属于哪个组”，从而做出分组相关的处理决策。官方文档给出了完整的校验示例：

import { SandboxedJobPro } from '@taskforcesh/bullmq-pro'; module.exports = function (job: SandboxedJobPro) { expect(job).to.have.property('gid'); expect(job.opts).to.have.property('group'); expect(job.opts.group).to.have.property('id'); expect(job.opts.group.id).to.be.a('string'); expect(job.opts.group.id).to.equal(job.gid); };

这段代码揭示了三个关键信息：

1. job.gid：任务所属分组的 id，直接挂在 job 顶层，是分组沙盒处理器特有的属性；
2. job.opts.group.id：任务选项中的分组配置，与gid保持一致；
3. 类型约定：job.opts.group.id是string类型，且在expect(job.opts.group.id).to.equal(job.gid)的约束下，gid与分组 id 始终相等——你可以放心地用任意一个值来标识分组。

在实际业务中，你可以据此为不同分组执行差异化逻辑：

import { SandboxedJobPro } from '@taskforcesh/bullmq-pro'; module.exports = async function (job: SandboxedJobPro) { const { gid } = job; if (gid === 'vip-users') { await processWithHighPriority(job); } else { await processWithDefaultPriority(job); } // 也可以直接读取 opts.group 中的完整配置 await doSomethingSpecialForMyGroup(job.opts.group); };

需要说明的是：gid属于 BullMQ Pro（@taskforcesh/bullmq-pro）的特性，在开源版（bullmq）的 sandboxed-job.ts 接口中并不存在该字段；分组队列的添加与消费示例（group: { id }选项、WorkerPro的 round-robin 调度）可参考 Groups 总览。

沙盒处理器底层如何工作

理解了用法后，再看开源仓库中沙盒机制的实现，能帮助你更好地评估分组沙盒的行为特征。沙盒核心链路分三层：

1. Worker 侧的调度器

Worker 构造处理器时，会把处理器文件包装为沙盒执行函数。见 worker.ts 中sandbox(...)的调用（src/classes/worker.ts#L396-L404附近）：当传入的是文件路径而非函数时，Worker 会把它交给 sandbox.ts 处理。

2. 父进程与子进程的消息代理

sandbox.ts 是整个沙盒机制的“桥”：它从ChildPool中取回（retain）一个空闲子进程，把任务 JSON（job.asJSONSandbox()）通过ChildCommand.Start发给子进程，然后监听子进程回传的消息：

• Completed/Failed/Error：完成或失败，直接 resolve/reject；
• Progress/Log：转发到父进程侧的job.updateProgress/job.log；
• MoveToDelayed/MoveToWait/MoveToWaitingChildren：代理状态迁移；
• Update：代理updateData；
• GetChildrenValues/GetIgnoredChildrenFailures/GetDependenciesCount/GetDependencies：代理依赖查询，并把结果以ChildCommand.*Response回传给子进程。

这意味着，即便处理器运行在隔离进程中，它依然能对父进程侧的任务执行完整的“远程调用”，例如延迟、等待子任务、更新进度、读取依赖等。任务执行完毕后，子进程会被释放回ChildPool复用（src/classes/sandbox.ts#L172-L178）。

3. 子进程侧的 job 包装

在子进程侧，child-processor.ts 的wrapJob方法（src/classes/child-processor.ts#L156-L341）把从父进程收到的 JSON 任务重新包装成SandboxedJob对象：解析data与returnValue，并把updateProgress、log、moveToDelayed、moveToWait、moveToWaitingChildren、updateData、getChildrenValues、getIgnoredChildrenFailures、getDependenciesCount、getDependencies等函数以 IPC 消息代理的形式挂到 job 上。由于分组信息（gid、opts.group）随任务 JSON 一并传输，所以这些字段在子进程中天然可用——这正是分组沙盒处理器能读到gid的原因。此外，子进程还支持通过 AbortController 信号取消当前任务（cancel方法，见src/classes/child-processor.ts#L130-L134），并与workerForkOptions/workerThreadsOptions等底层选项配合。

分组沙盒的配置细节与运行注意事项

沙盒进程的两种运行机制

根据 沙盒处理器指南 与 sandboxed-options.ts，沙盒处理器有两种运行方式：

两个选项都要求处理器以文件路径形式传入 Worker。在 Windows 系统上，官方还推荐使用pathToFileURL将文件路径转换为 URL 实例再传入，以避免路径解析问题。

并发与 stalled 的关系

沙盒处理器的设计初衷是把 CPU 密集逻辑从“簿记代码”中剥离，避免事件循环被占用而无法续租任务锁。但需要特别注意：并发因子（concurrency）仍需控制在合理范围内，否则子进程过多依然可能导致任务被判定为 stalled。分组队列在多个 Worker 或并发大于 1 时会并行处理任务，因此并发参数要结合分组数量与机器资源一起评估。

当前支持范围：只有 Groups 是 Pro 特性

这是官方明确标注的使用边界，务必注意：

Groups are the only Pro features supported by Sandboxed processors for now. （分组是目前沙盒处理器唯一支持的 Pro 特性。）

即：BullMQ Pro 的沙盒处理器目前只支持分组这一项 Pro 能力；其余 Pro 特性（如本地分组并发、本地分组限流等，见 Groups 目录 下的相关文档）尚未开放给沙盒处理器。在设计架构时，不要把其他 Pro 特性依赖在沙盒处理器内部。

与 Batches 的配合

如果你的场景把分组与批处理（Batches）结合使用，可以启用groupAffinity让每个批次只包含同一分组的任务（或非分组任务），分组之间仍按 round-robin 顺序跨批次调度。详细配置示例见 Batches 与 Group Affinity。

小结

分组沙盒处理器是 BullMQ Pro 中“公平调度”与“隔离执行”的结合点：启用方式与标准沙盒处理器相同，只需把处理器类型换成SandboxedJobPro；处理器内可通过job.gid与job.opts.group.id识别分组并执行差异化逻辑；底层则由 sandbox.ts 与 child-processor.ts 组成的 IPC 消息代理完成隔离执行与远程状态操作。使用时牢记官方边界——沙盒处理器目前仅支持 Groups 这一项 Pro 特性，并根据实际资源合理设置并发，即可在分组队列上获得稳定、隔离、可横向扩展的处理能力。

• 后端
• 消息队列
• 任务调度