当前位置：首页 > news >正文

Bull装饰器完全指南：@Processor、@OnQueueEvent与依赖注入

news 2026/8/2 22:51:55

Bull装饰器完全指南：@Processor、@OnQueueEvent与依赖注入

【免费下载链接】bullBull module for Nest framework (node.js) :cow:项目地址: https://gitcode.com/gh_mirrors/bul/bull

Bull是Nest框架中用于处理队列任务的强大模块，通过装饰器可以轻松实现任务处理、事件监听和依赖注入等核心功能。本文将详细介绍@Processor、@OnQueueEvent装饰器的使用方法及依赖注入技巧，帮助开发者快速掌握Bull在Nest项目中的应用。

一、@Processor装饰器：定义队列处理器

@Processor装饰器用于将类标记为队列处理器，使其能够处理指定队列中的任务。它支持多种配置方式，满足不同场景需求。

1.1 基础使用方式

最简单的用法是直接指定队列名称：

@Processor('user-queue') export class UserProcessor { // 任务处理方法 }

这种方式会创建一个与指定队列关联的处理器，用于处理该队列中的所有任务。

1.2 高级配置选项

@Processor装饰器还支持传入ProcessorOptions对象进行更详细的配置：

@Processor({ name: 'email-queue', scope: Scope.REQUEST, configKey: 'email-queue-config' }) export class EmailProcessor { // 任务处理方法 }

其中，name指定队列名称，scope设置处理器的生命周期作用域，configKey用于指定配置键，以便从配置中获取队列连接信息。

1.3 结合Worker选项

除了基础配置外，还可以传入WorkerOptions来设置Bull的工作选项：

@Processor('image-queue', { concurrency: 5, removeOnComplete: true }) export class ImageProcessor { // 任务处理方法 }

这里的concurrency设置并发处理数量，removeOnComplete设置任务完成后是否自动删除。

二、@OnQueueEvent装饰器：监听队列事件

@OnQueueEvent装饰器用于注册队列事件监听器，使方法能够响应队列的各种事件。

2.1 常用事件类型

Bull队列支持多种事件，如任务完成、失败、重试等：

@Processor('order-queue') export class OrderProcessor { @OnQueueEvent('completed') handleCompleted(job: Job) { console.log(`Job ${job.id} completed successfully`); } @OnQueueEvent('failed') handleFailed(job: Job, error: Error) { console.error(`Job ${job.id} failed: ${error.message}`); } }

上述代码分别监听了任务完成和失败事件，并在事件发生时执行相应的处理逻辑。

2.2 事件处理方法

事件处理方法会接收事件相关的参数，如job对象和错误信息等。通过这些参数，可以获取任务的详细信息并进行相应的处理。

三、依赖注入：@InjectQueue装饰器

@InjectQueue装饰器用于在类中注入队列实例，以便直接操作队列。

3.1 注入默认队列

如果只有一个队列或需要注入默认队列，可以不指定名称：

export class OrderService { constructor( @InjectQueue() private orderQueue: Queue ) {} async createOrder(data: OrderData) { await this.orderQueue.add('create-order', data); } }

3.2 注入指定队列

当项目中有多个队列时，可以通过名称指定要注入的队列：

export class NotificationService { constructor( @InjectQueue('email-queue') private emailQueue: Queue, @InjectQueue('sms-queue') private smsQueue: Queue ) {} async sendEmail(data: EmailData) { await this.emailQueue.add('send-email', data); } async sendSms(data: SmsData) { await this.smsQueue.add('send-sms', data); } }

四、实际应用示例

4.1 完整的处理器示例

下面是一个包含任务处理和事件监听的完整处理器示例：

@Processor('report-queue', { concurrency: 3 }) export class ReportProcessor { constructor( private reportService: ReportService ) {} @Process('generate-report') async handleGenerateReport(job: Job<ReportData>) { const result = await this.reportService.generate(job.data); return result; } @OnQueueEvent('completed') handleCompleted(job: Job) { console.log(`Report ${job.data.reportId} generated successfully`); } @OnQueueEvent('failed') handleFailed(job: Job, error: Error) { console.error(`Report generation failed: ${error.message}`); } }

4.2 队列模块配置

在Nest模块中配置Bull队列：

@Module({ imports: [ BullModule.registerQueue({ name: 'report-queue', redis: { host: 'localhost', port: 6379, }, }), ], providers: [ReportProcessor, ReportService], }) export class ReportModule {}