このページは Cloud Translation API によって翻訳されました。

gRPC-Java スタートガイド - ストリーミング

1. はじめに

この Codelab では、gRPC-Java を使用して、Java で記述されたルートマッピングアプリケーションの基盤となるクライアントとサーバーを作成します。

このチュートリアルを完了すると、gRPC を使用してリモートサーバーに接続し、クライアントのルート上の機能に関する情報を取得し、クライアントのルートの概要を作成し、サーバーや他のクライアントとトラフィックの更新などのルート情報を交換するクライアントが作成されます。

サービスは Protocol Buffers ファイルで定義されます。このファイルは、クライアントとサーバーが相互に通信できるように、クライアントとサーバーのボイラープレートコードを生成するために使用されます。これにより、この機能を実装する時間と労力を節約できます。

この生成されたコードは、サーバーとクライアント間の通信の複雑さだけでなく、データのシリアル化と逆シリアル化も処理します。

学習内容

プロトコルバッファを使用してサービス API を定義する方法。
自動コード生成を使用して、プロトコルバッファ定義から gRPC ベースのクライアントとサーバーを構築する方法。
gRPC を使用したクライアントサーバーストリーミング通信の理解。

この Codelab は、gRPC を初めて使用する Java デベロッパー、gRPC の復習を希望するデベロッパー、分散システムの構築に関心のある方を対象としています。gRPC の経験は必要ありません。

2. 始める前に

前提条件

JDK バージョン 24。

コードを取得する

この Codelab では、完全にゼロから始める必要がないように、アプリケーションのソースコードのスケルトンが用意されています。次の手順では、プロトコルバッファコンパイラプラグインを使用してボイラープレート gRPC コードを生成するなど、アプリケーションを完成させる方法について説明します。

まず、Codelab の作業ディレクトリを作成し、そのディレクトリに移動します。

mkdir streaming-grpc-java-getting-started && cd streaming-grpc-java-getting-started

Codelab をダウンロードして解凍します。

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-1/codelabs/grpc-java-streaming/start_here

または、Codelab ディレクトリのみを含む .zip ファイルをダウンロードして、手動で解凍することもできます。

実装の入力をスキップする場合は、完成したソースコードを GitHub で入手できます。

3. メッセージとサービスを定義する

まず、プロトコルバッファを使用して、アプリケーションの gRPC サービス、RPC メソッド、リクエストとレスポンスのメッセージタイプを定義します。サービスでは以下を提供します。

サーバーが実装し、クライアントが呼び出す RPC メソッド ListFeatures、RecordRoute、RouteChat。
上記のメソッドを呼び出すときにクライアントとサーバーの間で交換されるデータ構造であるメッセージタイプ Point、Feature、Rectangle、RouteNote、RouteSummary。

Protocol Buffers は一般に protobuf と呼ばれます。gRPC の用語の詳細については、gRPC のコアコンセプト、アーキテクチャ、ライフサイクルをご覧ください。

この RPC メソッドとそのメッセージ型はすべて、提供されたソースコードの proto/routeguide/route_guide.proto ファイルで定義されます。

route_guide.proto ファイルを作成しましょう。

この例では Java コードを生成するため、.proto で java_package ファイルオプションを指定しています。

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

メッセージタイプを定義する

ソースコードの proto/routeguide/route_guide.proto ファイルで、まず Point メッセージタイプを定義します。Point は、地図上の緯度と経度の座標ペアを表します。この Codelab では、座標に整数を使用します。

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

数値 1 と 2 は、message 構造内の各フィールドの一意の ID 番号です。

次に、Feature メッセージタイプを定義します。Feature は、Point で指定された場所にあるものの名前または郵便番号に string フィールドを使用します。

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

領域内の複数のポイントをクライアントにストリーミングできるようにするには、緯度経度の長方形を表す Rectangle メッセージが必要です。これは、対角線上の 2 つのポイント lo と hi で表されます。

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

また、特定の時点で送信されたメッセージを表す RouteNote メッセージは次のようになります。

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

最後に、RouteSummary メッセージが必要です。このメッセージは、次のセクションで説明する RecordRoute RPC のレスポンスとして受信されます。受信した個々のポイントの数、検出された特徴の数、各ポイント間の距離の累積合計としてカバーされた合計距離が含まれます。

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

サービスメソッドを定義する

サービスを定義するには、.proto ファイルで名前付きサービスを指定します。route_guide.proto ファイルには、アプリケーションのサービスによって提供される 1 つ以上のメソッドを定義する RouteGuide という名前の service 構造があります。

サービス定義内で RPC メソッドを定義するときは、リクエストとレスポンスの型を指定します。この Codelab のこのセクションでは、次のことを定義します。

ListFeatures

指定された Rectangle 内で使用可能な Feature オブジェクトを取得します。長方形が広い範囲をカバーし、多数の対象物を含む可能性があるため、結果は一度に返されるのではなく、ストリーミングされます。

このアプリケーションでは、サーバーサイド ストリーミング RPC を使用します。クライアントがサーバーにリクエストを送信し、一連のメッセージを読み取るストリームを取得します。クライアントは、受信したストリームをメッセージがなくなるまで読み取ります。例に示すように、レスポンス型の前に stream キーワードを配置して、サーバーサイドストリーミングメソッドを指定します。

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

通過中のルート上のポイントのストリームを受け取り、通過が完了すると RouteSummary を返します。

この場合は、クライアントサイドストリーミング RPC が適しています。クライアントは、提供されたストリームを使用して、一連のメッセージを書き込み、サーバーに送信します。クライアントは、メッセージの書き込みを終了すると、サーバーがすべてのメッセージを読み取ってレスポンスを返すのを待ちます。クライアントサイドストリーミングメソッドを指定するには、リクエストタイプの前に stream キーワードを配置します。

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

ルートの移動中に送信された RouteNotes のストリームを受け入れ、他の RouteNotes（他のユーザーからのものなど）を受信します。

これはまさに双方向ストリーミングのユースケースです。双方向ストリーミング RPC。両側が読み取り / 書き込みストリームを使用して一連のメッセージを送信します。2 つのストリームは独立して動作するため、クライアントとサーバーは任意の順序で読み取りと書き込みを行うことができます。たとえば、サーバーはすべてのクライアントメッセージを受信してからレスポンスを書き込むことも、メッセージを読み取ってからメッセージを書き込むことを交互に行うこともできます。各ストリーム内のメッセージの順序は保持されます。このタイプのメソッドを指定するには、リクエストとレスポンスの両方の前に stream キーワードを配置します。

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. クライアントとサーバーのコードを生成する

次に、.proto サービス定義から gRPC クライアントとサーバーのインターフェースを生成する必要があります。これを行うには、特別な gRPC Java プラグインを備えたプロトコルバッファコンパイラ protoc を使用します。gRPC サービスを生成するには、proto3 コンパイラ（proto2 と proto3 の両方の構文をサポート）を使用する必要があります。

Gradle または Maven を使用する場合、protoc ビルドプラグインはビルドの一部として必要なコードを生成できます。独自の .proto ファイルからコードを生成する方法については、grpc-java README をご覧ください。

Gradle 構成を提供しました。

streaming-grpc-java-getting-started ディレクトリから、次のコマンドを入力します。

$ chmod +x gradlew
$ ./gradlew generateProto

次のクラスは、サービス定義（build/generated/sources/proto/main/java の下）から生成されます。

メッセージタイプごとに 1 つずつ（Feature.java、Rectangle.java, ...）。これらには、リクエストとレスポンスのメッセージタイプを生成、シリアル化、取得するためのすべてのプロトコルバッファコードが含まれています。
RouteGuideGrpc.java。RouteGuide サーバーが実装する基本クラス RouteGuideGrpc.RouteGuideImplBase と、RouteGuide サービスで定義されたすべてのメソッド、クライアントが使用するスタブクラス（他の有用なコードも含む）が含まれています。

5. サービスを実装する

まず、RouteGuide サーバーの作成方法を見てみましょう。RouteGuide サービスが機能するには、次の 2 つの要素が必要です。

サービス定義から生成されたサービスインターフェースの実装: サービスの実際の「作業」を行う。
gRPC サーバーを実行して、クライアントからのリクエストをリッスンし、適切なサービス実装にディスパッチします。

RouteGuide を実装する

生成された RouteGuideGrpc.RouteGuideImplBase クラスを拡張する RouteGuideService クラスを実装します。実装は次のようになります。

public void listFeatures(Rectangle request, StreamObserver<Feature> responseObserver) {
        ...
}

public StreamObserver<Point> recordRoute(final StreamObserver<RouteSummary> responseObserver) {

        ...
}

public StreamObserver<RouteNote> routeChat(final StreamObserver<RouteNote> responseObserver) {

        ...
}

各 RPC 実装について詳しく見ていきましょう

サーバーサイドストリーミング RPC

次に、ストリーミング RPC の 1 つを見てみましょう。ListFeatures はサーバーサイドのストリーミング RPC であるため、複数の Features をクライアントに送り返す必要があります。

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();
}

単純な RPC と同様に、このメソッドはリクエストオブジェクト（クライアントが Features を検索する Rectangle）と StreamObserver レスポンスオブザーバーを取得します。

今回は、クライアントに返すために必要な数の Feature オブジェクトを取得し（この場合は、リクエストの Rectangle 内にあるかどうかに基づいて、サービスの機能コレクションから選択します）、各オブジェクトを onNext() メソッドを使用してレスポンスオブザーバーに順番に書き込みます。最後に、単純な RPC と同様に、レスポンスオブザーバーの onCompleted() メソッドを使用して、レスポンスの書き込みが完了したことを gRPC に伝えます。

クライアントサイドストリーミング RPC

次に、もう少し複雑なクライアントサイドストリーミングメソッド RecordRoute() を見てみましょう。ここでは、クライアントから Points のストリームを取得し、旅行に関する情報を含む単一の RouteSummary を返します。

@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();
    }
  };
}

ご覧のとおり、前のメソッドと同様に、このメソッドは StreamObserver responseObserver パラメータを取得しますが、今回はクライアントが Points を書き込むための StreamObserver を返します。

メソッドの本文で、返す匿名 StreamObserver をインスタンス化します。この中で、次の処理を行います。

onNext() メソッドをオーバーライドして、クライアントがメッセージストリームに Point を書き込むたびに、機能などの情報を取得します。
onCompleted() メソッド（クライアントがメッセージの書き込みを完了したときに呼び出される）をオーバーライドして、RouteSummary を入力してビルドします。次に、メソッド独自のレスポンスオブザーバーの onNext() を RouteSummary で呼び出し、その onCompleted() メソッドを呼び出してサーバー側からの呼び出しを終了します。

双方向ストリーミング 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, "Encountered error in routeChat", t);
    }

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

クライアントサイドのストリーミングの例と同様に、StreamObserver を取得して返しますが、今回はクライアントがメッセージストリームにメッセージを書き込んでいる間に、メソッドのレスポンスオブザーバーを介して値を返します。ここでの読み取りと書き込みの構文は、クライアントストリーミングメソッドとサーバーストリーミングメソッドの構文とまったく同じです。各サイドは常に相手のメッセージを書き込まれた順に取得しますが、クライアントとサーバーの両方が任意の順序で読み取りと書き込みを行うことができます。ストリームは完全に独立して動作します。

サーバーを起動する

すべてのメソッドを実装したら、クライアントが実際にサービスを使用できるように、gRPC サーバーを起動する必要があります。次のスニペットは、RouteGuide サービスでこれを行う方法を示しています。

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 を使用してサーバーをビルドして起動します。

Google による処理方法:

ビルダーの forPort() メソッドを使用して、クライアントリクエストをリッスンするために使用するアドレスとポートを指定します。
サービス実装クラス RouteGuideService のインスタンスを作成し、ビルダーの addService() メソッドに渡します。
ビルダーで build() と start() を呼び出して、サービスの RPC サーバーを作成して起動します。

ServerBuilder にはすでにポートが組み込まれているため、ポートを渡すのはロギングに使用するためだけです。

6. クライアントを作成する

このセクションでは、RouteGuide サービスのクライアントを作成する方法について説明します。完全なサンプルクライアントコードは、../complete/src/main/java/io/grpc/complete/routeguide/ RouteGuideClient.java で確認できます。

スタブをインスタンス化する

サービスメソッドを呼び出すには、まず スタブ（実際には 2 つのスタブ）を作成する必要があります。

ブロッキング/同期スタブ: RPC 呼び出しはサーバーからのレスポンスを待機し、レスポンスを返すか例外を発生させます。
サーバーに非ブロック呼び出しを行う非ブロック/非同期スタブ。レスポンスは非同期で返されます。特定のタイプのストリーミング呼び出しは、非同期スタブを使用した場合にのみ行うことができます。

まず、接続先のサーバーのアドレスとポートを指定して、スタブの gRPC チャネルを作成する必要があります。

  public static void main(String[] args) throws InterruptedException {
    String target = "localhost:8980";
    if (args.length > 0) {
      if ("--help".equals(args[0])) {
        System.err.println("Usage: [target]");
        System.err.println("");
        System.err.println("  target  The server to connect to. Defaults to " + target);
        System.exit(1);
      }
      target = args[0];
    }

    List<Feature> features;
    try {
      features = RouteGuideUtil.parseFeatures(RouteGuideUtil.getDefaultFeaturesFile());
    } catch (IOException ex) {
      ex.printStackTrace();
      return;
    }

    ManagedChannel channel = Grpc.newChannelBuilder(target, InsecureChannelCredentials.create())
        .build();
    try {
      RouteGuideClient client = new RouteGuideClient(channel);

      // Looking for features between 40, -75 and 42, -73.
      client.listFeatures(400000000, -750000000, 420000000, -730000000);

      // Record a few randomly selected points from the features file.
      client.recordRoute(features, 10);

      // Send and receive some notes.
      CountDownLatch finishLatch = client.routeChat();

      if (!finishLatch.await(1, TimeUnit.MINUTES)) {
        client.warning("routeChat did not finish within 1 minutes");
      }
    } finally {
      channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
    }
  }

ManagedChannelBuilder を使用してチャネルを作成します。

これで、チャネルを使用して、.proto から生成した RouteGuideGrpc クラスで提供される newStub メソッドと newBlockingStub メソッドを使用してスタブを作成できるようになりました。

public RouteGuideClient(Channel channel) {
    blockingStub = RouteGuideGrpc.newBlockingStub(channel);
    asyncStub = RouteGuideGrpc.newStub(channel);
  }

ブロックされていない場合は非同期です。

サービスメソッドを呼び出す

次に、サービスメソッドを呼び出す方法を見てみましょう。ブロッキングスタブから作成された RPC はブロッキング/同期モードで動作します。つまり、RPC 呼び出しはサーバーからの応答を待機し、応答またはエラーを返します。

サーバーサイドストリーミング RPC

次に、ListFeatures へのサーバーサイドストリーミング呼び出しを見てみましょう。これは、地理的な Feature のストリームを返します。

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

ご覧のとおり、Getting_Started_With_gRPC_Java の Codelab で説明した単純な単項 RPC と非常によく似ています。ただし、単一の Feature を返す代わりに、クライアントが返されたすべての Features を読み取るために使用できる Iterator を返します。

クライアントサイドストリーミング RPC

次に、少し複雑なクライアントサイドストリーミングメソッド RecordRoute を見てみましょう。ここでは、Points のストリームをサーバーに送信し、単一の RouteSummary を取得します。このメソッドでは、非同期スタブを使用する必要があります。サーバーの作成をすでに読んでいる場合は、このコードがよく似ていることに気づくでしょう。非同期ストリーミング RPC は両側で同様の方法で実装されます。

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

ご覧のとおり、このメソッドを呼び出すには、サーバーが RouteSummary レスポンスで呼び出すための特別なインターフェースを実装する StreamObserver を作成する必要があります。StreamObserver で、次の処理を行います。

サーバーが RouteSummary をメッセージストリームに書き込んだときに返された情報を出力するように、onNext() メソッドをオーバーライドします。
onCompleted() メソッド（サーバー側で呼び出しが完了したときに呼び出される）をオーバーライドして CountDownLatch を減らし、サーバーが書き込みを完了したかどうかを確認できるようにします。

次に、StreamObserver を非同期スタブの recordRoute() メソッドに渡し、サーバーに送信する Points を書き込むために、独自の StreamObserver リクエストオブザーバーを取得します。ポイントの書き込みが完了したら、リクエストオブザーバーの onCompleted() メソッドを使用して、クライアント側での書き込みが完了したことを gRPC に通知します。完了したら、CountDownLatch を確認して、サーバー側で完了したかどうかを確認します。

双方向ストリーミング RPC

最後に、双方向ストリーミング RPC RouteChat() を見てみましょう。

public CountDownLatch routeChat() {
    info("*** RouteChat");
    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) {
            warning("RouteChat Failed: {0}", Status.fromThrowable(t));
            finishLatch.countDown();
          }

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

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

      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();

    // return the latch while receiving happens asynchronously
    return finishLatch;
  }

クライアントサイドストリーミングの例と同様に、StreamObserver 応答オブザーバーを取得して返します。ただし、今回は、サーバーがメッセージストリームにメッセージを書き込んでいる間に、メソッドの応答オブザーバーを介して値を送信します。ここでの読み取りと書き込みの構文は、クライアントストリーミングメソッドとまったく同じです。各サイドは常に相手のメッセージを書き込まれた順に取得しますが、クライアントとサーバーの両方が任意の順序で読み取りと書き込みを行うことができます。ストリームは完全に独立して動作します。

7. ぜひお試しください。

start_here ディレクトリから:

$ ./gradlew installDist

これにより、コードがコンパイルされ、jar にパッケージ化され、例を実行するスクリプトが作成されます。これらは build/install/start_here/bin/ ディレクトリに作成されます。スクリプトは route-guide-server と route-guide-client です。

クライアントを起動する前に、サーバーが実行されている必要があります。

サーバーを実行します。

$ ./build/install/start_here/bin/route-guide-server

クライアントを実行します。

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

8. 次のステップ

gRPC の概要とコアコンセプトで gRPC の仕組みを確認する
基本チュートリアルに取り組む
API リファレンスをご覧ください。

gRPC-Java スタートガイド - ストリーミング

1. はじめに

学習内容

2. 始める前に

前提条件

コードを取得する

3. メッセージとサービスを定義する

メッセージ タイプを定義する

サービス メソッドを定義する

ListFeatures

RecordRoute

RouteChat

4. クライアントとサーバーのコードを生成する

5. サービスを実装する

RouteGuide を実装する

サーバーサイド ストリーミング RPC

クライアントサイド ストリーミング RPC

双方向ストリーミング RPC

サーバーを起動する

6. クライアントを作成する

スタブをインスタンス化する

サービス メソッドを呼び出す

サーバーサイド ストリーミング RPC

クライアントサイド ストリーミング RPC

双方向ストリーミング RPC

7. ぜひお試しください。

8. 次のステップ

メッセージタイプを定義する

サービスメソッドを定義する

サーバーサイドストリーミング RPC

クライアントサイドストリーミング RPC

サービスメソッドを呼び出す

サーバーサイドストリーミング RPC

クライアントサイドストリーミング RPC