gRPC背压流控、压缩及JSON通信【知识笔记】

目录

1. 一、压缩
2. 1.Server端所有方法压缩
3. 2.Server单独方法压缩
4. 3.Client请求内容压缩
5. 二、使用JSON通信
6. 1.方法描述使用JSON编译
7. 2.JSON编译具体过程
8. 三、手动流量控制
9. 1.Consuming Side
10. 2.Producing Side
11. 四、系列文章

本文继续整理gRPC的使用，走查解读官方给出的压缩示例、使用JSON通信以及手动流量控制。

一、压缩

1.Server端所有方法压缩

1. server = ServerBuilder.forPort(port)
2. .intercept(new ServerInterceptor() {
3. @Override
4. public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
5. ServerCallHandler<ReqT, RespT> next) {
6. // @1 在拦截器中设置压缩算法
7. call.setCompression("gzip");
8. return next.startCall(call, headers);
9. }
10. })
11. .addService(new GreeterImpl())
12. .build()
13. .start();

备注：如果需要在Server端所有方法进行压缩，可以在ServerInterceptor拦击器中通过setCompression进行设置。

2.Server单独方法压缩

如果不想对所有的方法传输内容压缩，gPRC提供了单独方法的压缩。

1. int port = 50051;
2. server = ServerBuilder.forPort(port)
3. .addService(new GreeterImpl())
4. .build()
5. .start();
6. static classGreeterImplextendsGreeterGrpc.GreeterImplBase{
7. 

8. @Override
9. public void sayHello(HelloRequest req, StreamObserver<HelloReply> plainResponseObserver) {
10. ServerCallStreamObserver<HelloReply> responseObserver =
11. (ServerCallStreamObserver<HelloReply>) plainResponseObserver;
12. // @1 对单个方法传输内容进行压缩
13. responseObserver.setCompression("gzip");
14. HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
15. responseObserver.onNext(reply);
16. responseObserver.onCompleted();
17. }
18. }

备注：单个方法的压缩通过ServerCallStreamObserver的setCompression进行单独设置。

3.Client请求内容压缩

客户端对请求内容进行压缩，下面示例通过gzip进行压缩。

1. publicvoidgreet(String name) {
2. HelloRequest request = HelloRequest.newBuilder().setName(name).build();
3. HelloReply response;
4. try {
5. // @1 对请求内容设置压缩类型
6. response = blockingStub.withCompression("gzip").sayHello(request);
7. } catch (StatusRuntimeException e) {
8. logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
9. return;
10. }

二、使用JSON通信

gRPC可以通过Json格式进行通信，虽然并不建议这么做，Json的效率要远低于ProtoBuf。看下示例是如何通过Json格式通信的。

1.方法描述使用JSON编译

对方法的出参和入参使用JSON适配器，示例中通过MethodDescriptor.toBuilder重写出入参数的解析格式。

1. static final MethodDescriptor<HelloRequest, HelloReply> METHOD_SAY_HELLO =
2. GreeterGrpc.getSayHelloMethod()
3. .toBuilder(
4. // @1 请求参数使用JSON编译 JsonMarshaller.jsonMarshaller(HelloRequest.getDefaultInstance()),
5. // @2 返回参数使用JSON编译
6. JsonMarshaller.jsonMarshaller(HelloReply.getDefaultInstance()))
7. .build();

2.JSON编译具体过程

既然通过对方法的出入参数编译成JSON格式，看下gRPC是如何做的呢？

1. public static <T extends Message> Marshaller<T> jsonMarshaller(final T defaultInstance) {
2. final Parser parser = JsonFormat.parser();
3. final Printer printer = JsonFormat.printer();
4. return jsonMarshaller(defaultInstance, parser, printer);
5. }
6. public static <T extends Message> Marshaller<T> jsonMarshaller(
7. final T defaultInstance, final Parser parser, final Printer printer) {
8. final Charset charset = Charset.forName("UTF-8");
9. return new Marshaller<T>() {
10. @Override
11. public InputStream stream(T value) {
12. try {
13. // @1 通过printer.print将出入参数转换为JSON格式
14. return new ByteArrayInputStream(printer.print(value).getBytes(charset));
15. } catch (InvalidProtocolBufferException e) {
16. // ...
17. }
18. }
19. // ....
20. }

备注：在JsonFormat.print方法中进行具体的请求/返回参数转换为JSON的具体实现。

请求转换JSON格式截图

返回转换JSON格式截图

3.Client使用JSON格式的方法描述

1. public HelloReply sayHello(HelloRequest request) {
2. // @1 使用JSON格式的方法描述METHOD_SAY_HELLO
3. return blockingUnaryCall(
4. getChannel(), METHOD_SAY_HELLO, getCallOptions(), request);
5. }

4.Server使用JSON格式的方法描述

1. public ServerServiceDefinition bindService() {
2. return ServerServiceDefinition
3. .builder(GreeterGrpc.getServiceDescriptor().getName())
4. // @1 使用JSON格式的方法描述METHOD_SAY_HELLO
5. .addMethod(HelloJsonClient.HelloJsonStub.METHOD_SAY_HELLO,
6. asyncUnaryCall(
7. new UnaryMethod<HelloRequest, HelloReply>() {
8. @Override
9. publicvoidinvoke(
10. HelloRequest request, StreamObserver<HelloReply> responseObserver) {
11. sayHello(request, responseObserver);
12. }
13. }))
14. .build();
15. }

三、手动流量控制

gRPC的流量控制基于HTTP/2的流量控制，即背压模式。关于gRPC和HTTP/2背压模式原理和关系，请看下面摘录。

1. At the bottom is the HTTP/2's byte-based flow control. HTTP/2 works on streams of bytes and is completely unaware of gRPC messages or reactive streams. By default, the stream consumer allocates a budget of 65536 bytes.
2. The stream producer can send up to this many bytes before backpressure engages. As the consumer reads bytes, WINDOW_UPDATE messages are sent to the producer to increase its send budget.
3. 

4. In the middle is the gRPC-Java message-based flow control. gRPC's flow control adapts the stream-based flow control of HTTP/2 to a message-based flow control model.
5. Importantly, gRPC's flow control is aware of how it interacts with HTTP/2 and the network.
6. 

7. On producing side, an on-ready handler reads a message, serializes it into bytes using protobuf, and then queues it up for transmission over the HTTP/2 byte stream.
8. If there is insuficient room in the HTTP/2 flow control window to transmit, backpressure engages an no more messages are requested from the producer until space becomes available.
9. 

10. On the consuming side, each time a consumer calls request(x), gRPC attempts to read and deserialize x messages from the HTTP/2 stream.
11. Since the size of a protobuf encoded message is variable, there is not a one-to-one correlation between pulling messages from gRPC and pulling bytes over HTTP/2.
12. 


1.Consuming Side

1. publicstaticvoidmain(String[] args) throws InterruptedException, IOException {
2. // @1 Server端服务实现
3. StreamingGreeterGrpc.StreamingGreeterImplBase svc = new StreamingGreeterGrpc.StreamingGreeterImplBase() {
4. @Override
5. public StreamObserver<HelloRequest> sayHelloStreaming(final StreamObserver<HelloReply> responseObserver) {
6. final ServerCallStreamObserver<HelloReply> serverCallStreamObserver =
7. (ServerCallStreamObserver<HelloReply>) responseObserver;
8. // @2 禁止自动流控模式，开启手动流控
9. serverCallStreamObserver.disableAutoInboundFlowControl();
10. // @3 背压模式流控，当消费端有足够空间时将会回调OnReadyHandler
11. // 默认空间大小为65536字节
12. classOnReadyHandlerimplementsRunnable{
13. private boolean wasReady = false;
14. 

15. @Override
16. publicvoidrun() {
17. if (serverCallStreamObserver.isReady() && !wasReady) {
18. wasReady = true;
19. logger.info("READY");
20. // @4 向HTTP/2流请求读取并解压(x)条消息
21. // 即发信号通知发送端发送继续发消息
22. serverCallStreamObserver.request(1);
23. }
24. }
25. }
26. final OnReadyHandler onReadyHandler = new OnReadyHandler();
27. serverCallStreamObserver.setOnReadyHandler(onReadyHandler);
28. // @5 处理具体进来的请求
29. return new StreamObserver<HelloRequest>() {
30. @Override
31. publicvoidonNext(HelloRequest request) {
32. try {
33. String name = request.getName();
34. logger.info("--> " + name);
35. Thread.sleep(100);
36. String message = "Hello " + name;
37. logger.info("<-- " + message);
38. HelloReply reply = HelloReply.newBuilder().setMessage(message).build();
39. // @6 向Client发送请求
40. responseObserver.onNext(reply);
41. if (serverCallStreamObserver.isReady()) {
42. // @7 向HTTP/2流请求读取并解压(x)条消息
43. serverCallStreamObserver.request(1);
44. } else {
45. onReadyHandler.wasReady = false;
46. }
47. } catch (Throwable throwable) {
48. //
49. }
50. }
51. @Override
52. publicvoidonError(Throwable t) {
53. t.printStackTrace();
54. responseObserver.onCompleted();
55. }
56. @Override
57. publicvoidonCompleted() {
58. logger.info("COMPLETED");
59. responseObserver.onCompleted();
60. }
61. };
62. }
63. };
64. 

65. final Server server = ServerBuilder
66. .forPort(50051)
67. .addService(svc)
68. .build()
69. .start();

2.Producing Side

1. publicstaticvoidmain(String[] args) throws InterruptedException {
2. final CountDownLatch done = new CountDownLatch(1);
3. ManagedChannel channel = ManagedChannelBuilder
4. .forAddress("localhost", 50051)
5. .usePlaintext()
6. .build();
7. StreamingGreeterGrpc.StreamingGreeterStub stub = StreamingGreeterGrpc.newStub(channel);
8. 

9. ClientResponseObserver<HelloRequest, HelloReply> clientResponseObserver =
10. new ClientResponseObserver<HelloRequest, HelloReply>() {
11. ClientCallStreamObserver<HelloRequest> requestStream;
12. @Override
13. publicvoidbeforeStart(final ClientCallStreamObserver<HelloRequest> requestStream) {
14. this.requestStream = requestStream;
15. // @1设置手动流量控制
16. requestStream.disableAutoInboundFlowControl();
17. // @2 当Consumer端有足够空间时自动回调
18. // 序列化protobuf先发送到缓存区（还未到Server端）
19. // Server端需要调用request()向Client拉取消息
20. requestStream.setOnReadyHandler(new Runnable() {
21. Iterator<String> iterator = names().iterator();
22. @Override
23. publicvoidrun() {
24. while (requestStream.isReady()) {
25. if (iterator.hasNext()) {
26. String name = iterator.next();
27. logger.info("--> " + name);
28. HelloRequest request = HelloRequest.newBuilder().setName(name).build();
29. // @3 将消息发送到缓存区
30. requestStream.onNext(request);
31. } else {
32. // @4 标记Client发送完成
33. requestStream.onCompleted();
34. }
35. }
36. }
37. });
38. }
39. 

40. @Override
41. publicvoidonNext(HelloReply value) {
42. // @5 接受Server端返回信息
43. logger.info("<-- " + value.getMessage());
44. // @6 通知Client继续发送
45. requestStream.request(1);
46. }
47. 

48. @Override
49. publicvoidonError(Throwable t) {
50. t.printStackTrace();
51. done.countDown();
52. }
53. 

54. @Override
55. publicvoidonCompleted() {
56. logger.info("All Done");
57. done.countDown();
58. }
59. };
60. stream processing.
61. stub.sayHelloStreaming(clientResponseObserver);
62. 

63. done.await();
64. 

65. channel.shutdown();
66. channel.awaitTermination(1, TimeUnit.SECONDS);
67. }

四、系列文章

gRPC示例初探【实战笔记】


gRPC四种类型示例分析【知识笔记】


gRPC中Header传值与错误拦截处理【知识笔记】


---

「瓜农老梁  学习同行」