NestJS如何配置gRPC客户端拦截器 解决@grpc-js偶发连接问题

1. 编写gRPC重试拦截器 #

该拦截器专门捕获连接相关异常（包括ECONNRESET、UNAVAILABLE等状态），自动触发重试逻辑，避免偶发错误直接抛出到业务层，代码实现如下：

import * as grpc from '@grpc/grpc-js';

// 重试拦截器工厂函数
export const retryInterceptor = (options: { maxRetries: number, initialBackoffMs: number }) => {
  return (callOptions: any, nextCall: any) => {
    return new grpc.InterceptingCall(nextCall(callOptions), {
      start: (metadata, listener, next) => {
        let retryCount = 0;
        const retryListener = {
          onReceiveStatus: (status, next) => {
            // 定义可重试的连接错误状态码
            const retryableStatuses = [
              grpc.status.UNAVAILABLE,
              grpc.status.ABORTED,
              grpc.status.RESOURCE_EXHAUSTED,
              grpc.status.UNKNOWN // 对应ECONNRESET等底层连接错误
            ];
            if (retryableStatuses.includes(status.code) && retryCount < options.maxRetries) {
              retryCount++;
              // 指数退避等待后触发重试
              setTimeout(() => {
                const newCall = nextCall(callOptions);
                newCall.start(metadata, retryListener);
              }, options.initialBackoffMs * Math.pow(2, retryCount - 1));
              return;
            }
            next(status);
          }
        };
        next(metadata, retryListener);
      }
    });
  };
};

// 实例化拦截器，可自定义重试次数和初始退避时间
const connectionRetryInterceptor = retryInterceptor({
  maxRetries: 3,
  initialBackoffMs: 100
});

2. 注入拦截器到NestJS gRPC客户端配置 #

在你原有的ClientsModuleOptions配置中，给每个gRPC客户端的options字段新增channelArgs配置项，将上述拦截器注入即可生效，修改后的配置如下：

const clientOptions: ClientsModuleOptions = [
  {
    name: 'HP_USERS_PACKAGE',
    transport: Transport.GRPC,
    options: {
      url: `${env.USERS_SERVICE}:${config.users.port}`,
      package: config.users.v1.package,
      protoPath: config.users.v1.protoPath,
      loader: {
        includeDirs: [config.protoLoaderDir],
        enums: String,
      },
      // 新增拦截器配置
      channelArgs: {
        interceptors: [connectionRetryInterceptor]
      }
    },
  },
  {
    name: 'HP_STAGING_PACKAGE',
    transport: Transport.GRPC,
    options: {
      url: `${env.STAGING_SERVICE}:${config.staging.port}`,
      package: config.staging.v1.package,
      protoPath: config.staging.v1.protoPath,
      loader: {
        includeDirs: [config.protoLoaderDir],
        enums: String,
      },
      // 新增拦截器配置
      channelArgs: {
        interceptors: [connectionRetryInterceptor]
      }
    },
  }]

3. 额外优化建议 #

• 可以根据业务场景调整maxRetries最大重试次数和initialBackoffMs初始退避时间，非核心业务可以适当提高重试次数
• 如果存在幂等性问题的接口，需要单独配置拦截器的重试白名单，避免非幂等接口重复提交
• 生产环境可以在拦截器中加入错误日志上报，统计连接错误的出现频率，辅助排查底层网络问题

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