grpc流服务实现传输文件及源码解析

文章目录

前言
一、grpc是什么？
二、简单的grpc接口
三、流服务接口
四、流服务原理
五、总结
六、参考资料
七、下载

前言

由于需要传送200M的压缩包文件，之前的同步存根Stub无法满足需求。再经过调研后，发现客户端流能够很好的解决这个问题。
注：流服务本质上就是通过异步存根Stub来实现的，具体到服务端和客户端只需实现观察者的接口来处理业务逻辑即可

一、grpc是什么？

在 gRPC 里客户端应用可以像调用本地对象一样直接调用另一台不同的机器上服务端应用的方法，使得您能够更容易地创建分布式应用和服务。与许多 RPC 系统类似，gRPC 也是基于以下理念：定义一个服务，指定其能够被远程调用的方法（包含参数和返回类型）。在服务端实现这个接口，并运行一个 gRPC 服务器来处理客户端调用。在客户端拥有一个存根能够像服务端一样的方法。

二、简单的grpc接口

一个grpc接口包括以下几个部分
注：由于不需要流，所以此处使用的为同步阻塞存根

1.定义proto文件

service RouteGuide {
  // 一个简单的rpc接口
  rpc GetFeature(Point) returns (Feature) {}
}
// 消息定义
message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}
message Feature {
  string name = 1;

  Point location = 2;
}

2.服务端代码

public void getFeature(Point request, StreamObserver responseObserver) {
            responseObserver.onNext(checkFeature(request));
            responseObserver.onCompleted();
        }

3.客户端代码

public void getFeature(int lat, int lon) {
        Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();

        Feature feature = blockingStub.getFeature(request);
        // 对返回数据进行校验
        if (RouteGuideUtil.exists(feature)) {
            info("Found feature called "{0}" at {1}, {2}",
                    feature.getName(),
                    RouteGuideUtil.getLatitude(feature.getLocation()),
                    RouteGuideUtil.getLongitude(feature.getLocation()));
        } else {
            info("Found no feature at {0}, {1}",
                    RouteGuideUtil.getLatitude(feature.getLocation()),
                    RouteGuideUtil.getLongitude(feature.getLocation()));
        }
    }

三、流服务接口

此处使用为客户端流来实现文件传输的

1.proto文件

service RouteGuide {
  // 客户端文件流例子
  rpc sendFile(stream FileInfo) returns (Info) {}
}
message FileInfo {
    int32 index = 1;

    bytes arrs = 2;
}

message Info {
    string msg = 1;
}

2.服务端代码

		// 测试文件流
        @Override
        public StreamObserver sendFile(StreamObserver responseObserver) {

            try {
                return new StreamObserver() {
                    final long startTime = System.nanoTime();
                    OutputStream os = new FileOutputStream(new File(System.currentTimeMillis() + ".zip"));
                    @Override
                    public void onNext(FileInfo fileInfo) {
                        try {
                            logger.log(Level.INFO, "接收到文件流");
                            fileInfo.getArrs().writeTo(os);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.out.println("发生错误");
                        logger.log(Level.WARNING, "sendFile cancelled");
                    }

                    @Override
                    public void onCompleted() {
                        System.out.println("完成");
                        // 关闭流
                        try {
                            os.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);
                        responseObserver.onNext(Info.newBuilder().setMsg("success, spend time :" + seconds).build());
                        responseObserver.onCompleted();
                    }
                };
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

3.客户端代码

public void sendFile() {
        final CountDownLatch finishLatch = new CountDownLatch(1);
        StreamObserver responseObserver = new StreamObserver() {
            @Override
            public void onNext(Info info) {
                logger.info("end :"+info.getMsg());
            }

            @Override
            public void onError(Throwable t) {
                warning("sendFile Failed: {0}", Status.fromThrowable(t));
                if (testHelper != null) {
                    testHelper.onRpcError(t);
                }
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                info("Finished RecordRoute");
                finishLatch.countDown();
            }
        };

        StreamObserver requestObserver = asyncStub.sendFile(responseObserver);
        try {
            InputStream is = new FileInputStream(new File("/home/test/test.zip"));
            byte[] buff = new byte[2048];
            int len;
            int index = 0;
            while ((len = is.read(buff)) != -1) {
                requestObserver.onNext(FileInfo.newBuilder().setIndex(index).setArrs(ByteString.copyFrom(buff)).build());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        // Mark the end of requests
        requestObserver.onCompleted();

        // Receiving happens asynchronously
        try {
            if (!finishLatch.await(1, TimeUnit.MINUTES)) {
                warning("send file can not finish within 1 minutes");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("sendFile success");
    }

4.运行结果

四、流服务原理

直接看源码

1.调用开始的地方
ServerCalls.class（io.grpc:grpc-stub:1.26.0）

public Listener startCall(ServerCall call, Metadata headers) {
            ServerCalls.ServerCallStreamObserverImpl responseObserver = new ServerCalls.ServerCallStreamObserverImpl(call);
            // 通过RouteGuideGrpc的invoke方法，拿到StreamObserver
            StreamObserver requestObserver = this.method.invoke(responseObserver);
            responseObserver.freeze();
            if (responseObserver.autoFlowControlEnabled) {
                call.request(1);
            }
            // 将观察者添加到上下文中
            return new ServerCalls.StreamingServerCallHandler.StreamingServerCallListener(requestObserver, responseObserver, call);
        }

2.实际添加观察者到上下文的代码
ServerCallImpl.class（io.grpc:grpc-core:1.26.0）

public ServerStreamListenerImpl(ServerCallImpl call, Listener listener, CancellableContext context) {
			// 校验
            this.call = (ServerCallImpl)Preconditions.checkNotNull(call, "call");
            this.listener = (Listener)Preconditions.checkNotNull(listener, "listener must not be null");
            this.context = (CancellableContext)Preconditions.checkNotNull(context, "context");
            // 实际添加
            this.context.addListener(new CancellationListener() {
                public void cancelled(Context context) {
                    ServerStreamListenerImpl.this.call.cancelled = true;
                }
            }, MoreExecutors.directExecutor());
        }

Context.class （io.grpc:grpc-context:1.26.0）

public void addListener(Context.CancellationListener cancellationListener, Executor executor) {
        checkNotNull(cancellationListener, "cancellationListener");
        checkNotNull(executor, "executor");
        if (this.canBeCancelled()) {
            Context.ExecutableListener executableListener = new Context.ExecutableListener(executor, cancellationListener);
            synchronized(this) {
                if (this.isCancelled()) {
                    executableListener.deliver();
                } else if (this.listeners == null) {
                    this.listeners = new ArrayList();
                    this.listeners.add(executableListener);
                    if (this.cancellableAncestor != null) {
                        this.cancellableAncestor.addListener(this.parentListener, Context.DirectExecutor.INSTANCE);
                    }
                } else {
                    this.listeners.add(executableListener);
                }
            }
        }
    }

3.每次收到客户端的消息调用发起的地方
在ServerImpl.class（io.grpc:grpc-core:1.26.0）中，主要为getListener().messagesAvailable(producer)发起调用

public void messagesAvailable(final MessageProducer producer) {
      PerfMark.startTask("ServerStreamListener.messagesAvailable", tag);
      final Link link = PerfMark.linkOut();

      final class MessagesAvailable extends ContextRunnable {

        MessagesAvailable() {
          super(context);
        }

        @Override
        public void runInContext() {
          PerfMark.startTask("ServerCallListener(app).messagesAvailable", tag);
          PerfMark.linkIn(link);
          try {
          	// 获取当前的监听者队列中的观察者，并消费客户端发来的消息
            getListener().messagesAvailable(producer);
          } catch (RuntimeException e) {
            internalClose();
            throw e;
          } catch (Error e) {
            internalClose();
            throw e;
          } finally {
            PerfMark.stopTask("ServerCallListener(app).messagesAvailable", tag);
          }
        }
      }

      try {
        callExecutor.execute(new MessagesAvailable());
      } finally {
        PerfMark.stopTask("ServerStreamListener.messagesAvailable", tag);
      }
    }

五、总结以上就是我对grpc流服务的一些理解了，欢迎在评论区留言讨论～六、参考资料

gRPC 官方文档中文版V1.0
官方github地址

七、下载

CSDN下载链接：下载

grpc流服务实现传输文件及源码解析

Go语言相关栏目本月热门文章