NestJS微服务gRPC模块间调用失败问题求助

项目架构 #

• core项目：作为API网关接收HTTP请求，同时提供名为ACCO的gRPC服务模块
• client项目：提供名为User的gRPC服务供core项目调用，同时需要反向调用core的ACCO服务

调用流程与代码实现 #

1. core项目UserSvcController接收HTTP请求 #

@Controller('user-svc')
export class UserSvcController {
  private logger = new Logger(UserSvcController.name);

  constructor(private readonly userSvcService: UserSvcService) {}

  @Get(':id')
  async getUser(@Param() params): Promise<GetUserResponse> {
    this.logger.debug(`call: params=${JSON.stringify(params)}`);
    return this.userSvcService.getUser(params.id);
  }
}

2. core项目UserSvcService通过gRPC调用client的User服务 #

import { Observable } from 'rxjs';
import { GetUserRequest } from '../dtos/get-user-request.interface';
import { GetUserResponse } from '../dtos/get-user-response.interface';
import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { userGrpcClientOptions } from '../protos/user-grpc.option';
import { Client, ClientGrpc } from '@nestjs/microservices';

interface IUserService {
  getUser(data: GetUserRequest): Observable<GetUserResponse>;
}

@Injectable()
export class UserSvcService implements OnModuleInit {
  private logger = new Logger(UserSvcService.name);

  @Client(userGrpcClientOptions) private readonly client: ClientGrpc;
  private userService: IUserService;

  onModuleInit() {
    this.logger.debug(`UserSvcService.onModuleInit`);
    this.userService = this.client.getService<IUserService>('UserService');
  }

  async getUser(id: string): Promise<GetUserResponse> {
    const user: Observable<GetUserResponse> = this.userService.getUser({
      id: id,
    } as GetUserRequest);

    return new Promise((resolve, reject) => {
      user.subscribe({
        next: (getUserResponse: GetUserResponse) => {
          this.logger.debug(
            `getUser=${JSON.stringify(getUserResponse, null, 2)}`,
          );
          resolve(getUserResponse);
        },
        error: (err) => {
          this.logger.error(`err=${JSON.stringify(err, null, 2)}`);
          reject(err);
        },
      });
    });
  }
}

3. client项目UserService调用AccoSvcService方法 #

import {
  BadRequestException,
  forwardRef,
  Inject,
  Injectable,
  Logger,
} from '@nestjs/common';
import { GetUserRequest } from '../dto/get-user-request.interface';
import { GetUserResponse } from '../dto/get-user-response.interface';
import { AccoSvcService } from '../../acco-svc/services/acco-svc.service';
import { FindAllByOwnerUidRequest } from '../../acco-svc/dto/find-all-by-owner-uid.request';

@Injectable()
export class UserService {
  private logger = new Logger(UserService.name);

  constructor(
    @Inject(forwardRef(() => AccoSvcService))
    private readonly accoService: AccoSvcService,
  ) {}

  async create(data: GetUserRequest): Promise<GetUserResponse> {
    const accoIds: string[] = await this.accoService
      .findAllByOwnerUid({ ownerUid: data.id } as FindAllByOwnerUidRequest)
      .then((response) => {
        return response.accoIds;
      })
      .catch((err) => {
        this.logger.error(`findByUid ${data.id} ${err}`);
        throw new BadRequestException(err);
      });
    this.logger.debug(`findByUid ${data.id} ${accoIds}`);

    return new Promise((resolve) => {
      resolve({
        username: 'test',
      });
    });
  }
}

4. client项目AccoSvcService调用core的ACCO gRPC服务 #

import { Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { accoGrpcClientOptions } from '../protos/acco-grpc.option';
import { Client, ClientGrpc } from '@nestjs/microservices';
import { FindAllByOwnerUidRequest } from '../dto/find-all-by-owner-uid.request';
import { Observable } from 'rxjs';
import { FindAllByOwnerUidResponse } from '../dto/find-all-by-owner-uid.response';

interface IAccoService {
  findAllByOwnerUid(
    data: FindAllByOwnerUidRequest,
  ): Observable<FindAllByOwnerUidResponse>;
}

@Injectable()
export class AccoSvcService implements OnModuleInit {
  private logger = new Logger(AccoSvcService.name);

  @Client(accoGrpcClientOptions) private readonly client: ClientGrpc;
  private accoService: IAccoService;

  async onModuleInit() {
    this.logger.debug(`AccoSvcService.onModuleInit`);
    this.accoService = await this.client.getService<IAccoService>(
      'AccoService',
    );
  }

  async findAllByOwnerUid(
    data: FindAllByOwnerUidRequest,
  ): Promise<FindAllByOwnerUidResponse> {
    this.logger.debug(
      `findAllByOwnerUid in AccoService ${JSON.stringify(data, null, 2)}`,
    );
    try {
      const accoIds: Observable<FindAllByOwnerUidResponse> =
        this.accoService.findAllByOwnerUid(data);

      this.logger.debug(
        `findAllByOwnerUid after calling accoService ${data.ownerUid}`,
      );
      return new Promise((resolve, reject) => {
        accoIds.subscribe({
          next: (value: FindAllByOwnerUidResponse) => {
            this.logger.debug(
              `findAllByOwnerUid ${JSON.stringify(value, null, 2)}`,
            );
            resolve(value);
          },
          error: (err: any) => {
            this.logger.error(`findAllByOwnerUid ${err}`);
            reject(err);
          },
        });
      });
    } catch (err) {
      this.logger.error(`acco-svc-service's findAllByOwnerUid ${err}`);
    }
  }
}

5. core项目AccoService与Controller实现gRPC服务 #

// AccoService
import { Injectable, Logger } from '@nestjs/common';
import { FindAllByOwnerUidResponse } from '../dtos/find-all-by-owner-uid.response';

@Injectable()
export class AccoService {
  private logger = new Logger(AccoService.name);

  findAll() {
    return `This action returns all acco`;
  }

  async findAllByOwnerUid(uid: string): Promise<FindAllByOwnerUidResponse> {
    this.logger.log(
      `In Acco Service, getUserResponse=${JSON.stringify(uid, null, 2)}`,
    );

    return new Promise((resolve) => {
      resolve({
        accoIds: ['1', '2'],
      } as FindAllByOwnerUidResponse);
    });
  }
}

// AccoController
import { Controller, Get, Logger } from '@nestjs/common';
import { AccoService } from '../services/acco.service';
import { GrpcMethod } from '@nestjs/microservices';
import { FindAllByOwnerUidRequest } from '../dtos/find-all-by-owner-uid.request';
import { FindAllByOwnerUidResponse } from '../dtos/find-all-by-owner-uid.response';

@Controller('acco')
export class AccoController {
  private logger = new Logger(AccoController.name);

  constructor(private readonly accoService: AccoService) {}

  @GrpcMethod('AccoService', 'FindAllByOwnerUid')
  async findAllByOwnerUid(
    data: FindAllByOwnerUidRequest,
  ): Promise<FindAllByOwnerUidResponse> {
    this.logger.log(
      `Acco Controller findAllByOwnerUid: ${JSON.stringify(data, null, 2)}`,
    );
    return await this.accoService.findAllByOwnerUid(data.ownerUid);
  }

  @Get()
  findAll() {
    return this.accoService.findAll();
  }
}

错误日志 #

[Nest] 9484  - 03/02/2023, 4:15:09 PM   ERROR [AccoSvcService] acco-svc-service's findAllByOwnerUid TypeError: Cannot read properties of undefined (reading 'findAllByOwnerUid')
[Nest] 9484  - 03/02/2023, 4:15:09 PM   ERROR [UserService] findByUid 123 TypeError: Cannot read properties of undefined (reading 'accoIds')
[Nest] 9484  - 03/02/2023, 4:15:09 PM    WARN DEPRECATED! Passing the error cause as the first argument to HttpException constructor is deprecated. You should use the "options" parameter instead: new HttpException("message", 400, { cause: new Error("Some Error") }) 
[Nest] 9484  - 03/02/2023, 4:15:09 PM   ERROR [RpcExceptionsHandler] Cannot read properties of undefined (reading 'accoIds')
BadRequestException: Cannot read properties of undefined (reading 'accoIds')
    at /Users/ihyeonjun/WebstormProjects/mcservice1/src/user/services/user.service.ts:30:15
    at processTicksAndRejections (node:internal/process/task_queues:95:5)
    at UserService.create (/Users/ihyeonjun/WebstormProjects/mcservice1/src/user/services/user.service.ts:23:31)
    at /Users/ihyeonjun/WebstormProjects/mcservice1/node_modules/@nestjs/microservices/context/rpc-proxy.js:11:32
    at Object.GetUser (/Users/ihyeonjun/WebstormProjects/mcservice1/node_modules/@nestjs/microservices/server/server-grpc.js:148:40)

问题排查与修复方案 #

1. 修复AccoSvcService的onModuleInit方法 #

client.getService是同步方法，不需要使用async/await，异步初始化会导致服务调用时accoService还未赋值，出现undefined错误。

修改后的代码：

onModuleInit() {
  this.logger.debug(`AccoSvcService.onModuleInit`);
  this.accoService = this.client.getService<IAccoService>('AccoService');
}

2. 对齐UserService的方法名 #

core项目调用的是UserService的getUser方法，但client项目中只定义了create方法，导致gRPC调用路由错误，后续流程异常。将create方法重命名为getUser：

async getUser(data: GetUserRequest): Promise<GetUserResponse> {
  const accoIds: string[] = await this.accoService
    .findAllByOwnerUid({ ownerUid: data.id } as FindAllByOwnerUidRequest)
    .then((response) => {
      return response.accoIds;
    })
    .catch((err) => {
      this.logger.error(`findByUid ${data.id} ${err}`);
      throw new BadRequestException({ message: err.message, cause: err });
    });
  this.logger.debug(`findByUid ${data.id} ${accoIds}`);

  return { username: 'test' };
}

3. 完善AccoSvcService的错误处理 #

原代码中try/catch捕获错误后未抛出，导致返回undefined，上层读取response.accoIds时报错。在catch块中抛出错误：

async findAllByOwnerUid(
  data: FindAllByOwnerUidRequest,
): Promise<FindAllByOwnerUidResponse> {
  this.logger.debug(
    `findAllByOwnerUid in AccoService ${JSON.stringify(data, null, 2)}`,
  );
  try {
    const accoIds$: Observable<FindAllByOwnerUidResponse> =
      this.accoService.findAllByOwnerUid(data);

    this.logger.debug(
      `findAllByOwnerUid after calling accoService ${data.ownerUid}`,
    );
    return new Promise((resolve, reject) => {
      accoIds$.subscribe({
        next: (value: FindAllByOwnerUidResponse) => {
          this.logger.debug(
            `findAllByOwnerUid ${JSON.stringify(value, null, 2)}`,
          );
          resolve(value);
        },
        error: (err: any) => {
          this.logger.error(`findAllByOwnerUid ${err}`);
          reject(err);
        },
      });
    });
  } catch (err) {
    this.logger.error(`acco-svc-service's findAllByOwnerUid ${err}`);
    throw err;
  }
}

4. 修正HttpException的使用方式 #

按照警告提示，将BadRequestException的参数改为符合新规范的写法：

throw new BadRequestException({
  message: 'Failed to fetch acco ids',
  cause: err
});

内容的提问来源于stack exchange，提问作者Jun