AigoTools任务队列设计全解析BullRedis实现站点抓取的重试、并发与优雅停止【免费下载链接】aigotoolsAigoTools can help users quickly create and manage website directory, with built-in site auto-crawling features, and also provides internationalization, SEO, image storage, and other functions. It allows users to quickly deploy their own directory site online.项目地址: https://gitcode.com/gh_mirrors/ai/aigotoolsAigoTools 是一款帮助你快速创建和管理网站目录的开源工具内置站点自动抓取功能并支持国际化、SEO 与图片存储等能力。它的crawler服务基于NestJS Bull Redis构建了一套站点抓取任务队列提交一个站点后后台自动完成页面截图、AI 摘要、分类推荐整个过程具备失败重试、并发控制与优雅停止三大关键能力。本文将完整拆解这套队列是如何设计的帮助你在自己的项目中复刻类似的可靠抓取系统。一、为什么站点抓取需要任务队列AigoTools 抓取一个站点并不只是请求一下页面这么简单每个任务要完成用 Playwright 打开页面并截图耗时几秒到几十秒不等抓取页面正文内容调用 AI 模型生成站点摘要、关键词、分类推荐两次 LLM 调用将结果写回 MongoDB 并更新站点状态这类任务有典型的三高一长特点耗时不定、失败率高网站打不开、AI 接口超时、资源占用高、用户等待久。如果走同步接口一个慢站点就能把请求线程卡死批量提交更不可行。引入任务队列后整个流程变成提交即返回管理后台 → HTTP /dispatch → Bull 队列Redis 存储 → 并发消费 → 重试 → 写库Redis 在这里身兼两职既是Bull 队列的任务持久化存储也承担了抓取过程中的内容缓存截图与正文缓存 10 分钟避免重复抓取。Redis 连接封装在 redis.service.ts 中并打印连接生命周期日志方便排查断线问题。二、队列整体架构Module、Producer、Consumer 三件套AigoTools 把队列代码集中在packages/crawler/src/site-queue/目录下职责划分非常清晰文件职责site-queue.module.ts注册队列与 Bull Board 监控面板site-queue.producer.ts生产端添加、批量添加、停止任务site-queue.consumer.ts消费端执行抓取、处理失败site-queue.constant.ts队列名与任务名常量队列注册两行代码接入 Bull 监控面板在 site-queue.module.ts 中通过BullModule.registerQueueAsync注册名为site-queue的队列同时用BullBoardModule.forFeature把该队列挂到可视化监控面板上——这一步让后续看队列变得零成本。全局 Redis 连接配置在 app.module.ts 的BullModule.forRootAsync中完成host/port/password/db全部来自环境变量部署时只需改配置。生产端单个与批量入队生产端 site-queue.producer.ts 只有三个方法却覆盖了队列最常用的三类操作addCrawlJob(siteId)单个站点入队batchAddCrawlJobs(siteIds)用queue.addBulk一条命令批量入队提交 100 个站点时比循环调用 100 次add高效得多stopCrawlJob / batchStopCrawlJob按站点 ID 精确移除任务详见第四节入队时附带了两个关键选项await this.siteQueue.add(SITE_CRAWL_JOB, siteId, { attempts: 3, // 最多尝试 3 次 backoff: 10000, // 失败后间隔 10 秒重试 });消费端Processor 声明式消费消费端 site-queue.consumer.ts 使用Processor(SITE_QUEUE_NAME)声明处理器crawlSite方法负责真正的抓取逻辑。任务数据job.data只存站点 ID而非整个站点对象——消费时再从数据库findById读取最新数据这避免了队列里存旧数据的一致性问题。抓取过程本身也有性能设计截图与 AI 摘要两个耗时操作通过Promise.all并行执行见 site-queue.consumer.ts单任务耗时约等于两者中较慢的那个而不是两者之和。三、重试设计失败自动重试 3 次彻底失败才标记新手最常问的问题网络抖动导致的偶发失败为什么要让用户手动重新点一遍AigoTools 的答案写在入队选项里——attempts: 3backoff: 10000即每次失败后间隔 10 秒自动重试最多 3 次。但重试耗尽之后的收尾工作很关键。消费端定义了失败钩子OnQueueFailed({ name: SITE_CRAWL_JOB }) async handleCrawlJobFailed(job: Jobstring) { ... }在 site-queue.consumer.ts 中逻辑分两层还有重试次数时直接返回attemptsMade attempts把任务交给 Bull 自动重试不做任何状态变更重试彻底耗尽时才将 MongoDB 中该站点的processStage更新为failprocessStage是一个四阶段状态机site.schema.tspending待处理→processing处理中→success成功/fail失败这样后台列表页就能按状态筛选抓取失败的站点一键重新派发。整个重试状态由 Bull 维护业务代码只需关心最终失败这一个时刻职责边界非常干净。四、并发控制10 个任务同时跑速度提升 10 倍Bull 默认一次只执行一个任务。对于截图 AI 摘要这种无 CPU 密集计算、纯等待网络 I/O的任务串行执行是巨大的浪费。AigoTools 在Process装饰器上指定了并发数Process({ name: SITE_CRAWL_JOB, concurrency: 10 }) async crawlSite(job: Jobstring) { ... }见 site-queue.consumer.ts。concurrency: 10表示同一个队列同时跑 10 个抓取任务对批量导入 100 个站点的场景吞吐直接提升一个数量级。这个数值其实是调出来的平衡点太小 → 批量任务排队太久太大 → Playwright 浏览器实例与 AI 接口配额被撑爆如果你的任务依赖下游服务建议从 3~5 起步逐步上调。五、优雅停止移除任务 拦截写库双保险防止半截数据停止抓取看似只是删掉队列里的任务但有一个隐蔽的坑任务正在执行中时你停止它之后它可能还会把半成品数据写进数据库比如截图成功了、AI 摘要失败此时把processStage写成 success 就错了。AigoTools 用两道防线解决防线一按状态批量移除队列任务site-queue.producer.ts 中的stopCrawlJob依次检查三种状态的任务并全部移除active正在执行的waiting等待调度的delayed因重试而延迟等待的批量版本batchStopCrawlJob用Promise.allSettled执行移除单个任务移除失败不会中断其他任务停止操作的健壮性更好。防线二写库前校验任务是否仍然有效更精妙的是assertJobActivesite-queue.consumer.ts消费逻辑的finally块在保存数据之前会重新从队列查询该任务是否仍处于active状态——如果用户刚刚点了停止任务已被移除这里直接抛错中断写库保证被停止的任务不会留下脏数据。上层服务 app.service.ts 则把停止队列任务和数据库状态重置组合成原子操作移除任务的同时把站点processStage从processing回滚为pending让用户随时可以重新派发。整个停止链路对外暴露为管理后台的两个按钮——单站点停止和筛选后的批量停止前端调用见 actions.ts。六、可观测性内置队列监控面板问题定位不靠猜很多团队的队列看不见摸不着出了问题只能翻日志。AigoTools 在 app.module.ts 中注册了Bull Board监控面板挂载在/queues路由下并套上 Basic Auth 中间件保护basic-auth.middleware.ts。打开面板后你能直观看到每个队列的active / waiting / delayed / failed / completed任务数量单个任务的执行次数与最近一次报错直接对失败任务执行重试、删除等操作配合抓取接口本身的限流配置每分钟 200 次的 Throttler 规则入队、执行、停止、监控四个环节都有了抓手——这正是生产级任务队列与内存里跑个数组的本质区别。七、核心文件清单与总结模块路径关键作用队列模块site-queue.module.ts注册队列 监控面板任务生产site-queue.producer.ts入队、批量入队、移除任务任务消费site-queue.consumer.ts抓取执行、并发、失败钩子派发/停止服务app.service.ts状态机流转站点状态定义site.schema.ts四阶段 processStage前端调用actions.ts管理后台一键派发/停止总结一下这套设计的可复用经验✅入队即配置重试attempts backoff两行选项解决 90% 的偶发失败✅并发数显式声明I/O 密集型任务用concurrency放大吞吐✅失败钩子只做最终失败收尾重试中间态不污染业务数据✅停止 移除任务 写库前校验 状态回滚三步配合杜绝脏数据✅Bull Board 面板让队列状态从黑盒变成白盒如果你也在用 Next.js / NestJS 构建带后台抓取、邮件发送、报告生成等异步任务的产品这套Module Producer Consumer 监控面板的目录结构完全可以照搬。动手前建议先克隆仓库对照packages/crawler/src/site-queue/目录逐个文件阅读200 多行代码就能读完整个队列实现是学习 Bull 实战的绝佳样本 【免费下载链接】aigotoolsAigoTools can help users quickly create and manage website directory, with built-in site auto-crawling features, and also provides internationalization, SEO, image storage, and other functions. It allows users to quickly deploy their own directory site online.项目地址: https://gitcode.com/gh_mirrors/ai/aigotools创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考