基本チュートリアル

このチュートリアルでは、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 では、4 種類のサービスメソッドを定義できます。これらはすべて RouteGuide サービスで使用されています。

• クライアントがスタブを使用してサーバーにリクエストを送信し、通常の関数呼び出しのようにレスポンスが返されるのを待つ、*シンプルな RPC*。

  // 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*。2 つのストリームは独立して動作するため、クライアントとサーバーは好きな順序で読み書きできます。たとえば、サーバーはクライアントのすべてのメッセージを受信してからレスポンスを書き込むことも、メッセージを読み取ってからメッセージを書き込むことも、読み取りと書き込みのその他の組み合わせを行うこともできます。各ストリーム内のメッセージの順序は保持されます。リクエストとレスポンスの両方の前に 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 クライアントとサーバーのインターフェースを生成する必要があります。これには、特別な gRPC Kotlin と Java プラグインを使用してプロトコルバッファコンパイラ protoc を使用します。

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 サーバーと通信するために使用するクラス。

  この Kotlin ファイルは、grpckt/io/grpc/examples/routeguide にあります。

サーバーの作成

まず、RouteGuide サーバーの作成方法を検討します。gRPC クライアントの作成のみに関心がある場合は、クライアントの作成 に進んでください。ただし、このセクションも興味深いと感じるかもしれません！

RouteGuide サーバーを作成する際には、主に 2 つのことを行う必要があります。

• RouteGuideCoroutineImplBase サービス基本クラスを拡張して、実際のサービス作業を実行します。
• クライアントからのリクエストをリッスンし、サービスレスポンスを返す gRPC サーバーを作成して実行します。

server/src/main/kotlin/io/grpc/examples フォルダーにある routeguide/RouteGuideServer.kt で、サンプルの RouteGuide サーバーコードを開きます。

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 フレームワークに返し、フレームワークがクライアントに返します。

サーバー側ストリーミング RPC

次に、ストリーミング RPC の 1 つを考えてみましょう。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 のインスタンスを作成し、ビルダーの addService() メソッドに渡します。
3. ビルダーで build() と start() を呼び出して、ルートガイドサービスの RPC サーバーを作成および開始します。
4. アプリケーションが終了シグナルを受信するまでメイン関数をブロックするには、サーバーで awaitTermination() を呼び出します。

クライアントの作成

このセクションでは、RouteGuide サービスのクライアントについて説明します。

完全なクライアントコードについては、client/src/main/kotlin/io/grpc/examples フォルダーにある routeguide/RouteGuideClient.kt を開きます。

スタブのインスタンス化

サービスメソッドを呼び出すには、まず ManagedChannelBuilder を使用して gRPC *チャネル* を作成する必要があります。このチャネルを使用してサーバーと通信します。

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

gRPC チャネルが設定されたら、RPC を実行するためのクライアント *スタブ* が必要です。.proto ファイルから生成されたパッケージから入手できる RouteGuideCoroutineStub をインスタンス化して取得します。

val stub = RouteGuideCoroutineStub(channel)

サービスメソッドの呼び出し

次に、サービスメソッドの呼び出し方法について考えます。

シンプルな RPC

単純なRPCであるGetFeature()の呼び出しは、ローカルメソッドを呼び出すのと同様に簡単です。

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

スタブメソッド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")
  }
}

スタブ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()の場合と同様に、要求メッセージを書き込むために使用するストリームをスタブメソッドに渡します。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