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

Posted 2021-04-11 瓜农老梁

目录

一、前言
二、ProtoBuf定义
三、代码结构
 1.服务端
 2.客户端
四、交互走查
 1.简单gRPC交互（UNARY）
 2.服务端到客户端流式交互
 3.客户端到服务端流式交互
 4.双向流式RPC
五、系列文章

一、前言

本文分析下gRPC支持类型的示例，Protobuf生成代码详见前面文章“Google Protocol Buffers三两事” 以及 Maven插件使用参见前面文章 “gRPC示例初探”；具体链接见本文结尾系列文章。gRPC提供四种服务类型，分别为：简单RPC、服务端到客户端流式RPC、客户端到服务端流式RPC、双向流式RPC。将“route_guide.proto”拷贝到工程目录，Maven编译时会生成代码。

二、ProtoBuf定义

下面Protobuf定义了gRPC提供的四种服务类型，走查下内容。

// @1 使用proto3语法
syntax = "proto3";
// @2 生成多个类
option java_multiple_files = true;
// @3 生成java类所在的包
option java_package = "io.grpc.examples.routeguide";
// @4 生成外层类类名
option java_outer_classname = "RouteGuideProto";
// @5 Objective-C类的前缀
option objc_class_prefix = "RTG";
// @6 .proto包名
package routeguide;
// @7 定义RPC服务RouteGuide
service RouteGuide {
 // @8 简单RPC接受Point参数返回Feature类型对象
 rpc GetFeature(Point) returns (Feature) {}
 // @9 服务端到客户端流式RPC，接受Rectangle对象参数，返回批量Feature数据
 rpc ListFeatures(Rectangle) returns (stream Feature) {}
 // @10 客户端到服务端流式RPC，接受批量Point数据，返回RouteSummary类型对象
 rpc RecordRoute(stream Point) returns (RouteSummary) {}
 // @11 双向流式RPC，接受批量RouteNote类型数据，返回批量RouteNote类型数据
 rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}
}
message Point {
 int32 latitude = 1;
 int32 longitude = 2;
}
message Rectangle {
 Point lo = 1;
 Point hi = 2;
}
message Feature {
 string name = 1;
 Point location = 2;
}
message FeatureDatabase {
 repeated Feature feature = 1;
}
message RouteNote {
 Point location = 1;
 string message = 2;
}
message RouteSummary {
 int32 point_count = 1;
 int32 feature_count = 2;
 int32 distance = 3;
 int32 elapsed_time = 4;
}

三、代码结构

编译工具生成了Protobuf定义的Message对应的类和Builder类、gRPC服务端对外提供的接口、客户端调用服务端的存根。

1.服务端

启动gRPC Server

public static void main(String[] args) throws Exception {
 RouteGuideServer server = new RouteGuideServer(8980);
 server.start();
 server.blockUntilShutdown();
 } // @1 启动Server监听处理客户端请求

 public RouteGuideServer(int port, URL featureFile) throws IOException {
 this(ServerBuilder.forPort(port), port, RouteGuideUtil.parseFeatures(featureFile));
 } // @2 将数据从文件route_guide_db.json读出

 public RouteGuideServer(ServerBuilder<?> serverBuilder, int port, Collection<Feature> features) {
 this.port = port;
 server = serverBuilder.addService(new RouteGuideService(features))
 .build();
 } // @3 将自定义实现的RouteGuideService注册到Server

实现gRPC服务接口

编译工具生成了gRPC服务端对外提供的接口，我们使用时需要实现该接口即可，即实际的Server端处理逻辑。

private static class RouteGuideService extends RouteGuideGrpc.RouteGuideImplBase {
}
// @1 RouteGuideGrpc.RouteGuideImplBase由编译器生成Server端接口类
// @2 RouteGuideService由用户实现的类处理Server端业务逻辑

小结：在服务端我们需要做实现生成的服务接口，并将该服务实现类注册到gRPC Server中。

2.客户端

构建通道和存根

 public static void main(String[] args) throws InterruptedException {
 String target = "localhost:8980";
 List<Feature> features;
 // @1 读取route_guide_db.json测试数据
 features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());
 // @2 构建与Server的RPC通道
 ManagedChannel channel = ManagedChannelBuilder.forTarget(target).usePlaintext().build();
 RouteGuideClient client = new RouteGuideClient(channel);
 }

 public RouteGuideClient(Channel channel) {
 // @3 构建同步调用存根
 blockingStub = RouteGuideGrpc.newBlockingStub(channel);
 // @4 构建异步调用存根
 asyncStub = RouteGuideGrpc.newStub(channel);
 }

四、交互走查

1 简单gRPC交互（UNARY）

客户端调用

RouteGuideClient client = new RouteGuideClient(channel);
client.getFeature(409146138, -746188906);

Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();
// @1 使用同步调用存根向服务端发起请求，入参为普通对象
feature = blockingStub.getFeature(request);

System.out.printf("Found feature called %s, at %s, %s!%n",
 feature.getName(),
 RouteGuideUtil.getLatitude(feature.getLocation()),
 RouteGuideUtil.getLongitude(feature.getLocation()));

服务端响应

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
 // @2 通过responseObserver.onNext向客户端发送消息
 responseObserver.onNext(checkFeature(request));
 // @3 标记服务端响应完成
 responseObserver.onCompleted();
}

输出内容

Found feature called Berkshire Valley Management Area Trail, Jefferson, NJ, USA, at 40.9146138, -74.6188906!

备注：使用观察者模式响应客户端请求。

2.服务端到客户端流式交互

客户端调用

 RouteGuideClient client = new RouteGuideClient(channel);
 client.listFeatures(400000000, -750000000, 420000000, -730000000);

 public void listFeatures(int lowLat, int lowLon, int hiLat, int hiLon) {
 Iterator<Feature> features;
 // @1 客户端调用同步存根向服务端发起调用
 features = blockingStub.listFeatures(request);
 for (int i = 1; features.hasNext(); i++) {
 Feature feature = features.next();
 info("Result #" + i + ": {0}", feature);
 }
 }

服务端响应

public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
 // ...
 for (Feature feature : features) {
 if (!RouteGuideUtil.exists(feature)) {
 continue;
 }
 int lat = feature.getLocation().getLatitude();
 int lon = feature.getLocation().getLongitude();
 if (lon >= left && lon <= right && lat >= bottom && lat <= top) {
 // @2 循环调用responseObserver.onNext向客户端回调发送数据
 responseObserver.onNext(feature);
 }
 }
 // @3 标记服务端响应完成
 responseObserver.onCompleted();
}

输出内容

Result #62: name: "3387 Richmond Terrace, Staten Island, NY 10303, USA"
location {
 latitude: 406411633
 longitude: -741722051
}
Result #63: name: "261 Van Sickle Road, Goshen, NY 10924, USA"
location {
 latitude: 413069058
 longitude: -744597778
}
//...

备注：服务端到客户端流式交互。即：SERVER_STREAMING。客户端通过存根发起RPC调用，由服务端多次调用onNext回调客户端完成响应。

3.客户端到服务端流式交互

客户端调用

RouteGuideClient client = new RouteGuideClient(channel);
client.recordRoute(features, 10);

public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
 info("*** RecordRoute");
 final CountDownLatch finishLatch = new CountDownLatch(1);
 // @1 创建responseObserver用于服务端回调客户端
 StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {
 @Override
 // @2 响应服务端responseObserver.onNext回调
 public void onNext(RouteSummary summary) {
 // print
 }
 @Override
 // @3 响应服务端responseObserver.onError回调
 public void onError(Throwable t) {
 // print
 }
 @Override
 // @4 响应服务端responseObserver.onCompleted的回调
 public void onCompleted() {
 // print
 finishLatch.countDown();
 }
 };
// @5 通过异步存根发起调用，参数为响应观察者responseObserver
StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
 try {
 for (int i = 0; i < numPoints; ++i) {
 int index = random.nextInt(features.size());
 Point point = features.get(index).getLocation();
 // print
 // @6 多次调用requestObserver.onNext向服务端写入数据
 requestObserver.onNext(point);
 Thread.sleep(random.nextInt(1000) + 500);
 if (finishLatch.getCount() == 0) {
 return;
 }
 }
 } catch (RuntimeException e) {
 requestObserver.onError(e);
 throw e;
 }
 // @7 标记客户端写入结束
 requestObserver.onCompleted();
 }

服务端响应

@Override
public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {
 // @1 构造观察者与客户端交互
 return new StreamObserver<Point>() {
 int pointCount;
 int featureCount;
 int distance;
 Point previous;
 final long startTime = System.nanoTime();

 @Override
 // @2 响应客户端
 public void onNext(Point point) {
 pointCount++;
 if (RouteGuideUtil.exists(checkFeature(point))) {
 featureCount++;
 }
 if (previous != null) {
 distance += calcDistance(previous, point);
 }
 previous = point;
 }

 @Override
 public void onError(Throwable t) {
 logger.log(Level.WARNING, "recordRoute cancelled");
 }

 @Override
 // @3 在客户端调用requestObserver.onCompleted()时触发，标记服务端处理完成
 public void onCompleted() {
 long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);
 // @4 回调客户端responseObserver.onNext
 responseObserver.onNext(RouteSummary.newBuilder()
 .setPointCount(pointCount)
 .setFeatureCount(featureCount).setDistance(distance)
 .setElapsedTime((int) seconds).build());
 // @5 回调客户端responseObserver.onCompleted标记完成
 responseObserver.onCompleted();
 }
 };
}

输出内容

*** RecordRoute
Visiting point 40.213, -74.361
Visiting point 41.478, -74.062
Visiting point 40.915, -74.619
Visiting point 40.213, -74.361
Visiting point 40.408, -74.612
Visiting point 40.431, -74.028
Visiting point 40.964, -74.602
Visiting point 40.007, -74.679
Visiting point 41.268, -74.261
Visiting point 40.812, -74.4
Finished trip with 10 points. Passed 5 features. Travelled 761,415 meters. It took 110 seconds.
Finished RecordRoute

小结：客户端到服务端流式交互，即：CLIENT_STREAMING。客户端由异步存根asyncStub发起调用，参数为“responseObserver”；服务端通过onNext响应客户端请求，在客户端触发写入结束响应onCompleted后，服务端onCompleted被触发，调用响应观察者“responseObserver”回调到客户端完成结束操作。

4.双向流式RPC

客户端调用

RouteGuideClient client = new RouteGuideClient(channel);
CountDownLatch finishLatch = client.routeChat();

public CountDownLatch routeChat() {
final CountDownLatch finishLatch = new CountDownLatch(1);

StreamObserver<RouteNote> requestObserver =
    // @1 使用异步存根asyncStub向服务端发起调用
    // 调用参数为响应观察者即：responseObserver
 asyncStub.routeChat(new StreamObserver<RouteNote>() {
 @Override
 // @2 响应服务端responseObserver回调onNext
 public void onNext(RouteNote note) {
 // print...
 }
 @Override
 public void onError(Throwable t) {
 // print...
 finishLatch.countDown();
 }
 @Override
 // @3 响应服务端responseObserver回调onCompleted
 public void onCompleted() {
 info("Finished RouteChat");
 finishLatch.countDown();
 }
 });

RouteNote[] requests =
 {newNote("First message", 0, 0), newNote("Second message", 0, 1),
 newNote("Third message", 1, 0), newNote("Fourth message", 1, 1)};

for (RouteNote request : requests) {
 // @4 循环调用requestObserver.onNext向服务端写入消息
 requestObserver.onNext(request);
}

// @5 标记客户端写入完成
requestObserver.onCompleted();

return finishLatch;
}

服务端响应

@Override
public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {
 return new StreamObserver<RouteNote>() {
 @Override
 // @1 当客户端调用requestObserver.onNext触发接受数据
 public void onNext(RouteNote note) {
 List<RouteNote> notes = getOrCreateNotes(note.getLocation());
 for (RouteNote prevNote : notes.toArray(new RouteNote[0])) {
 // @2 循环调用responseObserver.onNext回调客户端
 responseObserver.onNext(prevNote);
 }
 notes.add(note);
 }

 @Override
 public void onError(Throwable t) {
 logger.log(Level.WARNING, "routeChat cancelled");
 }

 @Override
 // @3 客户端调用requestObserver.onCompleted()时触发
 public void onCompleted() {
 // @4 回调客户端标记完成
 responseObserver.onCompleted();
 }
 };
}

输出内容

信息: *** RouteChat
信息: Sending message "First message" at 0, 0
信息: Sending message "Second message" at 0, 1
信息: Sending message "Third message" at 1, 0
信息: Sending message "Fourth message" at 1, 1
信息: Got message "First message" at 0, 0
信息: Got message "First message" at 0, 0
信息: Got message "First message" at 0, 0
信息: Got message "First message" at 0, 0
信息: Got message "First message" at 0, 0
信息: Got message "Second message" at 0, 1
信息: Got message "Second message" at 0, 1
信息: Got message "Second message" at 0, 1
信息: Got message "Second message" at 0, 1
信息: Got message "Second message" at 0, 1
信息: Got message "Third message" at 1, 0
信息: Got message "Third message" at 1, 0
信息: Got message "Third message" at 1, 0
信息: Got message "Third message" at 1, 0
信息: Got message "Third message" at 1, 0
信息: Got message "Fourth message" at 1, 1
信息: Got message "Fourth message" at 1, 1
信息: Got message "Fourth message" at 1, 1
信息: Got message "Fourth message" at 1, 1
信息: Got message "Fourth message" at 1, 1
信息: Finished RouteChat

小结：双向流式RPC，即：BIDI_STREAMING。客户端和服务端均通过StreamObserver来交互，客户端发起时传入responseObserver，服务端可以通过responseObserver对客户端进行回调。

五、系列文章

---

「瓜农老梁  学习同行」