后端消息队列任务调度【免费下载链接】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点击查看免费下载output_articleBullMQ for .NET 开发实战指南跨 Redis 与 PostgreSQL 的统一分布式队列BullMQ for .NET 是 BullMQ 官方维护的 .NET 移植版——一套基于 Redis 与 PostgreSQL 的快速、健壮的分布式任务队列。与 Node.js、Python、PHP、Elixir、Rust 等运行时共享同一套 LuaRedis与 SQLPostgreSQL脚本保证跨语言的原子性与语义完全一致Node.js 产出的队列可以被 .NET Worker 消费反之亦然。读完本文你将掌握 .NET 端队列/Worker 的完整接入方式、连接共享技巧、PostgreSQL 后端切换方法以及底层的脚本共享机制与开发测试流程。项目定位glue code 之上的跨语言一致性从 dotnet/README.md 的定义来看这个 .NET 绑定本质上是胶水代码glue code它不重新实现队列语义而是直接复用仓库根目录下rawScripts/中的 Lua 脚本以及src/postgres/下的 SQL 脚本。共享脚本正是“跨语言完全一致”的保障——不同语言的运行时执行的是同一份经过实战检验的原子脚本Redis 后端使用与 Node.js 主库 相同的 Lua 脚本如moveToActive-11.lua、addStandardJob-9.luaPostgreSQL 后端使用 src/postgres/commands 下的参数化 SQL 命令如add_job.sql、move_job_to_active.sql等。当前状态README 声明结合 dotnet/src/BullMQ 源码可验证数据库无关的queue-backend 契约已同时为 Redis 与 PostgreSQL 完整实现并有一套共享一致性测试套件同一批测试同时跑在两个适配器上高层的Queue、Worker、Job类在任一后端上均可运行FlowProducer、QueueEvents、JobScheduler亦已实现。选择后端只需在选项中设置Postgres。环境要求.NET 8.0 或更高版本使用 Redis 后端时Redis 6.2或兼容服务器如 Valkey、Dragonfly使用 PostgreSQL 后端时PostgreSQL 13最低版本校验实现在 PostgresConnection.cs 中MinimumPostgresMajor 13启动时会读取server_version_num校验可用SkipVersionCheck跳过。安装通过 NuGet 一键引入dotnet add package BullMQ快速开始队列 Worker 最小可运行示例以下示例完整来自 dotnet/README.md演示了创建队列、入队、启动 Worker 处理并监听完成事件using BullMQ; // 创建队列并添加一个任务。 await using var queue new Queue(emails, new QueueOptions { Connection ConnectionOptions.FromString(localhost:6379), }); await queue.AddAsync(welcome, new { to userexample.com }); // 用 Worker 处理任务。 await using var worker new Worker(emails, async (job, ct) { Console.WriteLine($Processing {job.Name} #{job.Id}); // ... 执行业务逻辑 ... return sent; }, new WorkerOptions { Connection ConnectionOptions.FromString(localhost:6379), Concurrency 5, }); worker.Completed (job, result) Console.WriteLine($Job {job.Id} completed with {result});几个关键点await using确保队列/Worker 随作用域退出被正确释放Queue.AddAsync(name, data)的返回值为Job对象处理器返回的对象会作为任务的完成结果持久化Worker 处理器签名是 Worker.cs 中定义的委托public delegate Taskobject? Processor(Job job, CancellationToken cancellationToken)ct用于协作式取消如强制关闭Worker 事件Active、Completed、Failed、Drained、Stalled、Error、LockRenewalFailed、LocksRenewed等定义在 Worker.cs用于观察与可观测性集成。Worker 的阻塞式取任务与锁机制从 Worker.cs 的注释可以看到底层实现原理任务获取使用后端的阻塞等待原语——Redis 端是对 marker 集合的BZPOPMINPostgreSQL 端是LISTEN因此空闲 Worker不会忙轮询。锁由LockManager周期续期后台还有一个 stalled-job 检查器负责恢复锁已过期的任务。共享连接复用 IConnectionMultiplexer每个队列/Worker 都维护自己的连接但可以通过传入已有的IConnectionMultiplexer在多个队列与 Worker 间复用连接。BullMQ 不会释放它不拥有的连接README 明确说明因此复用后由你自己管理其生命周期var mux await StackExchange.Redis.ConnectionMultiplexer.ConnectAsync(localhost:6379); var options new QueueOptions { Connection new ConnectionOptions { Multiplexer mux }, };对应 Options.cs 中的ConnectionOptions提供ConnectionString、Configuration已解析的ConfigurationOptions或Multiplexer三选一Multiplexer优先级最高另有Database逻辑库索引默认-1与便捷工厂方法FromString(connectionString)。使用 PostgreSQL 后端在选项上设置Postgres即可用完全相同的Queue/Worker/JobAPI 切换到 PostgreSQL。Schema默认bullmq作为所有队列的命名空间所需表与函数会在首次使用时自动迁移创建using BullMQ; using BullMQ.Postgres; var pg new PostgresOptions { ConnectionString Hostlocalhost;Databasebullmq;Usernamepostgres;Passwordpostgres, // Schema bullmq, // 可选此为默认值 }; await using var queue new Queue(emails, new QueueOptions { Postgres pg }); await queue.AddAsync(welcome, new { to userexample.com }); await using var worker new Worker(emails, async (job, ct) sent, new WorkerOptions { Postgres pg, Concurrency 5 });PostgresOptions 与 Schema 语义PostgresOptions.cs 定义了三个选项选项说明默认值ConnectionStringNpgsql 连接字符串如Hostlocalhost;Databasebullmq_test空字符串Schema命名所有队列的 Schema等价于 Redis 键前缀的 SQL 原生替代bullmqSkipVersionCheck跳过启动时的 PostgreSQL 最低版本检查falseSchema 的妙处在于它被钉在连接的search_path上PostgresConnection.cs因此共享的.sql文件全部使用不带 Schema 限定的可移植名称既跨库可移植又天然防注入。Schema 名称必须匹配^[A-Za-z_][A-Za-z0-9_$]*$且不超过 63 字符QuoteSchemaName。自动迁移的实现细节PostgresConnection.cs 展示了迁移的健壮设计迁移使用事务级 advisory lock键为0x42554C4C即字符串BULL所有语言运行时使用完全相同的键串行化跨进程并发启动保证迁移恰好执行一次迁移文件见 src/postgres/migrations按文件名中的数字前缀排序执行记录在bullmq_migration表中迁移后调用ReloadTypesAsync()重新加载 Npgsql 的类型目录迁移会创建如 job-state 枚举等自定义类型每个连接字符串在进程内只迁移一次s_migrated静态集合 静态s_migrateLock避免多实例并发启动时触发 Npgsql 类型目录重载风暴。阻塞等待LISTEN/NOTIFYPostgreSQL 后端使用专用LISTEN连接实现 Worker 的阻塞取任务PostgresConnection.cs订阅bullmq_jobs频道等待 NOTIFY超时或关闭时通过链接的CancellationTokenSource优雅结束不会阻塞进程退出。QueueEvents则订阅bullmq_events频道JobsChannel/EventsChannel常量见 PostgresConnection.cs。核心选项速查从源码注释出发的参数手册Options.cs 是全部高层类共享的选项定义以下参数均可在配置时直接使用QueueBaseOptionsQueue / Worker / QueueEvents 共用ConnectionRedis 后端连接设置默认后端PrefixRedis 键前缀默认bull与主库一致PostgresPostgreSQL 后端设置。一旦设置队列/Worker 改用 PostgreSQL且Connection被忽略。QueueOptionsDefaultJobOptions合并进该队列每个任务的默认JobsOptionsSkipMetasUpdate为true时启动不写队列 meta hash。从 Queue.cs 可见默认写入opts.maxLenEvents 10000与version bullmq:{Version.Value}。WorkerOptions参数默认值说明Concurrency1并发处理的最大任务数构造时校验必须 1LockDuration30000任务处理中的锁时长毫秒LockRenewTimeLockDuration / 2锁续期频率毫秒DrainDelay5Worker 循环前阻塞等待任务的秒数Namenull可读的 Worker 名称用于可观测性Autoruntrue创建后是否立即开始处理MaxStalledCount1任务在判定失败前可从 stalled 恢复的最大次数StalledInterval30000执行 stalled 检查的间隔毫秒SkipStalledCheckfalse禁用 stalled 检查器SkipLockRenewalfalse禁用周期性锁续期在 Worker.cs 中可以确认构造时若既无Postgres也无Connection会抛出ArgumentExceptionConcurrency 1抛ArgumentOutOfRangeExceptionAutorun为true默认时立即启动后台消费循环。QueueEventsOptionsAutorun默认true创建后立即开始消费事件LastEventId起始游标默认$只接收监听开始后的事件可传入已知事件 id 恢复BlockingTimeout默认10000每次对事件流的阻塞读超时毫秒。JobsOptions单任务配置JobId显式任务 id不能为0或以0:开头Queue.cs 会在AddAsync时校验Delay任务变为可用前的延迟毫秒Priority任务优先级1最高0或null表示无优先级Attempts任务完成前的总尝试次数Backoff自动重试的退避配置BackoffOptionsType如fixed/exponentialDelay为毫秒基数Lifo为true时任务加入等待列表右侧后进先出Timestamp创建时间戳毫秒默认当前时间RemoveOnComplete/RemoveOnFail完成后/失败后移除策略——true移除、数字表示保留条数、或KeepJobs策略对象KeepLogs任务保留的最大日志条数0表示不限。KeepJobs支持Count保留最大数量与Age保留最大秒数两种裁剪维度。高可用语义Queue 的管理能力Queue.cs 提供了与主库对齐的管理 APIPauseAsync()/ResumeAsync()/IsPausedAsync()全局暂停/恢复队列处理GetJobCountsAsync(params string[] types)按状态统计数量未传参时默认统计waiting, active, completed, failed, delayed, paused六种状态GetWaitingCountAsync()等待处理的任务数GetJobAsync(jobId)按 id 获取任务AddAsync(name, data, opts)入队并返回Job。开发与测试共享脚本的生成与验证流程README 的 Development 一节说明了共享脚本的组织方式dotnet/README.md仓库根目录rawScripts/下的 Lua 脚本会被复制到dotnet/src/BullMQ/Commands/嵌入程序集src/postgres/**会被复制到dotnet/src/BullMQ/Postgres/这些副本都是生成的、被 git 忽略的SqlLoader.cs 通过程序集内嵌资源加载命令与迁移均以内嵌.sql资源形式提供缺副本时会提示运行copy:sql:dotnet。在仓库根目录执行# 生成脚本副本需先执行一次 yarn install。 yarn copy:lua:dotnet yarn copy:sql:dotnet # 构建与测试。 cd dotnet dotnet build dotnet test # 需要 localhost:6379 上的 Redis 服务器或直接使用一键辅助脚本 dotnet/scripts/test.sh它会自动补全缺失的脚本副本、设置默认测试连接并把任意参数直接透传给dotnet testcd dotnet ./scripts/test.sh # 全量测试Redis PostgreSQL ./scripts/test.sh --filter Name~Flow # 只跑子集测试连接可通过环境变量指向其他服务器BULLMQ_TEST_REDIS默认localhost:6379BULLMQ_TEST_POSTGRES默认Hostlocalhost;Databasebullmq_test;Username$USER测试脚本还处理了dotnet不在 PATH 的场景回退到/opt/homebrew/opt/dotnet/libexec并导出DOTNET_CLI_TELEMETRY_OPTOUT1、DOTNET_NOLOGO1保持输出干净。共享一致性测试套件对应 dotnet/tests/BullMQ.Tests如BackendConformanceTests.cs同时针对两个适配器运行这正是 README 所述“同一套测试验证两个后端”的落地体现。跨语言互操作的实战意义由于所有运行时共享同一批 Lua/SQL 脚本.NET 可以消费 Node.js/Python/Elixir/Rust/PHP 生产者入队的任务反向为其他语言的 Worker 提供任务在 Redis 与 PostgreSQL 之间切换而不改业务代码仅改选项配置。这意味着团队可以按服务选择语言而队列语义、原子性、幂等与重试行为保持一致。从仓库的 README.md根目录可见BullMQ 生态覆盖 NodeJS、Python、.NET、Elixir、Rust、PHP 六种运行时.NET 绑定正是其中一环。许可证BullMQ for .NET 采用 MIT 许可证详见仓库根目录 LICENSE。 /output_article赞分享后端消息队列任务调度【免费下载链接】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 实战指南基于 Redis/PostgreSQL 的多语言分布式任务队列全面解析BullMQ 实战指南基于 Redis/PostgreSQL 的多语言分布式任务队列全面解析 本指南以 BullMQ 仓库根目录 README.md http后端消息队列任务调度Bull 队列完全指南基于 Redis 的 Node.js 分布式任务与消息队列实战Bull 队列完全指南基于 Redis 的 Node.js 分布式任务与消息队列实战 本文围绕开源仓库 bull一个基于 Redis 的 Node.js 任任务调度后端BullMQ 是什么基于 Redis 的分布式任务队列核心特性与设计原理BullMQ 是什么基于 Redis 的分布式任务队列核心特性与设计原理 本文以官方文档《What is BullMQ》为核心骨架结合仓库源码src 目录后端消息队列任务调度上一篇NemoClaw 发布上下文检查定位并审计最新全量 E2E main 运行的完整实践指南下一篇PyTextRank 教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考