在 Typescript 中实现 Bull Queue

Posted 2023-03-10

【中文标题】在 Typescript 中实现 Bull Queue

【英文标题】：Implementing Bull Queue in Typescript

【发布时间】：2020-11-03 17:37:27

【问题描述】：

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

@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();
    

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

【问题讨论】：

你创建了一个mail.processor吗？

是/////////////

【参考方案1】：

我最近写了一篇似乎与您的用例相关的博文：

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()。请参阅文档，但您可能会发现这些对日志记录或其他用途很有用。

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

您的邮件模块大概至少有一个带有配置的mail.module.ts，一个将作业添加到“邮件”队列的mail.service.ts，以及一个负责完成添加到“邮件”中的任何作业的mail.processor.ts排队。

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

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

【讨论】：

太棒了！谢谢伙计！还有一个问题，我如何处理 e.q. 中的逻辑。 @OnQueueActive 而不是记录？

嘿，我很高兴它有帮助。用@OnQueueActive() 装饰的方法将作为参数传递给作业（例如方法签名：onQueueActive(job: Job)）。因此，在该方法中，您可以访问job.id、job.name、job.data 并且可以从那里开始。不过，我认为您不一定需要在其中添加逻辑，因为在大多数情况下您可以编写一个 @Process() 装饰方法来更好地分解您的代码。

为了完善我的评论，@OnQueueCompleted() 装饰方法接收 job 和 result 作为参数，@OnQueueFailed() 装饰方法接收 job 和 error 作为参数。

我有最后一个问题，你能举一个 Jest 中牛队测试的例子吗？

那里的文章很棒且透彻。