gRPC从0到1系列【23】

gRPC拦截器

7.4 ClientInterceptor

gRPC 中的 ClientInterceptor（客户端拦截器）是一种强大的机制，允许开发者在客户端发起 gRPC 调用的前后插入自定义逻辑。这些逻辑可以用于日志记录、认证、监控、重试、超时控制、请求/响应修改等多种用途。

7.4.1 什么是ClientInterceptor?

ClientInterceptor 是 gRPC 客户端的一个扩展点，它可以在每次 RPC 调用执行前或执行后拦截请求和响应。它类似于 Web 开发中的“中间件”（middleware）。

在 gRPC 的 Java 实现中，ClientInterceptor 是一个接口：

public interface ClientInterceptor {<ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,CallOptions callOptions,Channel next);
}

• method：描述被调用的方法（包括方法名、请求/响应类型等）。
• callOptions：调用选项（如 deadline、权威、压缩等）。
• next：下一个 Channel（可以是另一个拦截器或最终的底层通道）。
• 返回一个 ClientCall 对象，代表这次调用的封装。

相关核心类:

// 客户端调用接口
public abstract static class ClientCall<ReqT, RespT> {public abstract void start(Listener<RespT> responseListener, Metadata headers);public abstract void sendMessage(ReqT message);public abstract void halfClose();public abstract void cancel(String message, Throwable cause);public abstract void request(int numMessages);
}// 响应监听器
public abstract static class Listener<T> {public void onHeaders(Metadata headers) {}public void onMessage(T message) {}public void onClose(Status status, Metadata trailers) {}public void onReady() {}
}

7.4.2 ClientInterceptor工作流程

当你将一个或多个 ClientInterceptor 应用于 gRPC 客户端时，它们会按照注册顺序形成一个调用链。

MyServiceGrpc.MyServiceBlockingStub stub = MyServiceGrpc.newBlockingStub(channel).withInterceptors(interceptor1, interceptor2);

调用链顺序为：

stub → interceptor1 → interceptor2 → 实际网络调用

每个拦截器可以选择：

• 修改请求/响应；
• 添加 metadata（如认证 token）；
• 记录日志；
• 处理异常；
• 控制调用流程（如提前终止、重试等）。

7.4.3 常见应用场景

7.4.4 实战示例

✅ 1. 添加认证信息

package cn.tcmeta.grpc.interceptor;import io.grpc.*;public class AuthInterceptor implements ClientInterceptor {private final String token;public AuthInterceptor(String token) {this.token = token;}@Overridepublic <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,CallOptions callOptions,Channel next) {ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);return new ForwardingClientCall.SimpleForwardingClientCall<>(call) {@Overridepublic void start(Listener<RespT> responseListener, Metadata headers) {headers.put(Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER), "Bearer " + token);super.start(responseListener, headers);}};}
}

✅ 2. 日志记录

package cn.tcmeta.grpc.interceptor;import io.grpc.*;public class LoggingInterceptor implements ClientInterceptor {@Overridepublic <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,CallOptions callOptions,Channel next) {System.out.println("Calling method: " + method.getFullMethodName());ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);return new ForwardingClientCall.SimpleForwardingClientCall<>(call) {@Overridepublic void sendMessage(ReqT message) {System.out.println("Sending request: " + message);super.sendMessage(message);}};}
}

7.4.5 重试、超时

可以通过包装 ClientCall.Listener 来监听失败，并在特定条件下重试.

虽然 CallOptions 支持 deadline，但拦截器可以动态设置：

CallOptions newOptions = callOptions.withDeadlineAfter(5, TimeUnit.SECONDS);
ClientCall<ReqT, RespT> call = next.newCall(method, newOptions);