【橘子分布式】gRPC(番外篇-监听流)

一、简介

我们前面介绍了一元操作下的拦截器的各种操作。但是我们说一元操作的特点它是一个请求一个响应，所以他的拦截器的时机就是对于这一个请求响应的拦截。这种特点决定了他无法对于流式的那种多个消息的拦截。试想一下你如何用一元拦截器拦截流式的某一个批次中的某一个消息呢？或者你能不断的拦截他的消息流吗。所以我们需要在流式操作下的拦截器来实现这个能力。
下面我们先来搭建流式的客户端和服务端代码。

流式通信方式：主要应用 gRPC除了 一元RPC之外的其他形式RPC操作的拦截工作
1. 服务端流式RPC
2. 客户端流式RPC
3. 双向流式RPC

二、服务搭建

1、编写流式的proto

我们在api模块的原来的基础上添加一个双端流的声明hello1。

syntax = "proto3";

package com.levi;

option java_multiple_files = false;
option java_package = "com.levi";
option java_outer_classname = "HelloProto";


message HelloRequest{
  string name = 1;
}

message HelloRespnose{
  string result = 1;
}

service HelloService{
  // 普通方法
  rpc hello(HelloRequest) returns (HelloRespnose);
  // 双端流方法
  rpc hello1(stream HelloRequest) returns (stream HelloRespnose);
}

然后通过maven来生成对应的实体类和service。

ok，我们现在完成了proto的编写和构建。

2、服务端

我们来实现服务端的业务类，也就是实现这个hello1方法。

package com.levi.service;

import com.levi.HelloProto;
import com.levi.HelloServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {

    @Override
    public StreamObserver<HelloProto.HelloRequest> hello1(StreamObserver<HelloProto.HelloRespnose> responseObserver) {
        return new StreamObserver<HelloProto.HelloRequest>() {
            @Override
            public void onNext(HelloProto.HelloRequest helloRequest) {
                log.debug("客户端的请求消息为:{} ", helloRequest.getName());
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                log.debug("客户端请求的消息全部接收到 ....");
                // 服务端可以发送多条消息给客户端,在全部收到客户端消息之时，你也可以在上面的onNext每收到一条就发送一条，看你业务
                responseObserver.onNext(HelloProto.HelloRespnose.newBuilder().setResult("result 1").build());
                responseObserver.onNext(HelloProto.HelloRespnose.newBuilder().setResult("result 2").build());
                responseObserver.onCompleted();
            }
        };
    }

}

然后我们发布到服务上。

package com.levi;
import com.levi.service.HelloServiceImpl;
import io.grpc.Server;
import io.grpc.ServerBuilder;

import java.io.IOException;

public class GrpcServer {
    public static void main(String[] args) throws InterruptedException, IOException {
        ServerBuilder<?> serverBuilder = ServerBuilder.forPort(9000);
        serverBuilder.addService(new HelloServiceImpl());
        Server server = serverBuilder.build();
        server.start();
        server.awaitTermination();
    }
}

至此，我们完成了服务端的代码。

3、客户端

package com.levi;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

@Slf4j
public class GrpcClient {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder
                .forAddress("localhost", 9000)
                .usePlaintext()
                .build();

        try {
            HelloServiceGrpc.HelloServiceStub helloServiceStub = HelloServiceGrpc.newStub(managedChannel);
            StreamObserver<HelloProto.HelloRequest> requestStreamObserver = helloServiceStub.hello1(new StreamObserver<HelloProto.HelloRespnose>() {
                @Override
                public void onNext(HelloProto.HelloRespnose helloRespnose) {
                    log.info("接收到的服务端响应为{}: ", helloRespnose.getResult());
                }

                @Override
                public void onError(Throwable throwable) {

                }

                // 服务端这边的onCompleted方法会触发这个
                @Override
                public void onCompleted() {
                    log.info("服务端响应完成");
                }
            });
            for (int i = 0; i < 5; i++) {
                requestStreamObserver.onNext(HelloProto.HelloRequest.newBuilder().setName("levi" + i).build());
            }
            // 会触发服务端那边的onCompleted监听事件方法
            requestStreamObserver.onCompleted();
            managedChannel.awaitTermination(5, TimeUnit.SECONDS);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            managedChannel.shutdown();
        }
    }
}

注意，对端调用他们的onCompleted,这一端的onCompleted监听才会被触发。

三、拦截器开发

1、客户端流式拦截器

我们的流式拦截器和之前的功能其实差不多，都是拦截请求拦截响应。
我们在流式这里需要的类是。
ClientStreamTracer ：目的就是拦截请求 拦截响应
ClientStreamTracerFactory：目的 就是用于创建ClientStreamTracer.

下面我们就来编程实现一下。

package com.levi.interceptor;

import io.grpc.*;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class CustomerClientInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
        log.debug("执行客户端拦截器...");
        /*
         * 把自己开发的ClientStreamTracerFactory融入到gRPC体系，交给grpc,
         * 把拦截器作为callOption选项交给clientCall，grpc就会基于这个clientCall来做grpc调用
         * 也就有了拦截器的实现了
         */
        callOptions = callOptions.withStreamTracerFactory(new CustomClientStreamTracerFactory<>());
        /*
         * 把clientCall传递下去，交给grpc并且携带者方法元数据和选项
         * 方法元数据：方法的全路径，方法的输入输出参数类型
         * 选项：调用的选项，比如超时时间，重试次数，调用的凭证等，1传递给grpc，grpc就基于这个clientCall来做grpc调用
         */
        return next.newCall(method, callOptions);
    }
}

/**
 * 实现ClientStreamTracer.Factory，通过工厂创建我们自己定义的CustomClientStreamTracer
 * 用于客户端流式拦截：拦截请求 拦截响应
 * @param <ReqT>
 * @param <RespT>
 */
class CustomClientStreamTracerFactory<ReqT, RespT> extends ClientStreamTracer.Factory {
    @Override
    public ClientStreamTracer newClientStreamTracer(ClientStreamTracer.StreamInfo info, Metadata headers) {
        return new CustomClientStreamTracer<>();
    }
}


/*
 作用： 用于客户端流式拦截：拦截请求 拦截响应
 其中的outbound 方法用于拦截请求，inbound 方法用于拦截响应,和netty一样的设计。
 这个out和in都是站在我们客户端的角度说的，站在客户端角度，out就是离开客户端的操作也就是请求，inbound 就是响应。
 */
@Slf4j
class CustomClientStreamTracer<ReqT, RespT> extends ClientStreamTracer {
    //outbound 对于请求相关操作的拦截
    @Override
    //用于输出请求头
    public void outboundHeaders() {
        log.debug("client: 用于输出请求头.....");
        super.outboundHeaders();
    }

    @Override
    //设置消息编号,流式通信是多个消息发送，每个消息都有自己的编号
    public void outboundMessage(int seqNo) {
        log.debug("client: 设置流消息的编号 {} ", seqNo);
        super.outboundMessage(seqNo);
    }

    @Override
    public void outboundUncompressedSize(long bytes) {
        log.debug("client: 获得未压缩消息的大小 {} ", bytes);
        super.outboundUncompressedSize(bytes);
    }

    @Override
    //用于获得 输出消息的大小
    public void outboundWireSize(long bytes) {
        log.debug("client: 用于获得 输出消息的大小 {} ", bytes);
        super.outboundWireSize(bytes);
    }

    @Override
    //拦截消息发送
    public void outboundMessageSent(int seqNo, long optionalWireSize, long optionalUncompressedSize) {
        log.debug("client: 监控请求操作 outboundMessageSent {} ", seqNo);
        super.outboundMessageSent(seqNo, optionalWireSize, optionalUncompressedSize);
    }


    //inbound  对于响应相关操作的拦截
    @Override
    public void inboundHeaders() {
        log.debug("用于获得响应头....");
        super.inboundHeaders();
    }

    @Override
    public void inboundMessage(int seqNo) {
        log.debug("获得响应消息的编号...{} ",seqNo);
        super.inboundMessage(seqNo);
    }

    @Override
    public void inboundWireSize(long bytes) {
        log.debug("获得响应消息的大小...{} ",bytes);
        super.inboundWireSize(bytes);
    }

    @Override
    public void inboundMessageRead(int seqNo, long optionalWireSize, long optionalUncompressedSize) {
        log.debug("集中获得消息的编号 ，大小 ，未压缩大小 {} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);
        super.inboundMessageRead(seqNo, optionalWireSize, optionalUncompressedSize);
    }

    @Override
    public void inboundUncompressedSize(long bytes) {
        log.debug("获得响应消息未压缩大小 {} ",bytes);
        super.inboundUncompressedSize(bytes);
    }

    @Override
    public void inboundTrailers(Metadata trailers) {
        log.debug("响应结束..，对方服务端调用了onCompleted方法");
        super.inboundTrailers(trailers);
    }
}

我们看到首先是定义拦截器类，CustomerClientInterceptor实现ClientInterceptor接口，然后你能看到他的范型是ReqT, RespT，可见它是请求响应一起拦截了，比一元的编程模型要简单一些。
然后自己定义拦截器实现，也就是CustomClientStreamTracer<ReqT, RespT> extends ClientStreamTracer
在里面覆盖诸多请求响应实现的拦截方法。最后我们通过自己定义工厂类CustomClientStreamTracerFactory<ReqT, RespT> extends ClientStreamTracer.Factory把拦截实现定义出来，
最后把拦截实现整合在拦截器里面。
拦截实现->拦截工厂->包装在拦截器里面。此时我们就有了拦截器了。然后我们就可以把拦截器整合在客户端调用上。

@Slf4j
public class GrpcClient {
    public static void main(String[] args) {
        ManagedChannel managedChannel = ManagedChannelBuilder
                .forAddress("localhost", 9000)
                .intercept(List.of(new CustomerClientInterceptor()))// 绑定拦截器
                .usePlaintext()
                .build();

        ......省略
    }
}

此时我们对服务端发起调用查看一下输出。

2025-07-26 12:18:47.982 [main] DEBUG com.levi.interceptor.CustomerClientInterceptor - 执行客户端拦截器...
2025-07-26 12:18:48.652 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 设置流消息的编号 0 
2025-07-26 12:18:48.662 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于输出请求头.....
2025-07-26 12:18:48.667 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 获得未压缩消息的大小 7 
2025-07-26 12:18:48.667 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于获得 输出消息的大小 7 
2025-07-26 12:18:48.667 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 监控请求操作 outboundMessageSent 0 
2025-07-26 12:18:48.670 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 设置流消息的编号 1 
2025-07-26 12:18:48.671 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 获得未压缩消息的大小 7 
2025-07-26 12:18:48.671 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于获得 输出消息的大小 7 
2025-07-26 12:18:48.672 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 监控请求操作 outboundMessageSent 1 
2025-07-26 12:18:48.672 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 设置流消息的编号 2 
2025-07-26 12:18:48.672 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 获得未压缩消息的大小 7 
2025-07-26 12:18:48.673 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于获得 输出消息的大小 7 
2025-07-26 12:18:48.673 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 监控请求操作 outboundMessageSent 2 
2025-07-26 12:18:48.674 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 设置流消息的编号 3 
2025-07-26 12:18:48.675 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 获得未压缩消息的大小 7 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于获得 输出消息的大小 7 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 监控请求操作 outboundMessageSent 3 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 设置流消息的编号 4 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 获得未压缩消息的大小 7 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 用于获得 输出消息的大小 7 
2025-07-26 12:18:48.676 [grpc-default-executor-0] DEBUG com.levi.interceptor.CustomClientStreamTracer - client: 监控请求操作 outboundMessageSent 4 
2025-07-26 12:18:48.852 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 用于获得响应头....
2025-07-26 12:18:48.866 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息的编号...0 
2025-07-26 12:18:48.866 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息的大小...10 
2025-07-26 12:18:48.866 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 集中获得消息的编号 ，大小 ，未压缩大小 0 10 -1
2025-07-26 12:18:48.866 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息未压缩大小 10 
2025-07-26 12:18:48.868 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 响应结束..，对方服务端调用了onCompleted方法
2025-07-26 12:18:48.873 [grpc-default-executor-1] INFO  com.levi.GrpcClient - 接收到的服务端响应为result 1: 
2025-07-26 12:18:48.881 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息的编号...1 
2025-07-26 12:18:48.881 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息的大小...10 
2025-07-26 12:18:48.881 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 集中获得消息的编号 ，大小 ，未压缩大小 1 10 -1
2025-07-26 12:18:48.882 [grpc-nio-worker-ELG-1-2] DEBUG com.levi.interceptor.CustomClientStreamTracer - 获得响应消息未压缩大小 10 
2025-07-26 12:18:48.882 [grpc-default-executor-1] INFO  com.levi.GrpcClient - 接收到的服务端响应为result 2: 
2025-07-26 12:18:48.885 [grpc-default-executor-1] INFO  com.levi.GrpcClient - 服务端响应完成

可以看到没啥问题，而且我们发现，这个拦截器他是一个消息就拦截完整的一次。所以每一段都输出了一遍。
下面我们来看服务端的流式拦截器开发。

2、服务端流式拦截器

服务端的实现其实和客户端基本是对应的，而且还更加简单。
他的目的还是为了拦截请求和响应。而且涉及到的类。

1. ServerStreamTracer 拦截实现
   inbound  请求的拦截
   outbound 响应的拦截 
2. ServerStreamTracerFactory进行创建我们的拦截实现

你会发现它这里没有拦截器那个东西，他最上层就是工厂，直接把工厂发布到服务端发布即可，下面我们来编程。

package com.levi.interceptor;

import io.grpc.Metadata;
import io.grpc.ServerStreamTracer;
import lombok.extern.slf4j.Slf4j;

/**
 * 拦截器工厂，创建我们的拦截实现
 */
public class CustomServerStreamFactory extends ServerStreamTracer.Factory {
    @Override
    public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
        return new CustomServerStreamTracer();
    }
}

/**
 * 服务端流式拦截实现
 */
@Slf4j
class CustomServerStreamTracer extends ServerStreamTracer {
    //inbound 拦截请求
    @Override
    public void inboundMessage(int seqNo) {
        super.inboundMessage(seqNo);
    }

    @Override
    public void inboundWireSize(long bytes) {
        super.inboundWireSize(bytes);
    }

    @Override
    public void inboundMessageRead(int seqNo, long optionalWireSize, long optionalUncompressedSize) {
        log.debug("server: 获得client发送的请求消息 ...{} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);
        super.inboundMessageRead(seqNo, optionalWireSize, optionalUncompressedSize);
    }

    @Override
    public void inboundUncompressedSize(long bytes) {
        super.inboundUncompressedSize(bytes);
    }

    //outbound 拦截请求

    @Override
    public void outboundMessage(int seqNo) {
        super.outboundMessage(seqNo);
    }


    @Override
    public void outboundMessageSent(int seqNo, long optionalWireSize, long optionalUncompressedSize) {
        log.debug("server: 响应数据的拦截 ...{} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);
        super.outboundMessageSent(seqNo, optionalWireSize, optionalUncompressedSize);
    }

    @Override
    public void outboundWireSize(long bytes) {
        super.outboundWireSize(bytes);
    }

    @Override
    public void outboundUncompressedSize(long bytes) {
        super.outboundUncompressedSize(bytes);
    }
}

此时我们只需要发布出去即可。

package com.levi;

import com.levi.interceptor.CustomServerStreamFactory;
import com.levi.service.HelloServiceImpl;
import io.grpc.Server;
import io.grpc.ServerBuilder;

import java.io.IOException;

public class GrpcServer {
    public static void main(String[] args) throws InterruptedException, IOException {
        ServerBuilder<?> serverBuilder = ServerBuilder.forPort(9000);
        serverBuilder.addService(new HelloServiceImpl());
        // 发布拦截器工厂
        serverBuilder.addStreamTracerFactory(new CustomServerStreamFactory());
        Server server = serverBuilder.build();
        server.start();
        server.awaitTermination();
    }
}

此时就完成了服务端的拦截，而且他的调用也是一个消息就拦截一次。
他的outBound和inBound都是站在自己的角度来说的，对于服务端，out就是离开，也就是服务端响应客户端的操作。
int就是进入，其实就是客户端请求进来的操作。

四、总结

拦截器实际上是一个比较重要的能力，不管是在grpc还是我们的其他框架中比如springmvc，因为他可以和主线业务解耦合。其实是个增强的能力。所以我们可以基于拦截器做很多业务之外的，比如在拦截器里面根据请求限流，根据请求做鉴权等等。后续我们会基于实际项目来做一个实战，拭目以待吧。