后端消息队列任务调度【免费下载链接】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 官方指南 auto-removal-of-jobs.md 展开系统讲解 Worker 级自动清理已完成/失败作业的三种策略全部删除、按数量保留、按年龄保留并结合仓库源码src/types/keep-jobs.ts、src/interfaces/worker-options.ts、src/commands/moveToFinished-14.lua剖析其底层实现与惰性清理机制。读完本文你将掌握在 TypeScript、Python、Elixir、Rust 四种语言中正确配置作业留存策略并理解清理触发的真实时机从而避免 Redis 被无用数据撑满。为什么需要自动清理作业默认情况下当一个作业完成completed或失败failed后BullMQ 会把它保存在两个特殊的集合中completed集合与failed集合。这样做的目的是便于你在开发早期检查作业的运行结果。但随着系统进入生产级阶段如果不对这些终态作业做限制Redis 中会积累大量不再有用且不断膨胀的数据。为此BullMQ 提供了自动清理auto-removal机制支持多种留存策略通过 Worker 的removeOnComplete与removeOnFail两个选项进行配置。从源码结构看这两个选项不仅存在于 Worker 配置中也可以作为作业级选项出现在BaseJobOptions里类型为boolean | number | KeepJobs这为按作业粒度控制留存策略提供了可能Worker 级配置是本文主角也是官方文档的主线。策略一删除所有已终态作业最简单的清理方式是设置removeOnComplete/removeOnFail为{ count: 0 }此时所有作业一旦终态完成或失败就会被立即自动删除。以只配置失败清理为例四种语言的写法如下。TypeScriptconst myWorker new Worker( myQueueName, async job { // do some work }, { connection, removeOnFail: { count: 0 }, }, );Pythonfrom bullmq import Worker def process_job(job): # do some work pass worker Worker( myQueueName, process_job, { connection: connection, removeOnFail: {count: 0}, }, )Elixirdefmodule MyWorker do use BullMQ.Worker def process_job(_job) do # do some work :ok end end {:ok, worker} BullMQ.Worker.start_link( queue: myQueueName, processor: MyWorker.process_job/1, connection: connection, remove_on_fail: %{count: 0} )Rustuse bullmq::{Worker, WorkerOptions}; use bullmq::types::RemoveOnFinish; let worker Worker::new(myQueueName, processor, WorkerOptions { remove_on_fail: Some(RemoveOnFinish::Count(0)), ..Default::default() }).await?;⚠️注意当count: 0时作业会被删除无论其名称是什么。底层原理count 为 0 时跳过入集合为什么count: 0能实现立即删除看终态处理脚本 moveToFinished-14.lua它会从反序列化后的opts[keepJobs]中取出count、age与limitlocal maxCount opts[keepJobs][count] local maxAge opts[keepJobs][age] local maxLimit opts[keepJobs][limit] or 1000随后在是否保留作业的分支中moveToFinished-14.lua只有maxCount ~ 0时才会把作业ZADD进completed/failed集合并写入finishedOn时间戳当maxCount 0时作业直接走删除路径从 Redis 中彻底移除。这就是{ count: 0 }语义的源码级验证。策略二保留一定数量的作业你也可以指定一个最大保留数量。官方文档给出的良好实践是保留少量 completed 作业便于近期排障但保留数量大得多的 failed 作业因为失败信息往往更值得复盘TypeScriptconst myWorker new Worker( myQueueName, async job { // do some work }, { connection, removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 }, }, );Pythonfrom bullmq import Worker def process_job(job): # do some work pass worker Worker( myQueueName, process_job, { connection: connection, removeOnComplete: {count: 1000}, removeOnFail: {count: 5000}, }, )Elixirdefmodule MyWorker do use BullMQ.Worker def process_job(_job) do # do some work :ok end end {:ok, worker} BullMQ.Worker.start_link( queue: myQueueName, processor: MyWorker.process_job/1, connection: connection, remove_on_complete: %{count: 1000}, remove_on_fail: %{count: 5000} )Rustuse bullmq::{Worker, WorkerOptions}; use bullmq::types::RemoveOnFinish; let worker Worker::new(myQueueName, processor, WorkerOptions { remove_on_complete: Some(RemoveOnFinish::Count(1000)), remove_on_fail: Some(RemoveOnFinish::Count(5000)), ..Default::default() }).await?;底层原理removeJobsByMaxCount按数量保留的实现由 Lua 包含片段removeJobsByMaxCount.lua负责当completed/failed集合中作业数量超过maxCount时按分数时间戳从小到大修剪超出部分保证最旧的作业优先被淘汰。这一逻辑在作业进入终态集合后立即执行与moveToFinished脚本处于同一个原子事务内。策略三按作业年龄保留KeepJobs 全参数第三种方式是保留到一定年龄为止。此时removeOn*接受一个KeepJobs对象包含三个字段字段含义说明age最多保留多老的作业单位秒超出该年龄的作业会被清理count最多保留多少作业用于限制总量防止短时间内涌入大量作业导致内存耗尽limit每次清理迭代最多删除多少个作业控制清理操作对性能的影响避免单次清理过重count在短时间内涌入异常数量的作业场景下尤其有用——即使age还没到也会因为数量超限而触发清理limit则通过限制单批处理量把清理操作对 Redis 和主流程的性能冲击控制在可接受范围。四语言完整示例TypeScriptconst myWorker new Worker( myQueueName, async job { // do some work }, { connection, removeOnComplete: { age: 3600, // keep up to 1 hour count: 1000, // keep up to 1000 jobs limit: 100, // remove up to 100 jobs per cleanup iteration }, removeOnFail: { age: 24 * 3600, // keep up to 24 hours limit: 50, // remove up to 50 jobs per cleanup iteration }, }, );Pythonfrom bullmq import Worker def process_job(job): # do some work pass worker Worker( myQueueName, process_job, { connection: connection, removeOnComplete: { age: 3600, # keep up to 1 hour count: 1000, # keep up to 1000 jobs limit: 100, # remove up to 100 jobs per cleanup iteration }, removeOnFail: { age: 24 * 3600, # keep up to 24 hours limit: 50, # remove up to 50 jobs per cleanup iteration }, }, )Elixirdefmodule MyWorker do use BullMQ.Worker def process_job(_job) do # do some work :ok end end {:ok, worker} BullMQ.Worker.start_link( queue: myQueueName, processor: MyWorker.process_job/1, connection: connection, remove_on_complete: %{ age: 3600, # keep up to 1 hour count: 1000, # keep up to 1000 jobs limit: 100 # remove up to 100 jobs per cleanup iteration }, remove_on_fail: %{ age: 24 * 3600, # keep up to 24 hours limit: 50 # remove up to 50 jobs per cleanup iteration } )Rustuse bullmq::{Worker, WorkerOptions}; use bullmq::types::{RemoveOnFinish, KeepJobs}; let worker Worker::new(myQueueName, processor, WorkerOptions { remove_on_complete: Some(RemoveOnFinish::Options(KeepJobs { age: Some(3600 * 1000), // keep up to 1 hour (in milliseconds) count: Some(1000), // keep up to 1000 jobs limit: Some(100), // remove up to 100 jobs per cleanup iteration })), remove_on_fail: Some(RemoveOnFinish::Options(KeepJobs { age: Some(24 * 3600 * 1000), // keep up to 24 hours (in milliseconds) count: None, limit: Some(50), // remove up to 50 jobs per cleanup iteration })), ..Default::default() }).await?;跨语言差异age 单位与类型形态TypeScript / Python / Elixirage单位为秒如3600表示 1 小时。Rustage单位为毫秒见 rust/src/types.rs 中KeepJobs定义示例里必须写3600 * 1000。这是最容易踩坑的差异点迁移代码时务必换算。Rust 侧把清理策略建模为RemoveOnFinish枚举Bool/Count/Options(KeepJobs)见 rust/src/types.rs并通过 rust/src/worker.rs 的write_keep_jobs序列化为与 Redis 脚本兼容的格式。age 与 count 同时存在时的语义从KeepJobs的类型注释可以确认当age与count同时指定时保留的作业必须同时满足两个条件satisfies both properties。也就是说任何一个维度超限都会触发清理这与策略一、二中单维度限制的语义不同配置时需要明确预期。惰性清理机制没有后台定时器⚠️重要自动清理是惰性lazy的。作业不会因为到了该清理的时候就被主动移除——只有当一个新的作业完成或失败、进入终态集合时清理逻辑才会被执行。这意味着如果一个作业完成/失败后队列长期没有新作业即使其年龄早已超过age阈值它也会一直留在集合里直到后续同类作业completed 或 failed再次终态时才被顺手清掉。源码注释对此有明确说明src/types/keep-jobs.ts 与 src/interfaces/worker-options.ts清理是基于**尽力而为best-effort**的评估——BullMQ 不会运行后台定时器主动驱逐过期作业。因此在持续有流量的生产队列上该机制足够及时基本能满足内存控制需求在流量稀疏的队列上若对清理时效有严格要求应结合定时任务或在业务低峰手动触发清理而不能依赖removeOn*自动完成。从 Worker 配置到 Redis 脚本的完整调用链综合源码removeOnComplete/removeOnFail从配置到执行的完整链路如下配置定义Worker 选项在 src/interfaces/worker-options.ts 中声明注释明确指出默认行为是保留在 completed/failed 集合中即不配置时不清理。作业终态作业完成时Job.moveToCompleted 把this.opts.removeOnComplete透传给后端失败时 Job.moveToFailed 以同样方式透传this.opts.removeOnFail。Redis 脚本执行终态处理统一进入 moveToFinished-14.lua从中解析出keepJobs.count / age / limitlimit未指定时默认1000。实际删除数量超限走包含片段removeJobsByMaxCount.lua年龄超限走removeJobsByMaxAge.lua二者按分数时间戳有序修剪集合保证旧作业优先被淘汰。默认值limit默认1000见 moveToFinished-14.lua即未显式指定时单次清理迭代最多处理 1000 个作业。值得注意的一点是清理操作与作业入集合、finishedOn写入、事件裁剪等动作共同处于moveToFinished这一个 Lua 脚本单次原子执行中因此清理失败不会导致作业状态不一致这也是该机制在生产环境可以放心开启的底气。实践建议小结开发环境保持默认不配置removeOn*方便随时检查作业结果。生产环境推荐少量 completed 大量 failed的组合如{count: 1000}与{count: 5000}兼顾排障能力与内存占用。高吞吐/长期运行队列优先使用KeepJobs的age count limit组合利用limit控制单次清理的峰值开销流量稀疏时需自行补充定时清理机制弥补惰性缺陷。跨语言迁移特别注意 Rust 的age单位为毫秒其余语言为秒limit未配置时各语言最终都会落到脚本默认值1000。赞分享后端消息队列任务调度【免费下载链接】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点击查看免费下载相关推荐RVC AI语音转换实战指南10分钟音频训练变声模型的完整教程RVC AI语音转换实战指南10分钟音频训练变声模型的完整教程 Retrieval based Voice Conversion WebUI简称 RVC是人工智能AI 应用语音音频深度学习BullMQ 失败任务自动重试完全指南attempts 与 Backoff 策略深度解析BullMQ 失败任务自动重试完全指南attempts 与 Backoff 策略深度解析 导读 在 BullMQ 中队列任务Job在长时间运行后不可避免后端消息队列任务调度Erlangshen-DeBERTa-v2性能评测在OCNLI和CMNLI任务中的表现分析Erlangshen DeBERTa v2性能评测在OCNLI和CMNLI任务中的表现分析 Erlangshen DeBERTa v2 186M Chines上一篇如何用Edge-TTS实现高质量语音合成3个关键技术解析与架构深度剖析下一篇Prisma Binding API 完整指南从 Prisma 服务到 GraphQL 委托解析器的编程接口创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考