0Pricing
gRPC & High Performance APIs · Lektion

Clientseitiges Streaming erklärt

Lernen Sie, clientseitiges Streaming zu implementieren, sodass ein Client eine Folge von Nachrichten an den Server senden kann.

Clientseitiges Streaming erklärt ist eine kostenlose gRPC & High Performance APIs-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des gRPC & High Performance APIs-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der gRPC & High Performance APIs-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

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!

Häufig gestellte Fragen

Ist die Lektion „Clientseitiges Streaming erklärt“ kostenlos?

Ja — der vollständige Text von „Clientseitiges Streaming erklärt“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des gRPC & High Performance APIs-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der gRPC & High Performance APIs-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Clientseitiges Streaming erklärt“?

Lernen Sie, clientseitiges Streaming zu implementieren, sodass ein Client eine Folge von Nachrichten an den Server senden kann. Du übst gRPC & High Performance APIs mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um gRPC & High Performance APIs zu starten?

Keine Vorkenntnisse erforderlich. gRPC & High Performance APIs auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Clientseitiges Streaming erklärt“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser gRPC & High Performance APIs-Lektion Code schreiben und ausführen?

Ja. Jede gRPC & High Performance APIs-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Serverseitiges Streaming erklärt
  2. Clientseitiges Streaming erklärt
  3. Bidirektionales Streaming
  4. Streaming-Flusskontrolle & Backpressure
← Zurück zu gRPC & High Performance APIs