基础教程

本教程为 Kotlin 程序员提供了使用 gRPC 的基本介绍。

通过学习本示例，您将了解如何

• 在 .proto 文件中定义服务。
• 使用协议缓冲区编译器生成服务器和客户端代码。
• 使用 Kotlin gRPC API 为您的服务编写一个简单的客户端和服务器。

您应该已经熟悉 gRPC 和协议缓冲区；如果不熟悉，请参阅gRPC 简介和 proto3 语言指南。

为什么要使用 gRPC？

我们的示例是一个简单的路线映射应用程序，它允许客户端获取其路线上的特征信息、创建路线摘要，并与服务器及其他客户端交换路线信息（例如交通更新）。

使用 gRPC，我们可以在一个 .proto 文件中定义一次服务，然后使用 gRPC 支持的任何语言生成客户端和服务器，这些客户端和服务器又可以在从大型数据中心内部的服务器到您自己的平板电脑等各种环境中运行 — 不同语言和环境之间的通信复杂性都由 gRPC 为您处理。我们还获得了使用协议缓冲区的全部优势，包括高效的序列化、简单的 IDL 和易于接口更新。

设置

本教程具有与快速入门相同的先决条件。在继续之前，请安装必要的 SDK 和工具。

获取示例代码

示例代码是 grpc-kotlin 仓库的一部分。

1. 将仓库下载为 zip 文件并解压缩，或克隆仓库

   git clone --depth 1 https://github.com/grpc/grpc-kotlin
   

2. 切换到 examples 目录

   cd grpc-kotlin/examples
   

定义服务

您的第一步（如您从gRPC 简介中了解到的）是使用协议缓冲区定义 gRPC 服务以及方法 请求和 响应类型。

如果您想通过查看完整的 .proto 文件来学习，请参见 protos/src/main/proto/io/grpc/examples 文件夹中的 routeguide/route_guide.proto 文件。

要定义服务，您可以在 .proto 文件中指定一个命名的 service，如下所示

service RouteGuide {
   ...
}

然后您在服务定义内部定义 rpc 方法，并指定它们的请求和响应类型。gRPC 允许您定义四种服务方法，所有这些方法都用于 RouteGuide 服务中

• 简单 RPC，客户端使用 stub 向服务器发送请求并等待响应返回，就像普通的函数调用一样。

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

• 服务器端流式 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，客户端写入一系列消息并使用提供的流将它们发送到服务器。客户端完成写入消息后，会等待服务器读取所有消息并返回其响应。您可以通过在 请求 类型之前放置 stream 关键字来指定客户端流式方法。

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

• 双向流式 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 消息类型

// Points are represented as latitude-longitude pairs in the E7 representation
// (degrees multiplied by 10**7 and rounded to the nearest integer).
// Latitudes should be in the range +/- 90 degrees and longitude should be in
// the range +/- 180 degrees (inclusive).
message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

生成客户端和服务器代码

接下来，您需要从 .proto 服务定义生成 gRPC 客户端和服务器接口。您可以使用协议缓冲区编译器 protoc 以及特殊的 gRPC Kotlin 和 Java 插件来完成此操作。

使用 Gradle 或 Maven 时，protoc 构建插件会在构建过程中生成必要的代码。有关 Gradle 示例，请参见 stub/build.gradle.kts。

如果您从 examples 文件夹运行 ./gradlew installDist，将从服务定义生成以下文件 — 您可以在 stub/build/generated/source/proto/main 下的子目录中找到生成的文件

• Feature.java、Point.java、Rectangle.java 等文件，其中包含所有用于填充、序列化和检索我们的请求和响应消息类型的协议缓冲区代码。

  您可以在 java/io/grpc/examples/routeguide 子目录中找到这些文件。

• RouteGuideOuterClassGrpcKt.kt，其中包含（但不限于）以下内容

  • RouteGuideGrpcKt.RouteGuideCoroutineImplBase，这是一个供 RouteGuide 服务器实现的抽象基类，包含在 RouteGuide 服务中定义的所有方法。
  • RouteGuideGrpcKt.RouteGuideCoroutineStub，这是一个客户端用于与 RouteGuide 服务器通信的类。

  您可以在 grpckt/io/grpc/examples/routeguide 下找到此 Kotlin 文件。

创建服务器

首先考虑如何创建 RouteGuide 服务器。如果您只对创建 gRPC 客户端感兴趣，请跳至创建客户端 — 尽管您可能仍然觉得本节有趣！

创建 RouteGuide 服务器时需要做两件主要的事情

• 扩展 RouteGuideCoroutineImplBase 服务基类以执行实际的服务工作。
• 创建并运行 gRPC 服务器以监听来自客户端的请求并返回服务响应。

打开 server/src/main/kotlin/io/grpc/examples 文件夹下的示例 RouteGuide 服务器代码 routeguide/RouteGuideServer.kt。

实现 RouteGuide

如您所见，服务器有一个扩展生成的服务基类的 RouteGuideService 类

class RouteGuideService(
  val features: Collection<Feature>,
  /* ... */
) : RouteGuideGrpcKt.RouteGuideCoroutineImplBase() {
  /* ... */
}

简单 RPC

RouteGuideService 实现了所有服务方法。首先考虑最简单的方法 GetFeature()，它从客户端获取一个 Point 并返回一个根据数据库中相应的特征信息构建的 Feature。

override suspend fun getFeature(request: Point): Feature =
    features.find { it.location == request } ?:
    // No feature was found, return an unnamed feature.
    Feature.newBuilder().apply { location = request }.build()

该方法接受客户端的 Point 消息请求作为参数，并返回一个 Feature 消息作为响应。该方法使用适当的信息填充 Feature，然后将其返回给 gRPC 框架，由 gRPC 框架将其发送回客户端。

服务器端流式 RPC

接下来，考虑其中一个流式 RPC。ListFeatures() 是一个服务器端流式 RPC，因此服务器可以向客户端发回多个 Feature 消息。

override fun listFeatures(request: Rectangle): Flow<Feature> =
  features.asFlow().filter { it.exists() && it.location in request }

请求对象是 Rectangle。服务器收集并返回给客户端，其集合中位于给定 Rectangle 内的所有 Feature 对象。

客户端流式 RPC

现在考虑一些更复杂的内容：客户端流式方法 RecordRoute()，其中服务器从客户端获取 Point 对象流，并返回一个包含客户端通过给定点的行程信息的单个 RouteSummary。

override suspend fun recordRoute(requests: Flow<Point>): RouteSummary {
  var pointCount = 0
  var featureCount = 0
  var distance = 0
  var previous: Point? = null
  val stopwatch = Stopwatch.createStarted(ticker)
  requests.collect { request ->
    pointCount++
    if (getFeature(request).exists()) {
      featureCount++
    }
    val prev = previous
    if (prev != null) {
      distance += prev distanceTo request
    }
    previous = request
  }
  return RouteSummary.newBuilder().apply {
    this.pointCount = pointCount
    this.featureCount = featureCount
    this.distance = distance
    this.elapsedTime = Durations.fromMicros(stopwatch.elapsed(TimeUnit.MICROSECONDS))
  }.build()
}

请求参数是以 Kotlin Flow 表示的客户端请求消息流。服务器返回单个响应，就像简单 RPC 的情况一样。

双向流式 RPC

最后，考虑双向流式 RPC RouteChat()。

override fun routeChat(requests: Flow<RouteNote>): Flow<RouteNote> =
  flow {
    // could use transform, but it's currently experimental
    requests.collect { note ->
      val notes: MutableList<RouteNote> = routeNotes.computeIfAbsent(note.location) {
        Collections.synchronizedList(mutableListOf<RouteNote>())
      }
      for (prevNote in notes.toTypedArray()) { // thread-safe snapshot
        emit(prevNote)
      }
      notes += note
    }
  }

类似于客户端流式示例，对于此方法，服务器将 RouteNote 对象流作为 Flow 获取。然而，这次服务器通过该方法返回的流返回 RouteNote 实例，同时客户端仍在向 其 消息流写入消息。

启动服务器

实现所有服务器方法后，您需要编写代码来创建 gRPC 服务器实例，如下所示

class RouteGuideServer(
    val port: Int,
    val features: Collection<Feature> = Database.features(),
    val server: Server =
      ServerBuilder.forPort(port)
        .addService(RouteGuideService(features)).build()
) {

  fun start() {
    server.start()
    println("Server started, listening on $port")
    /* ... */
  }
  /* ... */
}

fun main(args: Array<String>) {
  val port = 8980
  val server = RouteGuideServer(port)
  server.start()
  server.awaitTermination()
 }

服务器实例使用 ServerBuilder 构建和启动，如下所示

1. 使用 forPort() 指定服务器将监听客户端请求的端口。
2. 创建一个服务实现类 RouteGuideService 的实例，并将其传递给 builder 的 addService() 方法。
3. 在 builder 上调用 build() 和 start() 以创建并启动路由指南服务的 RPC 服务器。
4. 在服务器上调用 awaitTermination() 以阻塞主函数，直到应用程序收到终止信号。

创建客户端

在本节中，您将了解 RouteGuide 服务的客户端。

有关完整的客户端代码，请打开 client/src/main/kotlin/io/grpc/examples 文件夹下的 routeguide/RouteGuideClient.kt。

实例化 Stub

要调用服务方法，首先需要使用 ManagedChannelBuilder 创建一个 gRPC 通道。您将使用此通道与服务器通信。

val channel = ManagedChannelBuilder.forAddress("localhost", 8980).usePlaintext().build()

设置好 gRPC 通道后，您需要一个客户端 stub 来执行 RPC。通过实例化 RouteGuideCoroutineStub 获取它，该类可从 .proto 文件生成的包中获得。

val stub = RouteGuideCoroutineStub(channel)

调用服务方法

现在考虑如何调用服务方法。

简单 RPC

调用简单 RPC GetFeature() 就像调用本地方法一样简单明了

val request = point(latitude, longitude)
val feature = stub.getFeature(request)

stub 方法 getFeature() 执行相应的 RPC，并挂起直到 RPC 完成

suspend fun getFeature(latitude: Int, longitude: Int) {
  val request = point(latitude, longitude)
  val feature = stub.getFeature(request)
  if (feature.exists()) { /* ... */ }
}

服务器端流式 RPC

接下来，考虑服务器端流式 ListFeatures() RPC，它返回地理特征流

suspend fun listFeatures(lowLat: Int, lowLon: Int, hiLat: Int, hiLon: Int) {
  val request = Rectangle.newBuilder()
    .setLo(point(lowLat, lowLon))
    .setHi(point(hiLat, hiLon))
    .build()
  var i = 1
  stub.listFeatures(request).collect { feature ->
    println("Result #${i++}: $feature")
  }
}

stub 的 listFeatures() 方法以 Flow<Feature> 实例的形式返回特征流。flow 的 collect() 方法允许客户端处理服务器提供的可用特征。

客户端流式 RPC

客户端流式 RecordRoute() RPC 将 Point 消息流发送到服务器，并获得单个 RouteSummary 作为响应。

suspend fun recordRoute(points: Flow<Point>) {
  println("*** RecordRoute")
  val summary = stub.recordRoute(points)
  println("Finished trip with ${summary.pointCount} points.")
  println("Passed ${summary.featureCount} features.")
  println("Travelled ${summary.distance} meters.")
  val duration = summary.elapsedTime.seconds
  println("It took $duration seconds.")
}

该方法从与随机选择的特征列表相关的点生成路线点。随机选择取自预先加载的特征集合

fun generateRoutePoints(features: List<Feature>, numPoints: Int): Flow<Point> = flow {
  for (i in 1..numPoints) {
    val feature = features.random(random)
    println("Visiting point ${feature.location.toStr()}")
    emit(feature.location)
    delay(timeMillis = random.nextLong(500L..1500L))
  }
}

请注意，flow 点是延迟发出的，也就是说，只有当服务器请求它们时才会发出。一旦一个点已发送到 flow，点生成器就会挂起，直到服务器请求下一个点。

双向流式 RPC

最后，考虑双向流式 RPC RouteChat()。与 RecordRoute() 的情况一样，您向 stub 方法传递一个用于写入请求消息的流；就像在 ListFeatures() 中一样，您会获得一个可用于读取响应消息的流。但是，这次您通过我们方法的流发送值，同时服务器也在向 其 消息流写入消息。

suspend fun routeChat() {
  val requests = generateOutgoingNotes()
  stub.routeChat(requests).collect { note ->
    println("Got message \"${note.message}\" at ${note.location.toStr()}")
  }
  println("Finished RouteChat")
}

private fun generateOutgoingNotes(): Flow<RouteNote> = flow {
  val notes = listOf(/* ... */)
  for (note in notes) {
    println("Sending message \"${note.message}\" at ${note.location.toStr()}")
    emit(note)
    delay(500)
  }
}

这里的读写语法与客户端和服务器端流式方法非常相似。尽管每一方总是会按照消息写入的顺序接收对方的消息，但客户端和服务器都可以按任何顺序读写 — 流完全独立运行。

试一试！

从 grpc-kotlin/examples 目录运行以下命令

1. 编译客户端和服务器

   ./gradlew installDist
   

2. 运行服务器

   ./server/build/install/server/bin/route-guide-server
   Server started, listening on 8980
   

3. 从另一个终端运行客户端

   ./client/build/install/client/bin/route-guide-client
   

您将看到如下客户端输出

*** GetFeature: lat=409146138 lon=-746188906
Found feature called "Berkshire Valley Management Area Trail, Jefferson, NJ, USA" at 40.9146138, -74.6188906
*** GetFeature: lat=0 lon=0
Found no feature at 0.0, 0.0
*** ListFeatures: lowLat=400000000 lowLon=-750000000 hiLat=420000000 liLon=-730000000
Result #1: name: "Patriots Path, Mendham, NJ 07945, USA"
location {
  latitude: 407838351
  longitude: -746143763
}
...
Result #64: name: "3 Hasta Way, Newton, NJ 07860, USA"
location {
  latitude: 410248224
  longitude: -747127767
}

*** RecordRoute
Visiting point 40.0066188, -74.6793294
...
Visiting point 40.4318328, -74.0835638
Finished trip with 10 points.
Passed 3 features.
Travelled 93238790 meters.
It took 9 seconds.
*** RouteChat
Sending message "First message" at 0.0, 0.0
Sending message "Second message" at 0.0, 0.0
Got message "First message" at 0.0, 0.0
Sending message "Third message" at 1.0, 0.0
Sending message "Fourth message" at 1.0, 1.0
Sending message "Last message" at 0.0, 0.0
Got message "First message" at 0.0, 0.0
Got message "Second message" at 0.0, 0.0
Finished RouteChat