在 Typescript 中实现 Bull Queue

2022-01-21 00:00:00 queue message-queue javascript typescript job-scheduling

我尝试在 Typescript 和 NestJS 中实现 Bull 队列，我的代码:

I try to implement Bull queue in Typescript and NestJS, my code:

@Injectable()
export class MailService {
    constructor(
        @InjectQueue('mail')
        private readonly mailQueue: Queue
    ) {}

    async addToQueue(): Promise<void> {
        this.mailQueue.add(() => {
            return this.sendMail(); 
        })
    }
    
    
    async sendMail(): Promise<void> {

        //logic to implement

        this.addToQueue();
    }
}

快速问题:这个实现是否足以让我的工作排队工作?如果没有:我必须做什么?

fast question: Is this implementation sufficient for my job queuing to work?, If not: what i must to do?

推荐答案

我最近写了一篇博文，似乎与您的用例有关:

I recently wrote a blog post that seems to relate to your use-case:

https://firxworx.com/blog/coding/nodejs/email-module-for-nestjs-with-bull-queue-and-the-nest-mailer/

一些提示:

  • 在您的模块中，请务必导入您的 BullModule(来自 @nestjs/bull).例如，您需要配置您的队列名称(在您的情况下为邮件")并设置您的队列.常见的设置包括使用 redis 主机名和端口进行配置.
  • 在您的服务中，您需要将 jobs 以及可选的 payload 添加到队列中.在您的情况下，您正在尝试添加一个功能.相反，您应该添加一个作业名称，例如confirmationEmail"，并传递一个有效负载，例如user和token.我的示例如下所示:await this.mailQueue.add('confirmationEmail', { user, token })
  • 您需要为您的队列实现一个处理器.这是一个用 @nestjs/bull 中的 @Processor(QUEUE_NAME) 装饰器装饰的类(在您的情况下为 @Processor('mail')).处理器处理添加到队列中的作业.
  • 在您的处理器中，您可以实现一个方法，例如sendConfirmationEmail() 处理名为confirmationEmail"的作业.您可以使用 @Process(JOB_NAME) 来装饰该方法，例如@Process('confirmationEmail').该方法可以接收您的有效负载.根据我的示例，以下方法签名将提供 user 和 token:async sendConfirmationEmail(job: Job<{ user: User, token: string }>): Promise<any> (注意 Job 来自 bull 包，你可能希望输入你的回报与使用 any).这是您实际发送电子邮件的地方.
  • 在您的处理器类中，@nestjs/bull 还提供了特殊的方法装饰器，包括 @OnQueueActive()、@OnQueueCompleted()、<代码>@OnQueueFailed().请参阅文档，但您可能会发现这些对日志记录或其他用途很有用.
  • In your module, be sure to import your BullModule (from @nestjs/bull). For example, you need to configure with your queue name ("mail" in your case) and setup your queue. A common setup would include configuring with the redis hostname and port.
  • In your service, you need to add jobs to the queue, along with optional payload. In your case, you are trying to add a function. Instead, you should add a job name, e.g. "confirmationEmail", and pass a payload, e.g.user and token. My example would look like this: await this.mailQueue.add('confirmationEmail', { user, token })
  • You need to implement a processor for your queue. This is a class decorated with the @Processor(QUEUE_NAME) decorator from @nestjs/bull (@Processor('mail') in your case). The processor handles jobs that are added to the queue.
  • In your processor, you could implement a method e.g. sendConfirmationEmail() that handles the job named "confirmationEmail". You would decorate that method with @Process(JOB_NAME), e.g. @Process('confirmationEmail'). The method can receive your payload. Per my example, the following method signature would provide the user and token: async sendConfirmationEmail(job: Job<{ user: User, token: string }>): Promise<any> (note Job is from the bull package, and that you may wish to type your return vs. using any). Here is where you would actually send out the email.
  • In your processor class, @nestjs/bull also provides special method decorators including @OnQueueActive(), @OnQueueCompleted(), @OnQueueFailed(). Refer to the docs but you may find these useful for logging or other purposes.

这个想法是，当应用处于空闲状态时，您的处理器会处理队列中的作业.

The idea is that your processor handles jobs in the queue when the app is otherwise idle.

您的邮件模块可能至少有一个带有配置的 mail.module.ts，一个向邮件"添加作业的 mail.service.ts.队列，以及一个 mail.processor.ts，负责完成添加到邮件"中的任何作业.排队.

Your mail module would presumably have at least a mail.module.ts with configuration, a mail.service.ts that adds jobs to the "mail" queue, and a mail.processor.ts that takes care of completing any jobs added to the "mail" queue.

更多来自 NestJS 的文档，请访问:

Further documentation from NestJS is available at:

https://docs.nestjs.com/techniques/queues

相关文章