Mulai Menggunakan gRPC-Java - Streaming

1. Pengantar

Dalam codelab ini, Anda akan menggunakan gRPC-Java untuk membuat klien dan server yang membentuk dasar aplikasi pemetaan rute yang ditulis dalam Java.

Di akhir tutorial, Anda akan memiliki klien yang terhubung ke server jarak jauh menggunakan gRPC untuk mendapatkan informasi tentang fitur di rute klien, membuat ringkasan rute klien, dan bertukar informasi rute seperti info terbaru lalu lintas dengan server dan klien lain.

Layanan ini ditentukan dalam file Protocol Buffers, yang akan digunakan untuk membuat kode boilerplate bagi klien dan server sehingga keduanya dapat berkomunikasi satu sama lain, sehingga menghemat waktu dan upaya Anda dalam menerapkan fungsi tersebut.

Kode yang dihasilkan ini tidak hanya menangani kompleksitas komunikasi antara server dan klien, tetapi juga serialisasi dan deserialisasi data.

Yang akan Anda pelajari

• Cara menggunakan Protocol Buffers untuk menentukan API layanan.
• Cara membangun klien dan server berbasis gRPC dari definisi Protocol Buffers menggunakan pembuatan kode otomatis.
• Pemahaman tentang komunikasi streaming klien-server dengan gRPC.

Codelab ini ditujukan bagi developer Java yang baru mengenal gRPC atau ingin mempelajari kembali gRPC, atau siapa pun yang tertarik untuk membangun sistem terdistribusi. Tidak diperlukan pengalaman gRPC sebelumnya.

2. Sebelum memulai

Prasyarat

• Versi JDK 24.

Mendapatkan kode

Agar Anda tidak perlu memulai dari awal, codelab ini menyediakan kerangka kode sumber aplikasi yang dapat Anda selesaikan. Langkah-langkah berikut akan menunjukkan cara menyelesaikan aplikasi, termasuk menggunakan plugin compiler buffer protokol untuk membuat kode gRPC boilerplate.

Pertama, buat direktori kerja codelab dan cd ke dalamnya:

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

Download dan ekstrak 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

Atau, Anda dapat mendownload file .zip yang hanya berisi direktori codelab dan mengekstraknya secara manual.

Download ZIP

Kode sumber yang sudah selesai tersedia di GitHub jika Anda ingin melewati pengetikan implementasi.

3. Menentukan pesan dan layanan

Langkah pertama Anda adalah menentukan layanan gRPC aplikasi, metode RPC, serta jenis pesan permintaan dan responsnya menggunakan Protocol Buffers. Layanan Anda akan menyediakan:

• Metode RPC yang disebut ListFeatures, RecordRoute, dan RouteChat yang diimplementasikan server dan dipanggil klien.
• Jenis pesan Point, Feature, Rectangle, RouteNote, dan RouteSummary, yang merupakan struktur data yang dipertukarkan antara klien dan server saat memanggil metode di atas.

Protocol Buffers biasanya dikenal sebagai protobuf. Untuk mengetahui informasi selengkapnya tentang terminologi gRPC, lihat Konsep inti, arsitektur, dan siklus proses gRPC.

Metode RPC ini dan jenis pesannya akan ditentukan dalam file proto/routeguide/route_guide.proto kode sumber yang diberikan.

Mari buat file route_guide.proto.

Karena kita membuat kode Java dalam contoh ini, kita telah menentukan opsi file java_package dalam .proto:

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

Menentukan jenis pesan

Dalam file proto/routeguide/route_guide.proto kode sumber, tentukan terlebih dahulu jenis pesan Point. Point mewakili pasangan koordinat lintang-bujur di peta. Untuk codelab ini, gunakan bilangan bulat untuk koordinat:

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

Angka 1 dan 2 adalah nomor ID unik untuk setiap kolom dalam struktur message.

Selanjutnya, tentukan jenis pesan Feature. Feature menggunakan kolom string untuk nama atau alamat pos sesuatu di lokasi yang ditentukan oleh Point:

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

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

Agar beberapa titik dalam suatu area dapat di-streaming ke klien, Anda memerlukan pesan Rectangle yang merepresentasikan persegi panjang lintang-bujur, yang direpresentasikan sebagai dua titik yang berlawanan secara diagonal lo dan hi:

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

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

Selain itu, pesan RouteNote yang mewakili pesan yang dikirim saat berada di titik tertentu:

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

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

Terakhir, Anda memerlukan pesan RouteSummary. Pesan ini diterima sebagai respons terhadap RPC RecordRoute, yang dijelaskan di bagian berikutnya. Objek ini berisi jumlah titik individual yang diterima, jumlah fitur yang terdeteksi, dan total jarak yang ditempuh sebagai jumlah kumulatif jarak antara setiap titik.

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

Menentukan metode layanan

Untuk menentukan layanan, Anda menentukan layanan bernama dalam file .proto. File route_guide.proto memiliki struktur service bernama RouteGuide yang menentukan satu atau beberapa metode yang disediakan oleh layanan aplikasi.

Saat menentukan metode RPC di dalam definisi layanan, Anda menentukan jenis permintaan dan responsnya. Di bagian codelab ini, mari kita tentukan:

ListFeatures

Mendapatkan objek Feature yang tersedia dalam Rectangle tertentu. Hasil di-streaming, bukan ditampilkan sekaligus karena persegi panjang dapat mencakup area yang luas dan berisi sejumlah besar fitur.

Untuk aplikasi ini, Anda akan menggunakan RPC streaming sisi server: klien mengirimkan permintaan ke server dan mendapatkan stream untuk membaca kembali urutan pesan. Klien membaca dari stream yang ditampilkan hingga tidak ada lagi pesan. Seperti yang dapat Anda lihat dalam contoh kami, Anda menentukan metode streaming sisi server dengan menempatkan kata kunci stream sebelum jenis respons.

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

RecordRoute

Menerima aliran Poin pada rute yang dilalui, menampilkan RouteSummary saat penelusuran selesai.

RPC streaming sisi klien cocok dalam kasus ini: klien menulis urutan pesan dan mengirimkannya ke server, lagi-lagi menggunakan stream yang disediakan. Setelah klien selesai menulis pesan, klien akan menunggu server membaca semua pesan tersebut dan menampilkan responsnya. Anda menentukan metode streaming sisi klien dengan menempatkan kata kunci stream sebelum jenis permintaan.

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

RouteChat

Menerima aliran RouteNotes yang dikirim saat rute sedang dilalui, sambil menerima RouteNotes lain (misalnya dari pengguna lain).

Inilah kasus penggunaan yang tepat untuk streaming dua arah. RPC streaming dua arah di mana kedua sisi mengirimkan urutan pesan menggunakan stream baca-tulis. Kedua aliran beroperasi secara independen, sehingga klien dan server dapat membaca dan menulis dalam urutan apa pun yang mereka inginkan: misalnya, server dapat menunggu untuk menerima semua pesan klien sebelum menulis responsnya, atau dapat membaca pesan lalu menulis pesan, atau kombinasi pembacaan dan penulisan lainnya. Urutan pesan di setiap aliran akan tetap terjaga. Anda menentukan jenis metode ini dengan menempatkan kata kunci stream sebelum permintaan dan respons.

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

4. Membuat kode klien dan server

Selanjutnya, kita perlu membuat antarmuka klien dan server gRPC dari definisi layanan .proto. Kami melakukannya menggunakan compiler protocol buffer protoc dengan plugin gRPC Java khusus. Anda harus menggunakan compiler proto3 (yang mendukung sintaksis proto2 dan proto3) untuk membuat layanan gRPC.

Saat menggunakan Gradle atau Maven, plugin build protoc dapat membuat kode yang diperlukan sebagai bagian dari build. Anda dapat melihat README grpc-java untuk mengetahui cara membuat kode dari file .proto Anda sendiri.

Kami telah menyediakan konfigurasi Gradle.

Dari direktori streaming-grpc-java-getting-started, masukkan

$ chmod +x gradlew
$ ./gradlew generateProto

Class berikut dihasilkan dari definisi layanan kita (di bawah build/generated/sources/proto/main/java):

• Satu untuk setiap jenis pesan: Feature.java, Rectangle.java, ... yang berisi semua kode buffer protokol untuk mengisi, melakukan serialisasi, dan mengambil jenis pesan permintaan dan respons kita.
• RouteGuideGrpc.java yang berisi (bersama dengan beberapa kode berguna lainnya) class dasar untuk diterapkan server RouteGuide, RouteGuideGrpc.RouteGuideImplBase, dengan semua metode yang ditentukan di class layanan RouteGuide dan stub untuk digunakan klien

5. Menerapkan layanan

Pertama, mari kita lihat cara membuat server RouteGuide. Ada dua bagian untuk membuat layanan RouteGuide kami melakukan tugasnya:

• Mengimplementasikan antarmuka layanan yang dihasilkan dari definisi layanan kita: melakukan "pekerjaan" layanan kita yang sebenarnya.
• Menjalankan server gRPC untuk memproses permintaan dari klien dan mengirimkannya ke implementasi layanan yang tepat.

Mengimplementasikan RouteGuide

Kita akan mengimplementasikan class RouteGuideService yang akan memperluas class RouteGuideGrpc.RouteGuideImplBase yang dihasilkan. Seperti inilah tampilan implementasinya.

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

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

        ...
}

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

        ...
}

Mari kita pelajari setiap implementasi RPC secara mendetail

RPC streaming sisi server

Selanjutnya, mari kita lihat salah satu RPC streaming kami. ListFeatures adalah RPC streaming sisi server, jadi kita perlu mengirim kembali beberapa Features ke klien.

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

Seperti RPC sederhana, metode ini mendapatkan objek permintaan (Rectangle tempat klien ingin menemukan Features) dan pengamat respons StreamObserver.

Kali ini, kita mendapatkan sebanyak mungkin objek Feature yang perlu kita kembalikan ke klien (dalam hal ini, kita memilihnya dari kumpulan fitur layanan berdasarkan apakah objek tersebut berada di dalam Rectangle permintaan kita), dan menulisnya satu per satu ke pengamat respons menggunakan metode onNext()-nya. Terakhir, seperti pada RPC sederhana, kita menggunakan metode onCompleted() pengamat respons untuk memberi tahu gRPC bahwa kita telah selesai menulis respons.

RPC streaming sisi klien

Sekarang mari kita lihat sesuatu yang sedikit lebih rumit: metode streaming sisi klien RecordRoute(), tempat kita mendapatkan aliran Points dari klien dan menampilkan satu RouteSummary dengan informasi tentang perjalanannya.

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

Seperti yang dapat Anda lihat, seperti jenis metode sebelumnya, metode kita mendapatkan parameter StreamObserver responseObserver, tetapi kali ini metode tersebut menampilkan StreamObserver agar klien dapat menulis Points-nya.

Dalam isi metode, kita membuat instance StreamObserver anonim untuk ditampilkan, yang di dalamnya kita:

• Ganti metode onNext() untuk mendapatkan fitur dan informasi lainnya setiap kali klien menulis Point ke aliran pesan.
• Ganti metode onCompleted() (dipanggil saat klien selesai menulis pesan) untuk mengisi dan membuat RouteSummary. Kemudian, kita memanggil onNext() pengamat respons metode kita sendiri dengan RouteSummary kita, lalu memanggil metode onCompleted()-nya untuk menyelesaikan panggilan dari sisi server.

RPC streaming dua arah

Terakhir, mari kita lihat RPC streaming dua arah 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();
    }
  };
}

Seperti contoh streaming sisi klien, kita mendapatkan dan menampilkan StreamObserver, tetapi kali ini kita menampilkan nilai melalui pengamat respons metode saat klien masih menulis pesan ke aliran pesan mereka. Sintaksis untuk membaca dan menulis di sini sama persis dengan metode streaming klien dan streaming server kami. Meskipun setiap sisi akan selalu mendapatkan pesan dari sisi lain dalam urutan penulisannya, klien dan server dapat membaca dan menulis dalam urutan apa pun — aliran beroperasi sepenuhnya secara independen.

Mulai server

Setelah menerapkan semua metode, kita juga perlu memulai server gRPC agar klien dapat menggunakan layanan kita. Cuplikan berikut menunjukkan cara kami melakukannya untuk layanan RouteGuide kami:

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

Seperti yang dapat Anda lihat, kita membangun dan memulai server menggunakan ServerBuilder.

Untuk melakukannya, kami:

1. Tentukan alamat dan port yang ingin kita gunakan untuk memproses permintaan klien menggunakan metode forPort() builder.
2. Buat instance class implementasi layanan RouteGuideService dan teruskan ke metode addService() builder.
3. Panggil build() dan start() di builder untuk membuat dan memulai server RPC untuk layanan kita.

Karena ServerBuilder sudah menggabungkan port, satu-satunya alasan kita meneruskan port adalah untuk menggunakannya dalam logging.

6. Buat klien

Di bagian ini, kita akan melihat cara membuat klien untuk layanan RouteGuide. Anda dapat melihat kode klien contoh lengkap kami di ../complete/src/main/java/io/grpc/complete/routeguide/ RouteGuideClient.java.

Membuat instance stub

Untuk memanggil metode layanan, kita perlu membuat stub terlebih dahulu, atau lebih tepatnya, dua stub:

• stub pemblokiran/sinkron: berarti panggilan RPC menunggu server merespons, dan akan menampilkan respons atau memunculkan pengecualian.
• stub non-blocking/asinkron yang membuat panggilan non-blocking ke server, dengan respons ditampilkan secara asinkron. Anda dapat melakukan jenis panggilan streaming tertentu hanya dengan menggunakan stub asinkron.

Pertama, kita perlu membuat channel gRPC untuk stub, dengan menentukan alamat dan port server yang ingin kita hubungkan:

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

Kita menggunakan ManagedChannelBuilder untuk membuat channel.

Sekarang kita dapat menggunakan saluran untuk membuat stub menggunakan metode newStub dan newBlockingStub yang disediakan di class RouteGuideGrpc yang kita buat dari .proto.

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

Ingat, jika tidak memblokir, berarti asinkron

Panggil metode layanan

Sekarang, mari kita lihat cara memanggil metode layanan. Perhatikan bahwa RPC yang dibuat dari stub pemblokiran akan beroperasi dalam mode pemblokiran/sinkron, yang berarti bahwa panggilan RPC menunggu server merespons, dan akan menampilkan respons atau error.

RPC streaming sisi server

Selanjutnya, mari kita lihat panggilan streaming sisi server ke ListFeatures, yang menampilkan aliran Feature geografis:

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

Seperti yang dapat Anda lihat, RPC unary sederhana ini sangat mirip dengan yang kita lihat di codelab Getting_Started_With_gRPC_Java, kecuali metode ini menampilkan Iterator yang dapat digunakan klien untuk membaca semua Features yang ditampilkan, bukan menampilkan satu Feature.

RPC streaming sisi klien

Sekarang kita akan membahas sesuatu yang sedikit lebih rumit: metode streaming sisi klien RecordRoute, tempat kita mengirimkan aliran Points ke server dan mendapatkan kembali satu RouteSummary. Untuk metode ini, kita perlu menggunakan stub asinkron. Jika Anda telah membaca Membuat server, beberapa bagian ini mungkin terlihat sangat familiar - RPC streaming asinkron diimplementasikan dengan cara yang serupa di kedua sisi.

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

Seperti yang dapat Anda lihat, untuk memanggil metode ini, kita perlu membuat StreamObserver, yang menerapkan antarmuka khusus agar server dapat memanggil dengan respons RouteSummary-nya. Dalam StreamObserver, kita:

• Ganti metode onNext() untuk mencetak informasi yang ditampilkan saat server menulis RouteSummary ke aliran pesan.
• Ganti metode onCompleted() (dipanggil saat server telah menyelesaikan panggilan di sisinya) untuk mengurangi CountDownLatch sehingga kita dapat memeriksa apakah server telah selesai menulis.

Kemudian, kita meneruskan StreamObserver ke metode recordRoute() stub asinkron dan mendapatkan kembali pengamat permintaan StreamObserver kita sendiri untuk menulis Points yang akan dikirim ke server. Setelah selesai menulis titik, kita menggunakan metode onCompleted() pengamat permintaan untuk memberi tahu gRPC bahwa kita telah selesai menulis di sisi klien. Setelah selesai, kita memeriksa CountDownLatch untuk melihat apakah server telah selesai di sisinya.

RPC streaming dua arah

Terakhir, mari kita lihat RPC streaming dua arah 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;
  }

Seperti contoh streaming sisi klien, kita mendapatkan dan menampilkan pengamat respons StreamObserver, tetapi kali ini kita mengirim nilai melalui pengamat respons metode saat server masih menulis pesan ke aliran pesan mereka. Sintaksis untuk membaca dan menulis di sini sama persis dengan metode streaming klien kami. Meskipun setiap sisi akan selalu mendapatkan pesan dari sisi lain dalam urutan penulisannya, klien dan server dapat membaca dan menulis dalam urutan apa pun — aliran beroperasi sepenuhnya secara independen.

7. Cobalah!

1. Dari direktori start_here:

$ ./gradlew installDist

Tindakan ini akan mengompilasi kode Anda, mengemasnya dalam jar, dan membuat skrip yang menjalankan contoh. File akan dibuat di direktori build/install/start_here/bin/. Skripnya adalah: route-guide-server dan route-guide-client.

Server harus berjalan sebelum memulai klien.

2. Jalankan server:

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

3. Jalankan klien:

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

8. Langkah berikutnya

• Pelajari cara kerja gRPC di Pengantar gRPC dan Konsep inti
• Pelajari tutorial Dasar-Dasar.
• Pelajari referensi API.