基本チュートリアル

このチュートリアルでは、gRPC を扱うための基本的な Kotlin プログラマ向け入門を提供します。

この例を段階的に実行することで、以下のことを学びます。

• .protoファイルでサービスを定義する。
• プロトコルバッファコンパイラを使用して、サーバーとクライアントのコードを生成する。
• Kotlin gRPC API を使用して、サービス用のシンプルなクライアントとサーバーを記述します。

gRPC と protocol buffers にはすでに慣れているはずです。そうでない場合は、gRPC の概要と proto3 の言語ガイドを参照してください。

gRPCを使用する理由

私たちの例は、クライアントがルート上の機能に関する情報を取得したり、ルートの要約を作成したり、サーバーや他のクライアントと交通情報などのルート情報を交換したりできる、シンプルなルートマッピングアプリケーションです。

gRPCを使用すると、1つの.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 の概要で既に知っていると思いますが) 最初のステップは、protocol buffersを使用して 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ファイルには、サービスメソッドで使用されるすべてリクエストおよびレスポンスタイプの protocol buffer メッセージタイプ定義も含まれています。たとえば、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 プラグインとともに protocol buffer コンパイラprotocを使用して行います。

Gradle または Maven を使用する場合、protocビルドプラグインはビルドプロセスの一部として必要なコードを生成します。Gradle の例については、stub/build.gradle.ktsを参照してください。

examples フォルダーから./gradlew installDistを実行すると、サービス定義から次のファイルが生成されます。生成されたファイルは、stub/build/generated/source/proto/main以下のサブディレクトリにあります。

• Feature.java、Point.java、Rectangle.javaなど。これらには、リクエストおよびレスポンスメッセージタイプをポピュレート、シリアライズ、および取得するためのすべての protocol buffer コードが含まれています。

  これらのファイルは、java/io/grpc/examples/routeguideサブディレクトリにあります。

• RouteGuideOuterClassGrpcKt.kt。これには、とりわけ以下が含まれています。

  • RouteGuideCoroutineImplBase。これは、RouteGuideサーバーが実装するための抽象基底クラスで、RouteGuideサービスで定義されたすべてのメソッドが含まれています。
  • RouteGuideCoroutineStub。これは、クライアントがRouteGuideサーバーと通信するために使用するクラスです。

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

サーバーを作成する

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

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

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

example RouteGuideサーバーコードを、server/src/main/kotlin/io/grpc/examplesフォルダーの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 の 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

最後に、双方向ストリーミング RPCRouteChat()を検討します。

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
    }
  }

クライアントサイドストリーミング例と同様に、このメソッドでは、サーバーはFlowとしてRouteNoteオブジェクトのストリームを受け取ります。ただし、今回は、クライアントが自分のメッセージストリームにメッセージを書き込んでいる間に、メソッドの返されたストリームを介して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

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

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>インスタンスの形式でフィーチャーのストリームを返します。フローの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))
  }
}

フローポイントは遅延して発行されることに注意してください。つまり、サーバーが要求したときにのみ発行されます。ポイントがフローに発行されると、ポイントジェネレーターはサーバーが次のポイントを要求するまでサスペンドします。

双方向ストリーミング RPC

最後に、双方向ストリーミング RPCRouteChat()を検討します。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