node.js - 用于队列和 api 的嵌套公牛单独进程

Question

我有一个nestjs暴露几个 RESTAPI的应用程序。其中一个APIs 触发处理某些任务的作业。问题是当作业被触发时，应用程序停止服务 REST 请求，这导致负载均衡器的健康检查失败。我按照自述文件末尾给出的方法启动了一个单独的子进程来处理作业。但是，该作业不会在子进程中启动，并且API请求会停止。

这是我的工作：

import {
  BullQueueEvents,
  OnQueueActive,
  OnQueueEvent,
  Process,
  Processor,
} from 'nest-bull';
import { Job } from 'bull';
import { Logger } from '@nestjs/common';
import { AService } from './a-service';
import { AJobInterface } from '../AJobInterface';

@Processor({ name: 'a_queue' })
export class AJob {
  private readonly logger = new Logger('AQueue');

  constructor(private readonly service: AService) {}

  @Process({
    name: 'app',
    concurrency: 1
  })
  processApp(job: Job<AJobInterface>) {
    console.log('CHILD: ', process.pid);
    const { jobId } = job.data;
    return this.service.process(jobId);
  }

  @OnQueueActive()
  onActive(job: Job) {
    this.logger.log(
      `Processing job ${job.id} of type ${job.name} with data ${JSON.stringify(
        job.data,
      )}...`,
    );
  }

  @OnQueueEvent(BullQueueEvents.COMPLETED)
  onCompleted(job: Job) {
    this.logger.log(
      `Completed job ${job.id} of type ${job.name} with result ${job.returnvalue}`,
    );
  }
}

这是我的 app.module.ts：

import { Module, OnModuleInit } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { DatabaseModule } from './db/module';
import { BullModule } from 'nest-bull';
import { AJob } from './worker/a-job';
import { AService } from './worker/a-service';
import { join } from 'path';

@Module({
  imports: [
    TypeOrmModule.forRoot(),
    DatabaseModule,
    BullModule.register({
      name: 'a_queue',
      processors: [ join(__dirname, 'worker/a-job.js') ],
      options: {
        redis: {
          host: process.env.REDIS_URL || '127.0.0.1',
          port: 6379,
          showFriendlyErrorStack: true,
        },
        settings: {
          lockDuration: 300000,
          stalledInterval: 300000
        },
      },
    }),
  ],
  controllers: [AppController],
  providers: [AppService, AJob, AService],
})
export class AppModule implements OnModuleInit {
  onModuleInit() {
    console.log('MAIN: ', process.pid);
  }
}

有什么我做错了吗？

标签： node.jstypescriptnestjsbull.js

Answer 1

很抱歉这么晚才发布答案。事实证明，在子进程中设置工作人员是不可能的。我最终有一个单独worker.module.ts的和一个单独的worker.ts，并为 API 和工作人员创建了两个单独的进程。

worker.module.ts：

import { Module, OnModuleInit } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { AppService } from '../app.service';
import { DatabaseModule } from '../db/module';
import { BullModule } from 'nest-bull';
import { AJob } from './a-job';
import { AService } from './a-service';
import { join } from 'path';
import { Job, DoneCallback } from 'bull';

@Module({
  imports: [
    TypeOrmModule.forRoot(),
    DatabaseModule,
    BullModule.register({
      name: 'a_queue',
      processors: [ (job: Job, done: DoneCallback) => { done(null, job.data); } ],
      options: {
        redis: {
          host: process.env.REDIS_URL || '127.0.0.1',
          port: 6379,
          password: process.env.REDIS_PWD,
          showFriendlyErrorStack: true,
        },
        settings: {
          lockDuration: 300000,
          stalledInterval: 300000
        },
      },
    }),
  ],
  providers: [AppService, AJob, AService],
})
export class WorkerModule implements OnModuleInit {
  onModuleInit() {
    console.log('WORKER: ', process.pid);
  }
}

worker.ts：

import { NestFactory } from '@nestjs/core';
import { WorkerModule } from './worker/worker.module';

async function bootstrap() {
  const app = await NestFactory.create(WorkerModule);
  app.init();
}

bootstrap();

虽然app.module.ts现在看起来像这样：

//...imports
@Module({
  imports: [
    TypeOrmModule.forRoot(),
    DatabaseModule,
    BullModule.register({
      name: 'a_queue',
      processors: [ ],
      options: {
        redis: {
          host: process.env.REDIS_URL || '127.0.0.1',
          port: 6379,
          showFriendlyErrorStack: true,
        },
      },
    }),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule implements OnModuleInit {
  onModuleInit() {
    console.log('MAIN: ', process.pid);
  }
}

和相应的app.ts：

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { port } from './config';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.enableCors();
  await app.listen(port);
}

bootstrap();