BullMQ Pro 局部组并发（Local Group Concurrency）：为不同分组设置差异化并发因子

后端消息队列任务调度【免费下载链接】bullmqBullMQ - Message Queue and Batch processing for NodeJS, Python, .NET, Elixir, Rust and PHP based on Redis or PostgreSQL项目地址https://gitcode.com/gh_mirrors/bu/bullmq点击查看免费下载导读在 BullMQ Pro 的 Groups 体系中默认的组并发因子是全局统一作用于所有分组的但真实业务中往往需要为不同的分组例如代表不同用户、不同租户或不同服务等级的组配置彼此独立的并发上限。本文围绕官方文档docs/gitbook/bullmq-pro/groups/local-group-concurrency.md展开详细讲解如何通过setGroupConcurrency/getGroupConcurrency为指定分组设置局部并发值如何与 Worker 实例级的组并发配置协同工作以及这一特性在 Redis 中的存储与生命周期管理注意事项。读完本文你将能够为单个队列内的不同分组配置差异化并发策略并理解其底层约束与最佳实践。Groups 与组并发先厘清两个层次在进入局部组并发之前需要先区分 Groups 功能中的两个并发层次Worker 实例级全局组并发通过 WorkerPro 的group.concurrency选项配置作用于所有没有自己并发定义的分组。参见 concurrency.md。分组级局部组并发通过 QueuePro 的setGroupConcurrency方法配置只作用于指定的单个分组覆盖全局默认值。官方文档明确指出It is also possible to set a specific concurrency value to a given group. This is useful if you require that different groups should run with different concurrency factors.——这正是局部组并发的核心动机不同分组需要不同的并发因子。举一个典型场景假设一个队列同时承载免费用户与付费用户的任务免费用户的组并发上限设为 2付费用户的组并发上限设为 8其余新注册用户的组则回落到 Worker 层的默认并发值。这种按组差异化的能力即由局部组并发提供。全局组并发回顾默认行为根据 concurrency.md默认情况下每个组可以并行处理的任务数量是不受限制的即使设置了速率限制rate limit那也只限制处理速度不会限制同时处理的任务数。通过group.concurrency可以约束每个组最多同时处理多少个任务import { WorkerPro } from taskforcesh/bullmq-pro; const worker new WorkerPro(myQueue, processFn, { group: { concurrency: 3, // 限制每个组最多 3 个并行任务 }, concurrency: 100, connection, });该配置是全局的无论你实例化多少个 Worker也无论每个 Worker 的concurrency有多大任何组在任何时刻都不会被处理超过 3 个任务。局部组并发正是在此基础上为某个特定组提供独立的覆盖值。使用 setGroupConcurrency 设置局部并发要为特定分组设置独立的并发值使用 QueuePro 的setGroupConcurrency方法import { QueuePro } from taskforcesh/bullmq-pro; const queue new QueuePro(myQueue, { connection }); const groupId my group; await queue.setGroupConcurrency(groupId, 4);groupId要设置并发的分组标识与添加任务时job.opts.group.id中使用的 id 对应第二个参数4该分组的最大并发任务数即该组同时最多被处理的任务数量。设置完成后该组在任意 Worker 实例上的总并发都会被限制为 4即使 Worker 层的concurrency是 100 也是如此。使用 getGroupConcurrency 读取当前值需要确认某个分组当前的局部并发配置时使用getGroupConcurrencyconst concurrency await queue.getGroupConcurrency(groupId);该方法返回该组当前存储的局部并发值。从源码变更历史看getGroupConcurrency与setGroupConcurrency是配套设计的一组能力changelog 中对应 add getGroupConcurrency method 与 support for local group concurrency using setGroupConcurrency method 两条演进记录分别用于写入与读取。关键警告Worker 层 group.concurrency 是前置条件官方文档以醒目的 warning 提示强调请务必同时在 Worker 实例层面设置 Group Concurrency即group.concurrency这是该特性正常工作的前提同时它也会作为没有定义局部并发的那些分组的默认并发值。也就是说局部组并发是叠加在全局组并发之上的覆盖层二者缺一不可全局层group.concurrency提供默认值与运行前提所有未通过setGroupConcurrency设置并发值的组会统一使用 Worker 层配置的group.concurrency局部层setGroupConcurrency提供精细化覆盖被指定到的组以局部值为准。正确的最小配置示例import { QueuePro, WorkerPro } from taskforcesh/bullmq-pro; const queue new QueuePro(myQueue, { connection }); // 为 premium 组设置局部并发 8 await queue.setGroupConcurrency(premium, 8); // Worker 层必须设置 group.concurrency作为默认值与功能前提 const worker new WorkerPro(myQueue, async job { // 处理任务 }, { group: { concurrency: 2, // 其他组的默认并发值 }, concurrency: 100, connection, });在该配置下premium组的并发上限为 8其余组的并发上限为 2而 Worker 实例本身的concurrency: 100只影响总体的并行能力上限不影响按组约束。生命周期管理并发值存储在 Redis 中局部组并发的一个关键特性是持久化存储于 Redis而非内存中的临时配置。官方文档明确提示当为一个组指定并发因子时这个值会被存储在 Redis 中因此如果你不再使用它你有责任将其移除。这与分组本身的设计一致——Groups 说明文档指出空分组不会消耗 Redis 资源。但从实现演进看组相关的配置项如并发、速率限制是独立维护的元数据changelog 中有一条 remove concurrency when removing group 的修复记录说明删除组时应当同步清理其并发配置避免残留。因此建议在业务中通过setGroupConcurrency设置了局部并发后若该分组即将废弃例如用户注销、租户下线主动清理对应的并发配置在删除分组或清空队列数据时确认组并发元数据一并被移除使用getGroupConcurrency定期审计排查是否存在遗留的组级配置。与分组其他能力的协同局部组并发是 Groups 特性族中的一员与以下能力可以组合使用局部组速率限制与局部组并发类似local-group-rate-limit.md 允许通过setGroupRateLimit为特定组设置独立的速率限制同时要求 Worker 层提供group.limit作为默认值与检查前提——与局部组并发全局默认 局部覆盖的模型完全一致组最大大小max-group-size.md 允许通过group.maxSize限制组内任务数量上限分组优先级与暂停Groups 还支持组内优先级调整与暂停指定分组便于在运维层面精细控制不同分组的处理节奏。并发控制负责同一时刻最多处理多少速率限制负责单位时间最多处理多少二者是正交且互补的约束维度可按业务需要组合启用。小结局部组并发Local Group Concurrency为 BullMQ Pro 的多租户、多等级分组场景提供了按组细分的并发控制能力配置途径作用范围说明WorkerProgroup.concurrency全局默认所有组的默认并发上限也是局部并发功能的前提QueueProsetGroupConcurrency(groupId, n)指定分组为单个组覆盖默认并发值存储在 RedisQueueProgetGroupConcurrency(groupId)读取查询指定组的局部并发配置核心实践要点先保证 Worker 层配置group.concurrency再通过setGroupConcurrency做局部覆盖并发值持久化在 Redis 中废弃组时需自行清理建议结合getGroupConcurrency做配置审计。掌握这一组合即可在单个队列内实现不同分组不同并发、新分组自动回落默认值的弹性处理模型。赞分享后端消息队列任务调度【免费下载链接】bullmqBullMQ - Message Queue and Batch processing for NodeJS, Python, .NET, Elixir, Rust and PHP based on Redis or PostgreSQL项目地址https://gitcode.com/gh_mirrors/bu/bullmq点击查看免费下载相关推荐WSABuilds一个 .7z 装好带 Google Play 和 Magisk root 的 WSAWSABuilds一个 .7z 装好带 Google Play 和 Magisk root 的 WSA 2025 年 3 月微软正式终止 WSA 官方支持S后端消息队列任务调度WSABuilds 安装教程Windows 10/11 如何免费运行带 Google Play 和 Root 的安卓子系统WSABuilds 安装教程Windows 10/11 如何免费运行带 Google Play 和 Root 的安卓子系统 微软已在 2025 年 3 月终止后端消息队列任务调度BullMQ Worker 并发指南本地并发因子与多 Worker 横向扩展实践BullMQ Worker 并发指南本地并发因子与多 Worker 横向扩展实践 导读 本文以 BullMQ 官方文档 workers/concurrency后端消息队列任务调度上一篇一起听歌吧多房间版音乐播放核心实现原理下一篇Tess-4-27B量化版本测评Q4_K_M、Q6_K与Q8_0性能对比创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考