企业官网建设流程全解析

• 后端
• Web框架

本文以 NestJS 官方示例sample/26-queues为对象，系统讲解如何在 NestJS 应用中引入 Redis 与 Bull 队列：从全局队列配置、队列注册、生产者入队、消费者处理器编写，到本地启动与单元/端到端测试验证的完整闭环。读完本文，你将掌握BullModule.forRoot/BullModule.registerQueue/@Processor/@Process/@InjectQueue等核心 API 的用法，并能独立把示例跑起来、用测试确认任务入队与消费行为。

示例概览：这个示例在做什么

sample/26-queues是 NestJS 官方示例中演示基于 Bull 的队列任务处理的标准项目。它的业务场景非常简单而典型：通过 HTTP 接口提交一个"音频转码"任务，任务被投递到名为audio的队列中，由后台消费者异步执行（示例中消费端仅以日志模拟转码过程）。

从源码结构看，示例的核心文件如下：

• 队列全局配置与模块装配 ——BullModule.forRoot连接 Redis，注册audio队列
• 队列模块定义 —— 声明队列、控制器与处理器
• 生产者控制器 —— 提供POST /audio/transcode接口入队
• 消费者处理器 —— 通过@Processor/@Process消费任务
• 应用入口 —— 标准NestFactory.create引导，监听 3000 端口
• Redis 编排文件 —— 一键启动本地 Redis
• 单元测试与 e2e 测试 与 e2e 目录 —— 验证入队与消费行为

从 package.json 可以看到本示例锁定的技术栈版本：

注意：示例中package.json声明了"type": "module"，源码文件使用 ESM 风格导入（例如import { AudioController } from './audio.controller.js'），这是当前示例的运行前提。

环境准备：Redis 与依赖安装

Bull 队列依赖 Redis 作为后端存储与消息通道，因此运行本示例的第一步是准备一个可用的 Redis 实例。示例自带 docker-compose.yml，内容如下：

version: "3" services: redis: image: redis:alpine ports: - 6379:6379

启动 Redis：

docker compose up -d

随后安装依赖（对应 README 的 Installation 步骤）：

$ npm install

依赖安装完成后，即可进入队列配置与业务代码的研读环节。

全局队列配置：BullModule.forRoot 与 Redis 连接

在 app.module.ts 中，根模块通过BullModule.forRoot完成 Redis 连接配置，这是整个队列体系的地基：

import { Module } from '@nestjs/common'; import { BullModule } from '@nestjs/bull'; import { AudioController } from './audio/audio.controller.js'; import { AudioProcessor } from './audio/audio.processor.js'; @Module({ imports: [ BullModule.forRoot({ redis: { host: process.env.REDIS_HOST || 'localhost', port: parseInt(process.env.REDIS_PORT || '6379'), }, }), BullModule.registerQueue({ name: 'audio', }), ], controllers: [AudioController], providers: [AudioProcessor], }) export class AppModule {}

从源码可以提炼出两点实用配置信息：

• 连接地址通过环境变量注入：host取REDIS_HOST（默认localhost），port取REDIS_PORT（默认6379）。这意味部署时无需改代码，只需设置环境变量即可指向远端 Redis。
• BullModule.forRoot是全局配置：它在根模块注册，为整个应用范围内的所有 Bull 队列提供统一的 Redis 连接；而BullModule.registerQueue则用于声明具体队列。

注册队列并装配生产者与消费者：audio.module

业务模块 audio.module.ts 完成了"队列 → 控制器 → 处理器"的装配：

import { BullModule } from '@nestjs/bull'; import { Module } from '@nestjs/common'; import { AudioController } from './audio.controller.js'; import { AudioProcessor } from './audio.processor.js'; @Module({ imports: [ BullModule.registerQueue({ name: 'audio', }), ], controllers: [AudioController], providers: [AudioProcessor], }) export class AudioModule {}

这里BullModule.registerQueue({ name: 'audio' })完成两件事：一是向容器注册名为audio的 Bull 队列实例；二是为其生成可注入的 Token。该 Token 由@nestjs/bull提供的getQueueToken('audio')函数获得——这一点在 audio.controller.spec.ts 的测试替身（mock）中有直接印证：

providers: [ { provide: getQueueToken('audio'), useValue: mockQueue, }, ],

也就是说，凡是需要注入audio队列的地方，本质上都是按getQueueToken('audio')这个 Token 去容器里取实例。

生产者：@InjectQueue 注入队列并投递任务

audio.controller.ts 展示了标准的生产者写法——通过@InjectQueue('audio')注入队列，在路由处理器中调用queue.add(name, data)投递任务：

import { InjectQueue } from '@nestjs/bull'; import { Controller, Post } from '@nestjs/common'; import { Queue } from 'bull'; @Controller('audio') export class AudioController { constructor(@InjectQueue('audio') private readonly audioQueue: Queue) {} @Post('transcode') async transcode() { await this.audioQueue.add('transcode', { file: 'audio.mp3', }); } }

关键点拆解：

• @InjectQueue('audio')：注入参数化队列实例，类型为 Bull 的Queue。
• queue.add('transcode', { file: 'audio.mp3' })：第一个参数是任务名（job name），第二个参数是任务携带的数据负载。这里的任务名transcode必须与消费者端@Process声明的任务名一致，才能被正确路由消费。
• 路由：@Controller('audio')+@Post('transcode')组合出POST /audio/transcode接口；处理函数为async并await入队结果，保证响应与入队完成同步。

消费者：@Processor 与 @Process 异步消费任务

audio.processor.ts 是队列的消费者端：

import { Process, Processor } from '@nestjs/bull'; import { Logger } from '@nestjs/common'; import { Job } from 'bull'; @Processor('audio') export class AudioProcessor { private readonly logger = new Logger(AudioProcessor.name); @Process('transcode') handleTranscode(job: Job) { this.logger.debug('Start transcoding...'); this.logger.debug(job.data); this.logger.debug('Transcoding completed'); } }

• @Processor('audio')：类级装饰器，声明该类是audio队列的消费者；@nestjs/bull会在应用启动时自动把该类实例挂载到对应队列上开始监听。
• @Process('transcode')：方法级装饰器，将handleTranscode注册为transcode任务的处理器。处理器接收 Bull 的Job对象，通过job.data读取入队时携带的数据（本例为{ file: 'audio.mp3' }）。若省略任务名（即@Process()），则该方法会处理该队列中的全部任务。
• 示例消费端仅用Logger.debug打印"开始转码→任务数据→转码完成"三段日志来模拟真实业务；在实际项目中，把这段日志替换为 ffmpeg 调用等重活即可，处理完成即代表任务被成功消费。

至此，示例的完整调用链是：

POST /audio/transcode ↓ queue.add('transcode', { file: 'audio.mp3' }) Bull 队列（Redis 中转） ↓ @Process('transcode') 路由 AudioProcessor.handleTranscode(job) → 模拟转码日志

运行应用：三种启动模式

Redis 就绪且依赖安装完成后，即可按 README 的 Running the app 部分启动应用：

# development（开发模式） $ npm run start # watch mode（监听文件变更自动重启） $ npm run start:dev # production mode（生产模式，编译产物直启） $ npm run start:prod

从 package.json 的 scripts 可以看到这三个命令的真实含义：

• npm run start→nest start
• npm run start:dev→nest start --watch
• npm run start:prod→node dist/main（即先构建、再以 Node 直接运行编译产物）

应用默认监听3000端口（见 main.ts）。启动后，向POST http://localhost:3000/audio/transcode发送请求即可向队列投递一个转码任务，随后在终端观察AudioProcessor打印的消费日志。也可在应用内配置REDIS_HOST/REDIS_PORT环境变量以连接非本机 Redis。

测试验证：单元测试与端到端测试

示例为队列的生产端和消费端各提供了一套测试，对应 README 的 Test 部分：

# unit tests（单元测试） $ npm run test # e2e tests（端到端测试） $ npm run test:e2e # test coverage（覆盖率） $ npm run test:cov

对应 scripts 分别为vitest run、vitest run --config ./vitest.config.e2e.mts与vitest run --coverage（本示例当前使用 Vitest 而非 Jest，仓库同时保留了 jest.json 供参考）。

生产者单元测试（audio.controller.spec.ts）：通过getQueueToken('audio')注入 mock 队列，验证transcode()方法以('transcode', { file: 'audio.mp3' })参数调用queue.add，并校验调用次数与数据结构。

消费者单元测试（audio.processor.spec.ts）：以vi.spyOn(Logger.prototype, 'debug')监控日志输出，传入伪造的Job对象，验证处理器的三段日志顺序（Start transcoding...→job.data→Transcoding completed），以及连续处理多个任务时日志按任务数成倍增长。

端到端测试（audio.e2e-spec.ts）：基于supertest对真实 HTTP 服务发起请求，验证：

• POST /audio/transcode返回201（任务成功入队）；
• 连续 3 个并发请求均返回201（队列可承受并发入队）；
• GET /audio/transcode返回404（路由仅接受 POST 方法）。

这套测试既验证了入队接口的可用性，也间接验证了队列模块在真实 NestJS 容器中的装配正确性。

小结

sample/26-queues是一个"麻雀虽小、五脏俱全"的队列集成范例：用BullModule.forRoot建立 Redis 连接，用BullModule.registerQueue声明队列，用@InjectQueue在生产端投递任务，用@Processor/@Process在消费端异步处理，并配套完整的单元测试与 e2e 测试。以此为基础，你可以把示例中的日志模拟替换为真实的异步业务（如转码、发信、数据清洗），并进一步研究 Bull 的延迟任务、重试、优先级、并发与事件监听等进阶能力。

• 后端
• Web框架