如何在NestJS的GRPC服务端设置请求超时？需客户端服务端示例

1. GRPC服务端实现 #

GRPC服务端本身无内置超时配置，但可通过读取客户端传入的deadline参数，在业务逻辑中实现超时中断。以下是完整实现：

1.1 Protobuf 定义 (user.proto) #

syntax = "proto3";

package user;

service UserService {
  rpc GetUser(GetUserRequest) returns (GetUserResponse) {}
}

message GetUserRequest {
  int32 id = 1;
}

message GetUserResponse {
  int32 id = 1;
  string name = 2;
  string email = 3;
}

1.2 服务端业务逻辑 (user.service.ts) #

import { Injectable } from '@nestjs/common';
import { GrpcMethod } from '@nestjs/microservices';

@Injectable()
export class UserService {
  // 模拟数据库慢查询
  private async fetchUserFromDb(userId: number): Promise<any> {
    return new Promise((resolve) => {
      setTimeout(() => {
        resolve({ id: userId, name: 'John Doe', email: 'john@example.com' });
      }, 3000); // 模拟3秒响应
    });
  }

  @GrpcMethod('UserService', 'GetUser')
  async getUser(data: { id: number }, _metadata: any, call: any): Promise<any> {
    const deadline = call.getDeadline();
    if (!deadline) {
      return this.fetchUserFromDb(data.id);
    }

    const timeoutMs = deadline.getTime() - Date.now();
    if (timeoutMs <= 0) {
      throw new Error('DEADLINE_EXCEEDED');
    }

    // 用Promise.race实现超时中断
    const timeoutPromise = new Promise((_, reject) => {
      setTimeout(() => reject(new Error('DEADLINE_EXCEEDED')), timeoutMs);
    });

    return Promise.race([this.fetchUserFromDb(data.id), timeoutPromise]);
  }
}

1.3 服务端启动配置 #

app.module.ts

import { Module } from '@nestjs/common';
import { UserService } from './user.service';

@Module({
  providers: [UserService],
})
export class AppModule {}

main.ts

import { NestFactory } from '@nestjs/core';
import { Transport } from '@nestjs/microservices';
import { join } from 'path';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice(AppModule, {
    transport: Transport.GRPC,
    options: {
      package: 'user',
      protoPath: join(__dirname, './user.proto'),
      url: 'localhost:50051',
    },
  });
  await app.listen();
}
bootstrap();

2. GRPC客户端实现 #

客户端通过设置deadline参数指定请求超时时间，以下是实现示例：

2.1 客户端模块配置 (client.module.ts) #

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { join } from 'path';
import { UserClientService } from './user-client.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'USER_SERVICE',
        transport: Transport.GRPC,
        options: {
          package: 'user',
          protoPath: join(__dirname, './user.proto'),
          url: 'localhost:50051',
        },
      },
    ]),
  ],
  providers: [UserClientService],
  exports: [UserClientService],
})
export class ClientModule {}

2.2 客户端调用逻辑 (user-client.service.ts) #

import { Inject, Injectable } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { Observable } from 'rxjs';

interface UserService {
  getUser(data: { id: number }, metadata?: any, options?: any): Observable<any>;
}

@Injectable()
export class UserClientService {
  private userService: UserService;

  constructor(@Inject('USER_SERVICE') private client: ClientGrpc) {}

  onModuleInit() {
    this.userService = this.client.getService<UserService>('UserService');
  }

  // 发起带2秒超时的请求
  async getUserWithTimeout(userId: number) {
    const deadline = new Date();
    deadline.setSeconds(deadline.getSeconds() + 2);

    try {
      return await this.userService.getUser(
        { id: userId },
        {},
        { deadline },
      ).toPromise();
    } catch (error) {
      if (error.message === 'DEADLINE_EXCEEDED') {
        throw new Error('请求超时，请稍后重试');
      }
      throw error;
    }
  }
}

2.3 测试用控制器（可选） #

import { Controller, Get, Param } from '@nestjs/common';
import { UserClientService } from './user-client.service';

@Controller('users')
export class UserController {
  constructor(private readonly userClientService: UserClientService) {}

  @Get(':id')
  async getUser(@Param('id') id: string) {
    return this.userClientService.getUserWithTimeout(parseInt(id));
  }
}

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