当前位置： 首页 > news >正文

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

news 2025/7/27 8:21:29

一、简介

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

流式通信方式：主要应用 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 {@Overridepublic StreamObserver<HelloProto.HelloRequest> hello1(StreamObserver<HelloProto.HelloRespnose> responseObserver) {return new StreamObserver<HelloProto.HelloRequest>() {@Overridepublic void onNext(HelloProto.HelloRequest helloRequest) {log.debug("客户端的请求消息为:{} ", helloRequest.getName());}@Overridepublic void onError(Throwable throwable) {}@Overridepublic 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>() {@Overridepublic void onNext(HelloProto.HelloRespnose helloRespnose) {log.info("接收到的服务端响应为{}: ", helloRespnose.getResult());}@Overridepublic void onError(Throwable throwable) {}// 服务端这边的onCompleted方法会触发这个@Overridepublic 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 {@Overridepublic <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 {@Overridepublic 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);}@Overridepublic 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  对于响应相关操作的拦截@Overridepublic void inboundHeaders() {log.debug("用于获得响应头....");super.inboundHeaders();}@Overridepublic void inboundMessage(int seqNo) {log.debug("获得响应消息的编号...{} ",seqNo);super.inboundMessage(seqNo);}@Overridepublic void inboundWireSize(long bytes) {log.debug("获得响应消息的大小...{} ",bytes);super.inboundWireSize(bytes);}@Overridepublic void inboundMessageRead(int seqNo, long optionalWireSize, long optionalUncompressedSize) {log.debug("集中获得消息的编号 ，大小 ，未压缩大小 {} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);super.inboundMessageRead(seqNo, optionalWireSize, optionalUncompressedSize);}@Overridepublic void inboundUncompressedSize(long bytes) {log.debug("获得响应消息未压缩大小 {} ",bytes);super.inboundUncompressedSize(bytes);}@Overridepublic 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 {@Overridepublic ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {return new CustomServerStreamTracer();}
}/*** 服务端流式拦截实现*/
@Slf4j
class CustomServerStreamTracer extends ServerStreamTracer {//inbound 拦截请求@Overridepublic void inboundMessage(int seqNo) {super.inboundMessage(seqNo);}@Overridepublic void inboundWireSize(long bytes) {super.inboundWireSize(bytes);}@Overridepublic void inboundMessageRead(int seqNo, long optionalWireSize, long optionalUncompressedSize) {log.debug("server: 获得client发送的请求消息 ...{} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);super.inboundMessageRead(seqNo, optionalWireSize, optionalUncompressedSize);}@Overridepublic void inboundUncompressedSize(long bytes) {super.inboundUncompressedSize(bytes);}//outbound 拦截请求@Overridepublic void outboundMessage(int seqNo) {super.outboundMessage(seqNo);}@Overridepublic void outboundMessageSent(int seqNo, long optionalWireSize, long optionalUncompressedSize) {log.debug("server: 响应数据的拦截 ...{} {} {}", seqNo, optionalWireSize, optionalUncompressedSize);super.outboundMessageSent(seqNo, optionalWireSize, optionalUncompressedSize);}@Overridepublic void outboundWireSize(long bytes) {super.outboundWireSize(bytes);}@Overridepublic 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，因为他可以和主线业务解耦合。其实是个增强的能力。所以我们可以基于拦截器做很多业务之外的，比如在拦截器里面根据请求限流，根据请求做鉴权等等。后续我们会基于实际项目来做一个实战，拭目以待吧。

http://www.dtcms.com/a/299720.html

相关文章：

  • Thinkph6中常用的验证方式实例
  • 【时时三省】(C语言基础)用指向函数的指针作函数参数
  • 网络：应用层
  • 【硬件-笔试面试题】硬件/电子工程师，笔试面试题-30，（知识点：传输线特性阻抗，影响因素）
  • 【web应用】基于Vue3和Spring Boot的课程管理前后端数据交互过程
  • 1、虚拟机安装
  • InfluxDB Flux 查询协议实战应用（二）
  • Linux726 raid0,raid1,raid5；raid 创建、保存、停止、删除
  • Python 程序设计讲义（22）：循环结构——for 循环
  • 使用FRP搭建内网穿透工具，自己公网服务器独享内外网端口转发
  • C++ APM异步编程模式剖析
  • 2025微前端架构研究与实践方案
  • 【6G新技术探索】AG-UI（Agent User Interaction Protocol） 协议介绍
  • Flutter开发实战之动画与交互设计
  • Java 注解（Annotation）详解：从基础到实战，彻底掌握元数据驱动开发
  • 详细介绍MySQL的索引类型
  • mybatis-plus从入门到入土(三):持久层接口之IService
  • 【MySQL】MySQL 缓存方案
  • 【Redis】Linux 配置Redis
  • 基于华为ENSP的OSPFLSA深入浅出-0
  • 从三维Coulomb势到二维对数势的下降法推导
  • Netty中DefaultChannelPipeline源码解读
  • LangChain vs LangGraph：从困惑到清晰的认知之路(扫盲篇)
  • （一）使用 LangChain 从零开始构建 RAG 系统|RAG From Scratch
  • RM-R1: Reward Modeling as Reasoning
  • Java java.util.Scanner 使用教程
  • 工作流的研究方向
  • （Python）文件储存的认识，文件路径（文件储存基础教程）（Windows系统文件路径）（基础教程）
  • 嵌入式分享#27：原来GT911有两个I2C地址（全志T527）
  • 数据湖产品全解析：2025 年主流解决方案选型指南