0Pricing
gRPC & High Performance APIs · Pelajaran

Penjelasan Streaming Klien

Pelajari cara menerapkan streaming sisi klien yang memungkinkan klien mengirimkan rangkaian pesan ke server.

Penjelasan Streaming Klien adalah pelajaran gRPC & High Performance APIs gratis di CoddyKit. Ini adalah pelajaran 2 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar gRPC & High Performance APIs, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus gRPC & High Performance APIs mencakup 4 pelajaran total.

Bagian dari pelajaran ini belum diterjemahkan dan ditampilkan dalam bahasa Inggris.

What is Client Streaming?

Welcome to client streaming! In gRPC, client streaming is a communication pattern where the client sends a sequence of messages to the server.

Unlike a simple unary RPC (request-response), the client doesn't just send one message. Instead, it sends a stream of messages, and the server processes them, then sends back a single response at the end.

How Client Streaming Works

Imagine uploading a large file by sending it in many small chunks. The server collects all chunks, rebuilds the file, and then sends a single "upload complete" confirmation.

  • The client initiates the RPC.
  • The client sends multiple messages asynchronously.
  • The server receives and processes these messages.
  • Once the client finishes sending (signals completion), the server sends back a single response.

Defining Client Stream in Protobuf

To define a client-streaming method in your .proto file, you use the stream keyword for the request type, but not for the response type.

Here's an example for a log upload service:

syntax = "proto3";

package client_streaming;

service LogService {
  rpc UploadLogs (stream LogEntry) returns (UploadSummary);
}

message LogEntry {
  string message = 1;
  int64 timestamp = 2;
}

message UploadSummary {
  int32 uploaded_count = 1;
  string status_message = 2;
}

Server: Handling the Client Stream

On the server side, your method will receive a StreamObserver for the client's incoming messages and will use another StreamObserver to send its single response.

The server's StreamObserver will have onNext() for each incoming message, onError() for errors, and onCompleted() when the client finishes sending.

import io.grpc.stub.StreamObserver;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import client_streaming.LogEntry;
import client_streaming.LogServiceGrpc;
import client_streaming.UploadSummary;

public class LogServer {
    private Server server;

    private void start() throws Exception {
        int port = 50051;
        server = ServerBuilder.forPort(port)
            .addService(new LogServiceImpl())
            .build()
            .start();
        System.out.println("Server started, listening on " + port);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            LogServer.this.stop();
            System.err.println("*** server shut down");
        }));
    }

    private void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

    private void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

    public static void main(String[] args) throws Exception {
        final LogServer logServer = new LogServer();
        logServer.start();
        logServer.blockUntilShutdown();
    }

    static class LogServiceImpl extends LogServiceGrpc.LogServiceImplBase {
        @Override
        public StreamObserver<LogEntry> uploadLogs(StreamObserver<UploadSummary> responseObserver) {
            return new StreamObserver<LogEntry>() {
                private int logCount = 0;

                @Override
                public void onNext(LogEntry log) {
                    // Process each log entry as it arrives
                    System.out.println("Received log: " + log.getMessage() + " at " + log.getTimestamp());
                    logCount++;
                }

                @Override
                public void onError(Throwable t) {
                    System.err.println("UploadLogs cancelled or failed: " + t.getMessage());
                    responseObserver.onError(t);
                }

                @Override
                public void onCompleted() {
                    // After all logs are received, send a single summary response
                    UploadSummary summary = UploadSummary.newBuilder()
                        .setUploadedCount(logCount)
                        .setStatusMessage("Successfully processed " + logCount + " log entries.")
                        .build();
                    responseObserver.onNext(summary);
                    responseObserver.onCompleted();
                    System.out.println("Finished processing client stream. Sent summary.");
                }
            };
        }
    }
}

Client: Sending the Stream

On the client side, you'll get a StreamObserver to send your messages. You call onNext() for each message you want to send and finally onCompleted() to signal the end of the stream.

The server's single response will be handled by a separate StreamObserver you provide.

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import client_streaming.LogEntry;
import client_streaming.LogServiceGrpc;
import client_streaming.UploadSummary;

import java.util.concurrent.TimeUnit;

public class LogClient {
    private final LogServiceGrpc.LogServiceStub asyncStub;
    private final ManagedChannel channel;

    public LogClient(String host, int port) {
        channel = ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext() // For demonstration, use plaintext
            .build();
        asyncStub = LogServiceGrpc.newStub(channel);
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    public void uploadMultipleLogs() throws InterruptedException {
        StreamObserver<UploadSummary> responseObserver = new StreamObserver<UploadSummary>() {
            @Override
            public void onNext(UploadSummary summary) {
                System.out.println("Server Response: " + summary.getStatusMessage() + " (" + summary.getUploadedCount() + " logs)");
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("UploadLogs failed: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("Server has completed processing.");
            }
        };

        StreamObserver<LogEntry> requestObserver = asyncStub.uploadLogs(responseObserver);

        try {
            // Send multiple log entries
            LogEntry log1 = LogEntry.newBuilder().setMessage("User login attempt").setTimestamp(System.currentTimeMillis()).build();
            LogEntry log2 = LogEntry.newBuilder().setMessage("Database query executed").setTimestamp(System.currentTimeMillis() + 100).build();
            LogEntry log3 = LogEntry.newBuilder().setMessage("API call completed").setTimestamp(System.currentTimeMillis() + 200).build();

            requestObserver.onNext(log1);
            System.out.println("Client sent log 1");
            Thread.sleep(100); // Simulate some delay
            requestObserver.onNext(log2);
            System.out.println("Client sent log 2");
            Thread.sleep(100);
            requestObserver.onNext(log3);
            System.out.println("Client sent log 3");

            // Mark the end of the client stream
            requestObserver.onCompleted();
            System.out.println("Client finished sending logs.");

            // Wait for server response (handled by responseObserver)
            Thread.sleep(1000); // Give time for server to respond
        } catch (RuntimeException e) {
            requestObserver.onError(e);
            throw e;
        }
    }

    public static void main(String[] args) throws Exception {
        LogClient client = new LogClient("localhost", 50051);
        try {
            client.uploadMultipleLogs();
        } finally {
            client.shutdown();
        }
    }
}

Running the Example

To see client streaming in action:

  1. First, compile your .proto file to generate the necessary Java classes.
  2. Run the LogServer application. It will start listening for requests.
  3. Then, run the LogClient application. It will send three log entries and wait for the server's summary.

Observe the console output from both the client and server to understand the flow of messages.

Key StreamObserver Methods

The StreamObserver interface is crucial for handling streaming RPCs. Both the client and server use implementations of this interface.

  • onNext(T value): Called for each message received in the stream. The client uses this to send messages, the server uses it to receive.
  • onError(Throwable t): Called if an RPC fails or is cancelled.
  • onCompleted(): Called when the stream has finished. The client calls this after sending all messages; the server calls it after sending its final response.

When to Use Client Streaming

Client streaming is ideal for scenarios where a client needs to send a large amount of data or a series of related messages to a server, and only cares about a single final result.

  • Large File Uploads: Sending a file chunk by chunk.
  • Log Aggregation: A client sending many log entries to a central logging service.
  • Batch Operations: Sending a list of items to be processed as a single batch, receiving a summary.
  • Sensor Data Collection: Continuously sending readings from a sensor device.

Quick Check

You're designing a gRPC service where a client needs to send a series of sensor readings to a server, and the server will process them and return a single summary report.

Recap: Client Streaming

You've learned about client-side streaming in gRPC!

  • Client streaming allows a client to send a sequence of messages.
  • The server processes these messages and sends a single response back.
  • In Protobuf, you define it by using the stream keyword for the request type.
  • Both client and server use StreamObserver to manage the flow of messages (onNext(), onError(), onCompleted()).
  • It's great for tasks like uploading large data or sending continuous log entries.

Next, we'll explore server-side streaming!

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Penjelasan Streaming Klien” gratis?

Ya — teks lengkap “Penjelasan Streaming Klien” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus gRPC & High Performance APIs, upgrade ke CoddyKit PRO. Kursus gRPC & High Performance APIs mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Penjelasan Streaming Klien”?

Pelajari cara menerapkan streaming sisi klien yang memungkinkan klien mengirimkan rangkaian pesan ke server. Kamu berlatih gRPC & High Performance APIs dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai gRPC & High Performance APIs?

Tidak diperlukan pengalaman sebelumnya. gRPC & High Performance APIs di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 2 dari 4.

Berapa lama pelajaran “Penjelasan Streaming Klien” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran gRPC & High Performance APIs ini?

Ya. Setiap pelajaran gRPC & High Performance APIs menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

  1. Penjelasan Streaming Server
  2. Penjelasan Streaming Klien
  3. Streaming Dua Arah
  4. Kontrol Aliran & Tekanan Balik Streaming
← Kembali ke gRPC & High Performance APIs