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

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

news 2025/7/27 5:47:06

一、简介

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

流式通信方式：主要应用 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.xdnf.cn/news/1188829.html

相关文章：

  • 15.6 DeepSpeed+Transformers实战：LLaMA-7B训练效率提升210%，显存直降73%
  • 前端设计中如何在鼠标悬浮时同步修改块内样式
  • Cgroup 控制组学习(一)
  • 基于深度学习的图像分类：使用Inception-v3实现高效分类
  • 前端基础知识Vue系列 - 29（怎么处理vue项目中的错误）
  • vue 脚手架配置代理
  • RS485转Profinet网关配置指南：高效启动JRT激光测距传感器测量模式
  • 深入解析三大Web安全威胁：文件上传漏洞、SQL注入漏洞与WebShell
  • Qt 线程池设计与实现
  • HTML 音频/视频
  • 从一个“诡异“的C++程序理解状态机、防抖与系统交互
  • 2025年02月11日 Go生态洞察：Go 1.24 发布亮点全面剖析
  • 二叉搜索树（Binary Search Tree）详解与java实现
  • 【RK3568 PWM 子系统（SG90）驱动开发详解】
  • 记录和分享抓取的数字货币和大A时序数据
  • k8s:将打包好的 Kubernetes 集群镜像推送到Harbor私有镜像仓库
  • 容器化成本优化：K8s资源请求与限制的黄金法则——从资源画像分析到25%成本削减的实战指南
  • python面向对象编程详解
  • k8s之控制器详解
  • Go语言unsafe包深度解析
  • Go 多模块仓库标签管理教程
  • 嵌入式硬件篇---zigbee无线串口通信问题解决方法
  • Android 修改系统时间源码阅读
  • Linux的生态与软件安装
  • 主要分布在腹侧海马体（vHPC）CA1区域（vCA1）的混合调谐细胞（mixed-tuning cells）对NLP中的深层语义分析的积极影响和启示
  • 车载诊断刷写 --- Flash关于擦除和写入大小
  • 【MySQL】深入浅出事务：保证数据一致性的核心武器
  • 深度解析 noisereduce：开源音频降噪库实践
  • 【影刀RPA_初级课程_我的第一个机器人】
  • LeetCode｜Day26｜191. 位 1 的个数｜Python刷题笔记