gRpc指南

本文翻译自官网。原文：https://grpc.io/docs/languages/java/quickstart/

快速开始

下面通过一个简单的样例，让你快速上手基于java的gRpc的使用。

前置条件

• JDK7以上版本

获取示例代码

示例代码是grpc-java的一部分。

1. 从github仓库下载gprc代码压缩文件并解压，或者直接克隆代码：
   $ git clone -b v1.45.1 --depth 1 https://github.com/grpc/grpc-java
2. 进入示例代码路径：
   $ cd gprc-java/examples

运行示例

在示例代码路径下：

编译客户端和服务端
$ ./gradlew installDist

运行服务端

$ ./build/install/examples/bin/hello-world-server
INFO: Server started, listening on 50051

打开另一个命令行终端，运行客户端

$ ./build/install/examples/bin/hello-world-client
INFO: Will try to greet world ...
INFO: Greeting: Hello world

至此，一个client-server模式的gRPC应用就算运行起来了。

更新gRPC服务

在这一部分，你将通过添加另外的服务方法来更新应用。
gRpc服务通过protocol buffers来定义。你需要知道的是，服务端和客户端存根都有一个SayHello()的RPC方法，该方法接受客户端的HelloRequest请求参数，并由服务端返回HelloReply结果。方法定义如下：

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

打开src/main/proto/helloworld.proto添加一个新方法SayHelloAgain(), 请求参数和返回结果同样使用原来的类型：

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
  // Sends another greeting
  rpc SayHelloAgain (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

更新app

当你生成样例的时候，构建过程会生成GreeterGrpc.java，该文件包含系统生成的gRpc客户端和服务端类文件。除此之外，还会生成其他的序列化，检索请求和返回类型的代码文件。
开发人员只需要关注方法实现并发起请求调用，而无需关注系统自动实现的代码框架。

这个和通过idl文件生成代码存根是一样的

更新服务端

在相同路径下，打开src/main/java/io/grpc/examples/helloworld/HelloWorldServer.java，实现新的方法：

private class GreeterImpl extends GreeterGrpc.GreeterImplBase {

  @Override
  public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
    HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
    responseObserver.onNext(reply);
    responseObserver.onCompleted();
  }

  @Override
  public void sayHelloAgain(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
    HelloReply reply = HelloReply.newBuilder().setMessage("Hello again " + req.getName()).build();
    responseObserver.onNext(reply);
    responseObserver.onCompleted();
  }
}

更新客户端

同样的，修改src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java文件：

public void greet(String name) {
  logger.info("Will try to greet " + name + " ...");
  HelloRequest request = HelloRequest.newBuilder().setName(name).build();
  HelloReply response;
  try {
    response = blockingStub.sayHello(request);
  } catch (StatusRuntimeException e) {
    logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
    return;
  }
  logger.info("Greeting: " + response.getMessage());
  try {
    response = blockingStub.sayHelloAgain(request);
  } catch (StatusRuntimeException e) {
    logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
    return;
  }
  logger.info("Greeting: " + response.getMessage());
}

运行修改后的app

像上文一样运行服务端和客户端，分别执行以下命令：
首先编译代码

$ ./gradlew installDist

接着运行服务端

$ ./build/install/examples/bin/hello-world-server

再运行客户端

$ ./build/install/examples/bin/hello-world-client
INFO: Will try to greet world ...
INFO: Greeting: Hello world
INFO: Greeting: Hello again world

gRpc介绍

通过本章节，你可以了解到如何定义接口文件，如何生成服务端和客户端系统代码，何如实现简单的分布式服务。

在阅读本章节之前，假设你已经熟悉了protocol buffers。

注意，本章节的示例中使用的protocol buffers版本是proto3。如果你想了解更多，请参考proto3 language guide.

为什么要使用gRPC

示例代码是一个简单的路由映射程序。

通过gRPC，我们可以定义.proto文件，然后生成gPRC支持的任何其他语言的客户端和服务端系统代码。这些代码既可以运行在大型服务器上，也可以运行在你自己的平板电脑上。
换句话说，gRPC帮你屏蔽了不同语言，不同环境之间的通讯细节，使得开发人员只需要关注到业务逻辑的实现上，提升了开发效率。

另外，我们也能充分利用protocol buffers的优点，包括高效的序列化效率，简单的IDL文件和便利地接口更新。

示例代码和设置

示例代码路径：/grpc-java/examples/src/main/java/io/grpc/examples/routeguide。

进入examples路径：

cd grpc-java/examples

服务定义

首先需要通过protocol buffers来声明接口、请求参数和返回参数的类型。具体参见：grpc-java/examples/src/main/proto/route_guide.proto。其中包含以下语句，用来指定生成的服务端和客户端系统代码的文件包名：

option java_package = "io.grpc.examples.routeguide";

如果不指定该选项，则默认使用package指定的包名。当然，对于非java语言，option java_package是无效的。

服务定义如下：

service RouteGuide {
    ...
}

接下来在service里面定义方法，指定请求和返回类型。

简单rpc

客户端发起请求，服务端处理请求并返回结果，就像正常的函数调用一样。

// Obtains the feature at a given position.
rpc GetFeature(Point) returns (Feature) {}

服务端流rpc

客户端发起请求并得到一个stream返回，循环从stream中读取有序数据直到没有更多的数据为止。

服务端流rpc需要在返回类型前面加上stream关键字。

// Obtains the Features available within the given Rectangle.  Results are
// streamed rather than returned at once (e.g. in a response message with a
// repeated field), as the rectangle may cover a large area and contain a
// huge number of features.
rpc ListFeatures(Rectangle) returns (stream Feature) {}

客户端流rpc

客户端以流式方式将数据发送到服务端，当全部发送完成后，等待服务端的返回。服务端方面持续读取请求数据，读取完所有数据后进行处理并一次性返回结果。

客户端流rpc需要在请求类型前面加上stream关键字。

// Accepts a stream of Points on a route being traversed, returning a
// RouteSummary when traversal is completed.
rpc RecordRoute(stream Point) returns (RouteSummary) {}

双向流(bidirecectional)rpc

客户端请求流式发送数据，服务端返回也是流式返回数据。客户端的请求流和服务端返回流是独立的，这意味着客户端和服务端可以以它们想要的方式来读或者写数据。比如服务端可以等待所有数据接收完成后再返回数据，也可以接受完一部分数据之后就立即返回，再继续接收数据并返回。甚至能以其他的组合方式来读写数据。

双向流rpc需要在请求和返回参数前面都加上stream关键字。

// Accepts a stream of RouteNotes sent while a route is being traversed,
// while receiving other RouteNotes (e.g. from other users).
rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

当然，接口涉及的请求类型和返回类型也会在proto文件中定义，比如Point 的定义：

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

生成客户端和服务端存根代码

定义好了接口后就能通过protocol buffer编译器(protoc)来生成客户端和服务端的系统代码了(当然，不同的语言还需要各自不同的插件)。

如果在项目中使用Gradle或者Maven，可以将protoc编译器放到build过程中，具体参考：grpc-java README。

生成的系统代码包括：Feature.java, Point.java, Rectangle.java为输入输出的数据类型；RouteGuideGrpc.java为服务端功能实现类；客户端调用类。如下所示：

创建server

首先让我们看一看如何创建RouteGuide服务端：

• 覆写服务端功能实现基类，也就是上面提到的RouteGuideGrpc类，该类是由系统生成的。
• 运行服务端，监听客户端请求并返回。

服务端实现类：grpc-java/examples/src/main/java/io/grpc/examples/routeguide/RouteGuideServer.java。

RouteGuide功能实现

如代码所示，服务端实现类RouteGuideService继承了系统生成的基类RouteGuideGrpc.RouteGuideImplBase:

private static class RouteGuideService extends RouteGuideGrpc.RouteGuideImplBase {
 ...
}

简单rpc

RouteGuideService实现了服务定义中的所有方法。

简单rpc接口如GetFeature()，该方法请求参数为位置Point，返回该位置对应的Feature。如下所示：

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  responseObserver.onNext(checkFeature(request));
  responseObserver.onCompleted();
}

...

private Feature checkFeature(Point location) {
  for (Feature feature : features) {
    if (feature.getLocation().getLatitude() == location.getLatitude()
        && feature.getLocation().getLongitude() == location.getLongitude()) {
      return feature;
    }
  }

  // No feature was found, return an unnamed feature.
  return Feature.newBuilder().setName("").setLocation(location).build();
}

请求参数为两个： 位置Point和响应观察器StreamObserver。通过响应观察期的onNext()来返回Feature，然后通过onCompleted() 来表明本次rpc调用已完成。

服务端流rpc

ListFeatures是一个服务端流rpc接口，返回多个Feature。代码如下所示：

private final Collection<Feature> features;

...
@Override
public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
  int left = min(request.getLo().getLongitude(), request.getHi().getLongitude());
  int right = max(request.getLo().getLongitude(), request.getHi().getLongitude());
  int top = max(request.getLo().getLatitude(), request.getHi().getLatitude());
  int bottom = min(request.getLo().getLatitude(), request.getHi().getLatitude());

  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) {
      responseObserver.onNext(feature);
    }
  }
  responseObserver.onCompleted();
}

在for循环中获取符合条件的Feature返回。循环结束后通过onCompleted()表明本次rpc调用结束。

客户端流rpc

下面看一个更复杂一点的客户端流rpc接口：RecordRoute()，请求数据是流式的，返回一个RouteSunmmary。

@Override
public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {
  return new StreamObserver<Point>() {
    int pointCount;
    int featureCount;
    int distance;
    Point previous;
    long startTime = System.nanoTime();

    @Override
    public void onNext(Point point) {
      pointCount++;
      if (RouteGuideUtil.exists(checkFeature(point))) {
        featureCount++;
      }
      // For each point after the first, add the incremental distance from the previous point
      // to the total distance value.
      if (previous != null) {
        distance += calcDistance(previous, point);
      }
      previous = point;
    }

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

    @Override
    public void onCompleted() {
      long seconds = NANOSECONDS.toSeconds(System.nanoTime() - startTime);
      responseObserver.onNext(RouteSummary.newBuilder().setPointCount(pointCount)
          .setFeatureCount(featureCount).setDistance(distance)
          .setElapsedTime((int) seconds).build());
      responseObserver.onCompleted();
    }
  };
}

在服务端流rpc中，返回结果是流式的，而客户端流rpc中，请求数据是流式的。在该方法中，实例化了一个匿名类对象来返回，并在该匿名类中覆写了几个方法：

• 覆写onNext()，该方法用于接收客户端请求并进行处理。
• 覆写onCompleted()，当服务端接收完客户端的所有请求数据之后，该方法会被调用，构建一个RouteSummary对象。然后调用响应监听器的onNext()和onCompleted()来返回数据和结束rpc调用。

双向rpc

最后是双向rpc接口：routeChat()。

@Override
public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {
     return new StreamObserver<RouteNote>() {
       @Override
       public void onNext(RouteNote note) {
         List<RouteNote> notes = getOrCreateNotes(note.getLocation());

         // Respond with all previous notes at this location.
         for (RouteNote prevNote : notes.toArray(new RouteNote[0])) {
           responseObserver.onNext(prevNote);
         }

         // Now add the new note to the list
         notes.add(note);
       }

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

       @Override
       public void onCompleted() {
         responseObserver.onCompleted();
       }
     };
   }

就像客户端流rpc一样，双向流rpc需要两个StreamObserver对象，一个用于从客户端读数据，一个用于向客户端写数据。区别在于：双向流rpc的读和写是同时存在的。

启动server

启动Server的代码如下：

public RouteGuideServer(int port, URL featureFile) throws IOException {
      this(ServerBuilder.forPort(port), port, RouteGuideUtil.parseFeatures(featureFile));
}

// Create a RouteGuide server using serverBuilder as a base and features as data.
public RouteGuideServer(ServerBuilder<?> serverBuilder, int port, Collection<Feature> features) {
      this.port = port;
      server = serverBuilder.addService(new RouteGuideService(features)).build();
}
...
public void start() throws IOException {
      server.start();
      logger.info("Server started, listening on " + port);
...
}

如代码所示，我们通过ServerBuilder来构建并启动服务端：

• 指定服务端地址和端口，通过builder的forPort()来监听客户端请求。
• 创建一个服务实现类的实例RouteGuideService，并将其作为参数传递给builder的addService()方法中。
• 调用build()和start()方法来创建和启动rpc服务端。

创建client

在这一部分，我们将看看如何创建RouteGuide的客户端。完整的代码路径：
grpc-java/examples/src/main/java/io/grpc/examples/routeguide/RouteGuideClient.java.

实例化stub

为了调用服务方法，首先需要创建一个stub或者更确切的说，两个stub。一个是阻塞/同步的stub，一个是非阻塞/异步的stub。

public RouteGuideClient(String host, int port) {
  this(ManagedChannelBuilder.forAddress(host, port).usePlaintext());
}

/** Construct client for accessing RouteGuide server using the existing channel. */
public RouteGuideClient(ManagedChannelBuilder<?> channelBuilder) {
  channel = channelBuilder.build();
  blockingStub = RouteGuideGrpc.newBlockingStub(channel); // 同步存根
  asyncStub = RouteGuideGrpc.newStub(channel);  // 异步存根
}

调用服务方法

简单rpc

在阻塞stub上调用简单的rpc GetFeature与调用本地方法一样简单。

Point request = Point.newBuilder().setLatitude(lat).setLongitude(lon).build();
Feature feature;
try {
  feature = blockingStub.getFeature(request);
} catch (StatusRuntimeException e) {
  logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
  return;
}

我们创建并输入一个请求协议缓冲对象(例子中的Point)，将其作为参数传给getFeature()方法。如果调用过程中出现错误，将抛出StatusRuntimeException异常。

服务流rpc

下面是服务端流rpc ListFeatures的客户端调用：

Rectangle request =
    Rectangle.newBuilder()
        .setLo(Point.newBuilder().setLatitude(lowLat).setLongitude(lowLon).build())
        .setHi(Point.newBuilder().setLatitude(hiLat).setLongitude(hiLon).build()).build();
Iterator<Feature> features;
try {
  features = blockingStub.listFeatures(request);
} catch (StatusRuntimeException e) {
  logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
  return;
}

客户端的调用实际上与简单rpc差不多，唯一的差别在于服务端流rpc的返回结果是持续不断的，因此需要通过迭代器来接收，然后再从迭代器中取出所有的返回结果。

客户端流rpc

在客户端流rpc接口RecordRoute()中，我们流式发送Point到服务端，并得到一个RouteSummary结果。在这个方法中，我们需要使用异步stub。如下所示：

public void recordRoute(List<Feature> features, int numPoints) throws InterruptedException {
  info("*** RecordRoute");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteSummary> responseObserver = new StreamObserver<RouteSummary>() {
    @Override
    public void onNext(RouteSummary summary) {
      info("Finished trip with {0} points. Passed {1} features. "
          + "Travelled {2} meters. It took {3} seconds.", summary.getPointCount(),
          summary.getFeatureCount(), summary.getDistance(), summary.getElapsedTime());
    }

    @Override
    public void onError(Throwable t) {
      Status status = Status.fromThrowable(t);
      logger.log(Level.WARNING, "RecordRoute Failed: {0}", status);
      finishLatch.countDown();
    }

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

  StreamObserver<Point> requestObserver = asyncStub.recordRoute(responseObserver);
  try {
    // Send numPoints points randomly selected from the features list.
    Random rand = new Random();
    for (int i = 0; i < numPoints; ++i) {
      int index = rand.nextInt(features.size());
      Point point = features.get(index).getLocation();
      info("Visiting point {0}, {1}", RouteGuideUtil.getLatitude(point),
          RouteGuideUtil.getLongitude(point));
      requestObserver.onNext(point);
      // Sleep for a bit before sending the next one.
      Thread.sleep(rand.nextInt(1000) + 500);
      if (finishLatch.getCount() == 0) {
        // RPC completed or errored before we finished sending.
        // Sending further requests won't error, but they will just be thrown away.
        return;
      }
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

为了调用这一方法，我们需要创建一个StreamObserver，并覆写一些方法：

• onNext()：当服务端返回RouteSummary时，该方法负责将其输出。
• onCompleted()：当服务端返回数据时，该方法被调用，判断服务端数据是否写完。

从上文可以看出，客户端流方法调用需要实现两个StreamObserver，一个用于发送客户端请求，一个用于接收服务端的返回。

双向流rpc

双向流rpc接口RouteChat()客户端调用如下：

public void routeChat() throws Exception {
  info("*** RoutChat");
  final CountDownLatch finishLatch = new CountDownLatch(1);
  StreamObserver<RouteNote> requestObserver =
      asyncStub.routeChat(new StreamObserver<RouteNote>() {
        @Override
        public void onNext(RouteNote note) {
          info("Got message \"{0}\" at {1}, {2}", note.getMessage(), note.getLocation()
              .getLatitude(), note.getLocation().getLongitude());
        }

        @Override
        public void onError(Throwable t) {
          Status status = Status.fromThrowable(t);
          logger.log(Level.WARNING, "RouteChat Failed: {0}", status);
          finishLatch.countDown();
        }

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

  try {
    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) {
      info("Sending message \"{0}\" at {1}, {2}", request.getMessage(), request.getLocation()
          .getLatitude(), request.getLocation().getLongitude());
      requestObserver.onNext(request);
    }
  } catch (RuntimeException e) {
    // Cancel RPC
    requestObserver.onError(e);
    throw e;
  }
  // Mark the end of requests
  requestObserver.onCompleted();

  // Receiving happens asynchronously
  finishLatch.await(1, TimeUnit.MINUTES);
}

与客户端流rpc类似，双向流rpc也有两个StreamObserver，一个用于发送请求，一个用于接收响应。区别在于：双向流rpc的读和写是同时存在的。