nestjs - 如何动态注册队列以嵌套js牛队列

Question

当您知道要注册的队列列表时，您可以在 nestjs 模块实例化中使用 registerQueue() 添加它们，但如何动态注册队列，例如。如果我有一个多租户架构，当创建一个新租户时我想动态注册一个新队列，是否可以使用嵌套 js 公牛队列？

Answer 1

我不知道您是否可以使用 Nest 包做到这一点。检查来自 Nestjs 的创建者的这个GitHub 问题响应。

但是，您可以使用原版 Bull 包。

例如，假设您有一个这样的控制器：

@Controller()
export class AppController {
  constructor(
    private queuesManager: QueuesManagerService
  ) {}

  @Post(['generate-queue'])
  generateQueue(
    @Body() generateQueueDto: GenerateQueueDto
  ): Promise<GenerateQueueResponse> {
    return this.queuesManager.generateQueue(generateQueueDto.name);
  }

  @Get(['get-all-jobs'])
  getAllJobsFromQueue(
    @Query() queryParameters: GetAllJobsEndpointQueryParameters
  ): Promise<Bull.Job[]> {
    return this.queuesManager.getAllJobsFromQueue(queryParameters.name);
  }
}

QueuesManagerService看起来像这样：

@Injectable()
export class QueuesManagerService {
  async generateQueue(name: string): Promise<GenerateQueueResponse> {
    const queue: Bull.Queue = new Bull(name);
    await queue.add({ test: 'test' });
    return {
      status: 200,
      message: `Queue with name ${name} generated successfully!`,
    };
  }

  async getAllJobsFromQueue(name: string): Promise<Bull.Job[]> {
    const jobStatuses: Bull.JobStatus[] = [
      'waiting',
      'delayed',
      'active',
      'completed',
      'failed',
    ];
    const queue: Bull.Queue = new Bull(name);
    const jobs: Bull.Job[] = await queue.getJobs(jobStatuses);
    return jobs;
  }
}

您可以使用以下方式与服务器交互curl：

$ curl -X POST -d 'name=myFirstQueue' localhost:3333/api/generate-queue

# response
{"status":200,"message":"Queue with name myFirstQueue generated successfully!"}

###

$ curl localhost:3333/api/get-all-jobs?name=myFirstQueue

# response
[{"id":"1","name":"__default__","data":{"test":"test"},"opts":{"attempts":1,"delay":0,"timestamp":1639085434398},"progress":0,"delay":0,"timestamp":1639085434398,"attemptsMade":0,"stacktrace":[],"returnvalue":null,"finishedOn":null,"processedOn":null}]

PS1：

公牛的 Github

queue.getJobs() 参考

PS2：

我的类和接口：

export class GetAllJobsEndpointQueryParameters {
  @IsNotEmpty()
  name!: string;
}

export class GenerateQueueDto {
  @IsNotEmpty()
  name!: string;
}

export interface GenerateQueueResponse {
  status: number;
  message: string;
}