后端稳定性建设｜容错四大金刚：重试（下）gRPC 统一对齐重试、配套全链路治理

syntax = "proto3";
import "google/protobuf/descriptor.proto";
package com.framework.grpc.options;
// 自定义方法维度重试标识扩展
extend google.protobuf.MethodOptions {
  // true：接口幂等，允许上游调用方重试；false/不声明：禁止重试
  bool retryable = 50001;
}
message RetrySuggest {
  int32 max_retries = 1;
  int64 min_backoff_ms = 2;
  int64 max_backoff_ms = 3;
  double multiplier = 4;
}

syntax = "proto3";
import "com/framework/protobuf/options/retry_option.proto";
service DataRpcService {
  // 只读查询，声明支持重试
  rpc BatchQueryData(BatchQueryReq) returns (BatchQueryResp) {
    option (com.framework.grpc.options.retryable) = true;
  };
  // 写操作，不添加标识，永久禁止重试
  rpc ModifyData(ModifyReq) returns (ModifyResp);
  // 双向流式，无论是否标注，拦截器直接屏蔽重试
  rpc StreamSyncData(stream SyncReq) returns (stream SyncResp);
}

package com.framework.grpc.retry;
import io.grpc.MethodDescriptor;
import io.grpc.protobuf.ProtoMethodDescriptorSupplier;
import com.framework.grpc.options.RetryOptionProto;
import lombok.extern.slf4j;

@Slf4j
public class GrpcRetryMetaResolver {
    private GrpcRetryMetaResolver() {}
    public static boolean isMethodRetryable(MethodDescriptor<?, ?> method) {
        Object schema = method.getSchemaDescriptor();
        if (!(schema instanceof ProtoMethodDescriptorSupplier)) {
            return false;
        }
        try {
            ProtoMethodDescriptorSupplier supplier = (ProtoMethodDescriptorSupplier) schema;
            com.google.protobuf.Descriptors.MethodDescriptor protoMethod = supplier.getMethodDescriptor();
            var opts = protoMethod.getOptions();
            return opts.hasExtension(RetryOptionProto.retryable) && opts.getExtension(RetryOptionProto.retryable);
        } catch (Exception e) {
            log.warn("读取RPC方法重试标识异常, method={}", method.getFullMethodName(), e);
            return false;
        }
    }
}

package com.framework.grpc.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.util.HashMap;
import java.util.Map;

@Data
@ConfigurationProperties(prefix="rpc.grpc.retry")
public class GrpcRetryProperties {
    private boolean enabled = false;
    private int maxRetryNextInstance = 1;
    private long minBackoffMs = 100;
    private long maxBackoffMs = 1000;
    private double multiplier = 2.0;
    private Map<String, ServiceRetryConfig> serviceConfigs = new HashMap<>();
    private String[] retryStatusCodes = {"UNAVAILABLE","ABORTED"};

    // 方法开关判断、参数合并逻辑和Feign完全一致
    public boolean isMethodOpen(String serviceName, String fullMethod) {
        if (!enabled) return false;
        ServiceRetry cfg = serviceConfigs.get(serviceName);
        if (cfg == null) return true;
        var methodCfg = cfg.getMethods().get(fullMethod);
        return methodCfg != null ? method.isEnabled() : cfg.isEnabled();
    }

    // 内部ServiceRetryConfig/MethodRetryConfig 实体同Feign结构，省略重复代码
}

package com.framework.grpc.retry;
import com.framework.grpc.config.GrpcRetryProperties;
import io.grpc.*;
import lombok.extern.slf4j;
import org.springframework.core.annotation.Order;
import net.devh.boot.grpc.client.interceptor.GrpcGlobalClientInterceptor;

@Slf4j
@Order(100)
@GrpcGlobalClientInterceptor
@Component
public class GrpcRetryInterceptor implements ClientInterceptor {
    private final GrpcRetryProperties retryProperties;
    // 全局单线程异步调度器，不阻塞Netty EventLoop
    private static final ScheduledExecutorService SCHEDULER = Executors.newSingleThreadScheduledExecutor(r->{
        Thread t=new Thread(r,"grpc-retry-scheduler");t.setDaemon();return t;
    });

    public GrpcRetryInterceptor(GrpcRetryProperties retryProperties) {
        this.retryProperties = retryProperties;
    }

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
        // 流式接口直接跳过重试
        if (method.getType() != MethodDescriptor.MethodType.UNARY) {
            return next.newCall(method, callOptions);
        }
        String fullMethod = method.getFullMethodName();
        String serviceName = method.getServiceName();
        // 条件1：服务端IDL是否允许重试
        boolean serverAllow = GrpcRetryMetaResolver.isMethodRetryable(method);
        // 条件2：调用方Nacos是否开启
        boolean clientOpen = retryProperties.isMethodOpen(serviceName, fullMethod);
        if (log.isDebugEnabled()) {
            log.debug("grpc retry judge method={}, server={}, client={}", fullMethod, serverAllow, clientOpen);
        }
        // 双条件全部满足，包装重试Call
        if (serverAllow && clientOpen) {
            var config = retryProperties.getEffectiveConfig(serviceName, fullMethod);
            return new RetryableClientCall<>(method, callOptions, next, config, SCHEDULER, retryProperties.getRetryStatusCodes());
        }
        return next.newCall(method, callOptions);
    }
}

package com.framework.grpc.retry;
import com.framework.grpc.config.GrpcRetryProperties;
import io.grpc.*;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RetryableClientCall<ReqT,RespT> extends ClientCall<ReqT,RespT> {
    private final MethodDescriptor<ReqT,RespT> method;
    private final CallOptions callOptions;
    private final Channel next;
    private final GrpcRetryProperties.ServiceRetryConfig config;
    private final ScheduledExecutorService scheduler;
    private final String[] retryCodes;
    private final AtomicInteger retryCount = new AtomicInteger(0);
    private volatile ClientCall<ReqT,RespT> delegate;
    private volatile Listener<RespT> responseListener;
    private volatile ReqT cacheReq;
    private volatile Metadata cacheMeta;
    private volatile boolean finished = false;
    private volatile boolean cancelled = false;

    public RetryableClientCall(MethodDescriptor<ReqT,RespT> method, CallOptions callOptions, Channel next,
                               GrpcRetryProperties.ServiceRetryConfig config, ScheduledExecutorService scheduler, String[] retryCodes) {
        this.method = method;
        this.callOptions = callOptions;
        this.next = next;
        this.config = config;
        this.scheduler = scheduler;
        this.retryCodes = retryCodes;
    }

    @Override
    public void start(Listener<RespT> listener, Metadata headers) {
        this.responseListener = listener;
        this.cacheMeta = headers;
        doCall();
    }

    private void doCall() {
        if (cancelled || finished) return;
        delegate = next.newCall(method, callOptions);
        delegate.start(new InnerListener(), cacheMeta);
        if (cacheReq != null) delegate.sendMessage(cacheReq);
        delegate.halfClose();
    }

    @Override
    public void sendMessage(ReqT message) {
        this.cacheReq = message;
    }

    @Override
    public void halfClose() {}

    @Override
    public void cancel(String msg, Throwable t) {
        cancelled = true;
        if (delegate != null) delegate.cancel(msg, t);
    }

    private boolean canRetry(Status status) {
        // 1、判断重试次数是否耗尽
        if (retryCount.get() >= config.getMaxRetryNextInstance()) return false;
        // 2、状态码是否在可重试白名单
        String codeName = status.getCode().name();
        for(String c : retryCodes) {
            if(c.equals(codeName)) return true;
        }
        return false;
    }

    // 计算指数退避时长
    private long calcBackoff(int times) {
        long wait = (long)(config.getMinBackoffMs() * Math.pow(config.getMultiplier(), times -1));
        return Math.min(wait, config.getMaxBackoffMs());
    }

    private class InnerListener extends Listener<RespT> {
        @Override
        public void onHeaders(Metadata headers) {
            if (!finished) responseListener.onHeaders(headers);
        }
        @Override
        public void onMessage(RespT message) {
            if (!finished) responseListener.onMessage(message);
        }
        @Override
        public void onClose(Status status, Metadata trailers) {
            if (finished || cancelled) return;
            // 正常成功，直接返回
            if (status.isOk()) {
                finished = true;
                responseListener.onClose(status, trailers);
                return;
            }
            // 检查Deadline是否过期，过期直接终止
            if (callOptions.getDeadline() != null && callOptions.getDeadline().isExpired()) {
                finished = true;
                responseListener.onClose(Status.DEADLINE_EXCEEDED, trailers);
                return;
            }
            // 判断是否允许重试
            if (!canRetry(status)) {
                finished = true;
                responseListener.onClose(status, trailers);
                return;
            }
            int current = retryCount.incrementAndGet();
            long waitMs = calcBackoff(current);
            // 打印ERROR告警日志，方便监控采集
            log.error("[GRPC重试触发] method={}, 重试次数={}, 状态码={}, 等待{}ms",
                    method.getFullMethodName(), current, status.getCode(), waitMs);
            // 异步退避后重新发起调用，自动换实例
            scheduler.schedule(() -> {
                if (cancelled || finished) return;
                // 再次校验截止时间
                if (callOptions.getDeadline() != null && callOptions.getDeadline().isExpired()) {
                    finished = true;
                    responseListener.onClose(Status.DEADLINE_EXCEEDED, trailers);
                    return;
                }
                doCall();
            }, waitMs, TimeUnit.MILLISECONDS);
        }
        @Override
        public void onReady() {
            if (!finished) responseListener.onReady();
        }
    }
}

package com.framework.grpc.config;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan("com.framework.grpc.retry")
@EnableConfigurationProperties(GrpcRetryProperties.class)
public class GrpcRetryAutoConfig {
}

rpc:
  # grpc与feign共用一套顶层前缀，配置结构对齐
  grpc:
    retry:
      enabled: true
      max-retry-next-instance: 2
      min-backoff-ms: 100
      max-backoff-ms: 1000
      multiplier: 2.0
      retryable-status-codes:
        - UNAVAILABLE
        - ABORTED
      service-configs:
        data-rpc-service:
          enabled: true
          methods:
            data-rpc-service/BatchQueryData:
              enabled: true
              max-retry-next-instance:3
            data-rpc-service/ModifyData:
              enabled: false