社区精选｜聊一聊 gRPC 中的拦截器

业界 作者：SegmentFault 2023-02-24 22:31:27 阅读：765

今天小编为大家带来的是社区作者 江南一点雨 的文章，让我们一起继续 gRPC 系列

• 一个简单的案例入门 gRPC
• https://mp.weixin.qq.com/s/OyfU0tLm4f9t3nZxce-Ksw
• 聊一聊 gRPC 的四种通信模式
• https://mp.weixin.qq.com/s/c-_D2RpLksIlYJDfaWOSkA

1. 服务端拦截器
2. 客户端拦截器

1. 服务端拦截器

public class BookServiceCallListener<R> extends ForwardingServerCallListener<R> {
    private final ServerCall.Listener<R> delegate;

    public BookServiceCallListener(ServerCall.Listener<R> delegate) {
        this.delegate = delegate;
    }

    @Override
    protected ServerCall.Listener<R> delegate() {
        return delegate;
    }

    @Override
    public void onMessage(R message) {
        System.out.println("这是客户端发来的消息，可以在这里进行预处理："+message);
        super.onMessage(message);
    }
}

public class BookServiceCall<ReqT,RespT> extends ForwardingServerCall.SimpleForwardingServerCall<ReqT,RespT> {
    protected BookServiceCall(ServerCall<ReqT, RespT> delegate) {
        super(delegate);
    }

    @Override
    protected ServerCall<ReqT, RespT> delegate() {
        return super.delegate();
    }

    @Override
    public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
        return super.getMethodDescriptor();
    }

    @Override
    public void sendMessage(RespT message) {
        System.out.println("这是服务端返回给客户端的消息："+message);
        super.sendMessage(message);
    }
}

public void start() throws IOException {
    int port = 50051;
    server = ServerBuilder.forPort(port)
            .addService(ServerInterceptors.intercept(new BookServiceImpl(), new ServerInterceptor() {
                @Override
                public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
                    String fullMethodName = call.getMethodDescriptor().getFullMethodName();
                    System.out.println(fullMethodName + ":pre");
                    Set<String> keys = headers.keys();
                    for (String key : keys) {
                        System.out.println(key + ">>>" + headers.get(Metadata.Key.of(key, ASCII_STRING_MARSHALLER)));
                    }
                    return new BookServiceCallListener<>(next.startCall(new BookServiceCall(call), headers));
                }
            }))
            .build()
            .start();
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        BookServiceServer.this.stop();
    }));
}

• 第一个参数 call 是消费传入的 RPC 消息的一个回调。
• 第二个参数 headers 则是请求的消息头，如果我们通过 JWT 进行请求校验，那么就从这个 headers 中提取出请求的 JWT 令牌然后进行校验。
• 第三个参数 next 就类似于我们在 Java 过滤器 filter 中的 filterChain 一样，让这个请求继续向下走。

2. 客户端拦截器

ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
        .usePlaintext()
        .intercept(new ClientInterceptor() {
            @Override
            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
                System.out.println("!!!!!!!!!!!!!!!!");
                callOptions = callOptions.withAuthority("javaboy");
                return next.newCall(method,callOptions);
            }
        })
        .build();
BookServiceGrpc.BookServiceStub stub = BookServiceGrpc.newStub(channel);

当我们的请求执行的时候，这个客户端拦截器就会被触发。

3. 小结