NestJS 分离队列答案

【问题标题】：NestJS separate QueuesNestJS 分离队列
【发布时间】：2021-07-30 08:23:18
【问题描述】：

根据文档NestJS Docs，有两种实现队列的方法。使用 redis 和命名或未命名的队列，例如像这里：Queue Redis Example 或这里 Queue Redis Example 2 和在带有处理器的单独进程中运行的队列。

我最近尝试实现Queues in separate processes 此处找到的第二种方式，但我的处理器无法识别，最终得到：

Error message

有没有人可以帮助我并提供一个最小的工作示例？

我看到已经有一个最近打开的问题：Queues Processes Docs Issue 但如果有人可以帮助我或提供的线程中的这个人，我可以进行 PR 或官方维护人员可以这样做。

提前致谢！最好的问候拉吉塔哈

【问题讨论】：

您能否提供代码示例或 repo 以供审核？该错误表明您的 csv-queue 处理器在您正在运行的进程的当前范围内不可访问。
你是对的，因为我试图从正常队列移动到单独的处理器。然后不再找到处理器。您是否体验过单独的队列？我回家后会提供样品。谢谢！
@TheGeek 请在这里找到一个最小的 repo：Github repo 这适用于普通队列。如果您克隆它，请在端口 6379 上启动一个 Redis Docker 容器。我的问题是，要弄清楚如何在单独的进程中实现相同的队列，如此处所述：NestJs Docs。我可以请你帮忙吗？提前致谢

标签： typescript multithreading nestjs task-queue

【解决方案1】：

NestJS 公牛包是公牛的很好的包装器，它为在单独进程中运行的作业提供支持。

要利用公牛自动分叉处理，您只需提供可充当作业处理器的文件的路径。

创建您要用来运行作业的单独文件

import { Logger } from '@nestjs/common';
import { DoneCallback, Job } from 'bull';

export default function (job: Job<{ message: string }>, cb: DoneCallback) {
  Logger.verbose(`${job.data.message} (pid ${process.pid})`, `SEPARATE`);
  cb(null, 'Hurrah');
}

请记住，此文件需要能够单独运行。当然，您可以添加额外的导入等，但它将在不同的进程中运行，因此您的应用程序及其资源将不可用。

注册队列时，指定单独文件的路径。在这里，我注册了 2 个队列。一个在与应用程序相同的进程中运行，一个将在单独的进程中运行

@Module({
  imports: [
    // register root
    BullModule.forRoot({
      redis: {
        host: 'localhost',
        port: 6379,
      },
    }),
    // register queues
    BullModule.registerQueue(
      {
        name: 'SAME', // this will run in same process as this module
      },
      {
        name: 'SEPARATE', // this will run in its own process
        processors: [join(__dirname, 'separate.process')],
      },
    ),
  ],
  controllers: [AppController],
  providers: [SameService],
})
export class AppModule {}

然后你可以正常解雇工作。下面我将一个作业触发到应用程序进程中的队列中，然后将另一个作业触发到单独的文件中

@Controller()
export class AppController {
  constructor(
    @InjectQueue('SAME') private readonly same: Queue,
    @InjectQueue('SEPARATE') private readonly separate: Queue,
  ) {}

  @Get()
  getHello(): string {

    // Example of adding a job processed in same thread
    this.same.add({ message: 'Knock knock.' });

    // Example of adding a job processed in separate thread
    this.separate.add({ message: 'FORK OFF.' });

    return 'ok';
  }
}

localhost:3000 然后输出

[Nest] 13400   - 08/05/2021, 16:49:18   [SAME] Knock knock. (pid 13400) +8821ms
[Nest] 2660   - 08/05/2021, 16:49:19   [SEPARATE] FORK OFF. (pid 2660)

Here is a link to the repo

希望这会有所帮助！

【讨论】：

你是我的英雄！对我来说缺少的部分是找出如何添加工作！谢谢！！！！！！！
回购链接已失效。
这很奇怪@MarkusEnde，但链接到错误的（私人）回购。感谢您引起我的注意。现在已经更正了。希望对您有所帮助！