后端消息队列任务调度【免费下载链接】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点击查看免费下载FIFOFirst-In, First-Out先进先出是 BullMQ 中最基础也最常用的一种任务Job类型任务按照被加入队列的顺序被处理先入队者先被消费。本文以官方指南文档 docs/gitbook/guide/jobs/fifo.md 为主体结合本仓库源码与测试讲解 FIFO 的顺序保证机制、入队 API 参数尤其是removeOnComplete/removeOnFail清理策略以及defaultJobOptions队列级默认配置并补充 Redis 与 PostgreSQL 两种后端下的底层实现细节帮助你在生产环境准确使用 FIFO 语义。FIFOBullMQ 的默认任务类型在 BullMQ 中任务类型决定了任务在队列中的排队与出队顺序。FIFO 是当你不做任何特殊指定时默认采用的类型其核心语义是任务按照被插入队列的先后顺序被处理先加入的任务先被取出。当向一个Queue实例调用add方法时如果没有显式指定lifo: trueLIFO后进先出或priority优先级等选项任务就走 FIFO 路径import { Queue } from bullmq; const myQueue new Queue(Paint); // 添加一个任务它将在所有已存在的任务之后被处理 await myQueue.add(wall, { color: pink });顺序保证的边界并发下的开始有序、完成无序文档明确指出一个容易被忽视的要点无论你有多少个处理器入队顺序都会被保留但是如果存在多个 worker 或单个 worker 的 concurrency并发度大于 1虽然 worker 会按顺序开始处理任务但任务可能以略有不同的顺序完成因为某些任务可能比另一些耗时更长。换句话说FIFO 保证的是出队/开始处理的顺序而非完成顺序。在设计依赖结果顺序的业务如流水线中的串联步骤时应将其建模为 Flows父子依赖或在工作任务内部自行协调而不是依赖 FIFO 的完成顺序。入队 APIQueue.add与任务选项Queue.add是 FIFO 任务的入队入口其签名定义于 src/classes/queue.tsasync add( name: NameType, // 任务名称如 wall data: DataType, // 任意可 JSON 序列化的数据负载 opts?: JobsOptions, // 影响任务如何处理的任务选项 ): PromiseJob...其中opts的类型JobsOptions继承自BaseJobOptions见 src/interfaces/base-job-options.ts包含一批可选项。文档重点演示了清理相关的两个removeOnComplete与removeOnFail完成/失败后的任务清理await myQueue.add( wall, { color: pink }, { removeOnComplete: true, removeOnFail: 1000 }, );上述示例的含义removeOnComplete: true任务成功完成后立即从队列中移除不保留在 completed 集合中removeOnFail: 1000任务在耗尽所有重试attempts后仍失败时最多保留最近 1000 个失败任务在 failed 集合中更早的失败任务会被清理。removeOnComplete和removeOnFail的完整取值见 src/interfaces/base-job-options.ts取值行为true任务结束后立即移除false默认任务保留在 completed / failed 集合中默认行为数字N保留最近 N 个任务更早的按数量淘汰KeepJobs对象按age存活秒数与count最大数量组合保留可附加limit单次最多删除数量KeepJobs类型的定义位于 src/types/keep-jobs.tsexport type KeepJobs | { count: number } | { age: number; // 保留的最大存活时间秒 count?: number; limit?: number; // 单次最多删除数量 };源码注释中有一个重要的时效性提示当使用age或count时淘汰是基于最佳努力best-effort在每次任务结束时评估的——BullMQ 不会运行后台定时器因此过期的任务只有在后续同类型completed 或 failed任务完成后才会被真正移除。如果队列长时间没有新任务完成已过期的已完成任务仍会保留在集合中。这一点在 src/interfaces/base-job-options.ts 与 src/types/keep-jobs.ts 中均有明确说明。队列级默认defaultJobOptions当队列中的大多数任务共享相同选项时逐个传入既冗长又容易遗漏。BullMQ 允许在实例化Queue时通过defaultJobOptions统一声明const queue new Queue(Paint, { defaultJobOptions: { removeOnComplete: true, removeOnFail: 1000, }, });此后所有通过该队列add/addBulk添加的任务若未显式覆盖都会继承这两项清理策略。其实现位于 src/classes/queue.ts构造函数将opts?.defaultJobOptions ?? {}存入this.jobsOpts而addJobsrc/classes/queue.ts在创建任务时执行合并const mergedOpts { ...this.jobsOpts, // 队列级默认 ...opts, // 单任务显式覆盖 jobId, };即单任务传入的opts优先于defaultJobOptions允许你为个别任务覆盖队列默认值。Queue还提供了只读的defaultJobOptionsgettersrc/classes/queue.ts返回当前默认选项的副本。QueueOptions的完整定义见 src/interfaces/queue-options.ts。底层原理FIFO 顺序如何被保证Redis 后端LPUSHRPOPLPUSH在 Redis 后端FIFO 入队逻辑位于 src/commands/includes/storeAndEnqueueJob.lualocal pushCmd opts[lifo] and RPUSH or LPUSH addJobInTargetList(waitKey, markerKey, pushCmd, isPausedOrMaxed, jobId)无lifo选项时使用LPUSH将 jobId 压入等待列表wait list头部消费者端 src/commands/includes/fetchNextJob.lua 使用RPOPLPUSH从列表尾部取出并原子地移入 active 列表local jobId rcall(RPOPLPUSH, waitKey, activeKey)LPUSH左压配合RPOP右取天然构成一个 FIFO 队列最早LPUSH进来的任务位于列表尾部总是先被RPOPLPUSH取出。整个取任务过程还通过RPOPLPUSH的原子性避免了多个 worker 并发抢到同一个任务的问题。注意优先级任务priority 0走的是独立的 sorted set 路径src/commands/includes/addJobWithPriority.lua在fetchNextJob.lua中wait 列表为空或遇到 marker 时会通过moveJobFromPrioritizedToActive处理且同优先级任务按 FIFO 顺序处理——这一点由测试 tests/worker.test.ts 中的用例should process jobs with the same priority in FIFO order覆盖。PostgreSQL 后端队列表 moveToActive本仓库同时提供了 PostgreSQL 后端实现其 FIFO 语义由后端接口moveToActive承担。测试 tests/postgres/operations.test.ts 直接验证了多任务下的 FIFO 顺序保持it(preserves FIFO order across multiple jobs, async () { const id1 await backend.addJob(makeJob({ name: first }), ); const id2 await backend.addJob(makeJob({ name: second }), ); const [j1] await backend.moveToActive(token); const [j2] await backend.moveToActive(token); expect((j1 as JobJson).id).toBe(id1); expect((j2 as JobJson).id).toBe(id2); });同文件中的用例adds a job (FIFO waiting) and reads it backtests/postgres/operations.test.ts则验证了任务入队后状态为waiting、计数为 1 的基础行为tests/postgres/worker.test.ts 的processes several jobs in FIFO order从 worker 层面再次印证了顺序保证。这说明 FIFO 语义并不依赖 Redis 特有的列表结构而是两种后端共同承诺的抽象行为对应接口见 src/interfaces/queue-backend.ts。相关测试佐证除上述 PostgreSQL 测试外仓库中还有多处 FIFO 相关验证tests/delay.test.tsshould process delayed jobs with exact same timestamps in correct order (FIFO)——延迟时间相同的任务仍按 FIFO 顺序处理tests/rate_limiter.test.ts限流场景下任务仍按 FIFO 顺序获取。小结与实践建议默认即 FIFO不传lifo、不传priority的任务在 BullMQ 中即为 FIFO 任务先入队者先被处理Redis 端由LPUSHRPOPLPUSH实现PostgreSQL 端由moveToActive实现。区分开始顺序与完成顺序多 worker / 高并发下只保证开始处理有序完成时间可能交错需要严格顺序完成的业务请使用 Flows 父子依赖建模。善用removeOnComplete/removeOnFail控制留存支持布尔值、数量与KeepJobsage/count/limit三种形态注意 age/count 为最佳努力式清理没有后台定时器。用defaultJobOptions收敛重复配置队列级默认值会被单任务选项覆盖减少每个add调用中的样板代码。更多任务类型对比可继续阅读同目录下的 lifo.md 与 prioritized.md以理解三种排序策略的适用场景。赞分享后端消息队列任务调度【免费下载链接】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点击查看免费下载相关推荐终极指南如何高效管理downkyicore下载队列轻松调整任务优先级与顺序终极指南如何高效管理downkyicore下载队列轻松调整任务优先级与顺序 downkyicore作为一款强大的哔哩哔哩视频下载工具不仅支持8K、HDR、GoDS优先队列任务调度中的优先级管理策略GoDS优先队列任务调度中的优先级管理策略 你是否还在为任务调度中的优先级混乱而头疼当系统中同时存在紧急任务和普通任务时如何确保关键操作优先执行本文将通后端WingOS输入系统设计PS/2键盘鼠标与现代输入设备支持指南WingOS输入系统设计PS/2键盘鼠标与现代输入设备支持指南 WingOS作为一个采用微内核架构的64位操作系统其 输入系统设计 体现了现代操作系统对硬件上一篇智慧树自动刷课插件终极指南5分钟实现高效学习下一篇终极智慧树学习助手5分钟配置智能刷课插件高效学习省时90%创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考