异步任务队列系统

异步任务队列系统

学习目标

• 理解 BullMQ + RabbitMQ 双队列架构的设计动机
• 掌握任务优先级与并发控制的实现
• 了解 Worker 锁扩展与优雅关机机制

前置知识

• Redis 基本操作
• 消息队列概念

---

项目实践

双队列架构

Firecrawl 使用两套队列系统并行运行：

为什么需要两套队列？

• BullMQ：适合低频、复杂的后台任务（Deep Research、LLMs.txt 生成等）。BullMQ 提供丰富的任务生命周期管理（重试、延迟、优先级）。
• NUQ（新一代队列系统）：专门为高频抓取任务设计，基于 RabbitMQ。抓取任务吞吐量高（每分钟数百次），BullMQ 的 Redis 存储在高并发下成为瓶颈。

队列分类

队列	后端	用途	TTL
generateLlmsTxtQueue	Redis	生成站点的 LLMs.txt 文件	25 小时
deepResearchQueue	Redis	深度研究任务	25 小时
billingQueue	Redis	计费检查（高频短 TTL）	1 分钟完成 / 1 小时失败
precrawlQueue	Redis	预爬取分析	24 小时
scrapeQueue (NUQ)	RabbitMQ	单页面抓取	5 分钟超时
crawlGroup (NUQ)	RabbitMQ	网站爬取任务组	可取消

任务优先级与团队并发控制

每个抓取任务通过 getJobPriority 计算优先级：

优先级 = 基础优先级 + 信用额度加成 + 团队并发信号量

团队信号量（teamConcurrencySemaphore）：每个团队有并发抓取数量上限。当达到上限时，新任务进入等待状态而非被拒绝。信号量的实现基于 Redis 计数器：

团队 A 并发上限: 10
当前运行: 8 → 允许新任务
当前运行: 10 → 新任务排队
任务完成: 9 → 排队任务启动

Worker 锁扩展

长任务（如 Deep Research 可能运行数分钟）面临锁超时问题。BullMQ 的锁默认在任务完成后自动释放，但如果任务运行时间超过锁超时，锁会被释放，其他 Worker 可能抢占同一任务。

Firecrawl 使用定时锁扩展机制：

const extendLockInterval = setInterval(async () => {
  logger.info(`Worker extending lock on job ${job.id}`);
  await job.extendLock(token, jobLockExtensionTime);
}, jobLockExtendInterval);

在任务执行期间，每隔一段时间扩展一次锁的超时时间。任务完成后，clearInterval 停止扩展。

NUQ 与 Crawl 任务组

NUQ 的爬取任务（Crawl）按任务组组织：

Crawl ID: abc-123
    ├── Scrape: /page1
    ├── Scrape: /page2
    ├── Scrape: /about
    └── Scrape: /pricing

每个 Scrape 任务是独立的 NUQ 任务，但通过 crawlGroup 关联。爬取完成后，finishCrawlSuper 处理任务组的最终状态（生成完成报告、触发 Webhook）。

Webhook 回调

任务完成时，系统通过 Webhook 通知客户端：

任务完成 → Webhook 队列 → POST 到客户 URL → 重试 3 次

Webhook 发送是异步的，不阻塞任务完成状态更新。

问题与规避

1. Redis 连接泄漏

getRedisConnection 使用单例模式创建 IORedis 连接。如果连接异常断开，所有依赖 Redis 的操作（限流、队列、缓存）都会失败。

规避：IORedis 自动重连机制。连接错误记录为 WARN 级别日志，不中断进程。

2. 队列积压

如果 Worker 处理速度跟不上任务入队速度，队列会积压大量待处理任务。对于 BullMQ，积压任务存储在 Redis 中，占用内存。

规避：

• removeOnComplete / removeOnFail 配置自动清理已完成/失败的任务
• 系统监控（system-monitor）检测 CPU/内存使用率，超过阈值时拒绝新任务

3. 任务重复执行

同一 Crawl 可能因为 Worker 重启而被重复处理。

规避：Crawl 状态存储在 Redis 中（crawl:id），新 Worker 启动时检查是否有进行中的 Crawl 任务（reconcileRunningMonitorChecks）。

设计取舍

为什么引入 NUQ（RabbitMQ）而不是只用 BullMQ？

BullMQ 的所有任务数据存储在 Redis 中。对于高频抓取场景（每秒数十次抓取），Redis 的内存和网络 I/O 成为瓶颈。RabbitMQ 专为高吞吐量消息队列设计，能够处理更大的并发量。

代价是运维复杂度增加 —— 需要同时维护 Redis 和 RabbitMQ 两套消息基础设施。

为什么用定时锁扩展而非设置超长锁超时？

定时锁扩展（每 X 秒延长 Y 秒）比一次性设置超长锁超时有两个优势：

1. 安全：如果 Worker 崩溃，锁会在最后一次扩展的超时后自动释放，而不是等待很长时间
2. 灵活：可以根据任务实际进度动态调整锁超时

参考来源

• Firecrawl 源码: queue-service.ts
• Firecrawl 源码: queue-worker.ts
• BullMQ 文档