📚 课程目录第3章服务端流模式详解

3. 服务端流模式详解：Proto定义服务端流RPC、服务端流式响应实现原理、流式消息的发送与结束

服务端流（Server Streaming）是 gRPC 四种通信模式中应用最广泛的模式之一。与一元 RPC 不同，服务端流允许客户端发起一次请求后，服务端通过同一个 RPC 调用连续返回多条消息，直到所有数据发送完毕。这种模式特别适合推送实时数据、批量查询结果、日志流、监控指标等场景。

3.1 Proto 定义服务端流 RPC

在 Protocol Buffers（Proto3）中，服务端流 RPC 的定义非常简洁：在服务端方法的返回类型前加上 stream 关键字，客户端请求保持普通消息类型。以下是一个典型的服务端流 RPC 定义示例：

syntax = "proto3";

package streaming;

// 客户端请求：指定要查询的传感器 ID 和时间范围
message SensorQueryRequest {
    string sensor_id = 1;
    int64 start_timestamp = 2;
    int64 end_timestamp = 3;
}

// 服务端流式响应：每条消息包含一个时间点的传感器读数
message SensorReading {
    int64 timestamp = 1;
    double temperature = 2;
    double humidity = 3;
    double pressure = 4;
}

// 定义服务端流 RPC 服务
service SensorService {
    // 服务端流方法：客户端发送一次请求，服务端返回多条 SensorReading
    rpc GetSensorReadings(SensorQueryRequest) returns (stream SensorReading);
}

关键语法点： stream 关键字只能修饰 returns 后面的响应类型，不能修饰请求类型。如果请求也需要流式传输，则属于双向流模式。

生成代码后，服务端需要实现 GetSensorReadings 方法，客户端则通过返回的流对象逐条读取响应消息。

3.2 服务端流式响应实现原理

服务端流的底层实现基于 HTTP/2 的帧复用机制。一次服务端流 RPC 在 HTTP/2 层面对应一个独立的流（Stream），所有响应消息通过该流上的多个 DATA 帧按序传输。下图展示了服务端流 RPC 的协议栈与消息传递过程：

服务端流的实现核心在于 gRPC 服务端流式响应观察者（StreamObserver）。当客户端发起调用后，gRPC 框架自动创建 StreamObserver<SensorReading> 实例并传递给服务端业务逻辑。服务端通过该观察者控制消息的发送与结束。

3.3 流式消息的发送与结束

服务端流的业务实现遵循固定的生命周期：初始化 → 逐条发送 → 正常结束 或 异常终止。以下是一个完整的 Java 服务端实现示例：

public class SensorServiceImpl extends SensorServiceGrpc.SensorServiceImplBase {

    @Override
    public void getSensorReadings(SensorQueryRequest request,
                                   StreamObserver<SensorReading> responseObserver) {
        try {
            // 1. 从请求中提取参数
            String sensorId = request.getSensorId();
            long startTs = request.getStartTimestamp();
            long endTs = request.getEndTimestamp();

            // 2. 模拟从数据库或缓存中批量查询传感器数据
            List<SensorData> dataList = sensorRepository.query(sensorId, startTs, endTs);

            // 3. 逐条发送响应消息
            for (SensorData data : dataList) {
                SensorReading reading = SensorReading.newBuilder()
                        .setTimestamp(data.getTimestamp())
                        .setTemperature(data.getTemperature())
                        .setHumidity(data.getHumidity())
                        .setPressure(data.getPressure())
                        .build();

                // 关键调用：发送单条消息
                responseObserver.onNext(reading);

                // 可选：模拟发送间隔，控制背压
                // Thread.sleep(10);
            }

            // 4. 所有消息发送完毕，正常结束流
            responseObserver.onCompleted();

        } catch (Exception e) {
            // 5. 异常处理：终止流并传递错误信息
            responseObserver.onError(
                Status.INTERNAL
                    .withDescription("读取传感器数据失败: " + e.getMessage())
                    .asRuntimeException()
            );
        }
    }
}

重要： onNext() 和 onCompleted() 必须在同一个 gRPC 工作线程中按序调用，不能并发调用。如果需要在多个线程中发送消息，必须使用同步机制保证顺序。gRPC 框架不保证并发 onNext() 的线程安全。

下表总结了服务端流中三个核心方法的作用与调用约束：

方法	作用	调用次数	调用后状态
`onNext(response)`	发送一条响应消息给客户端	0 到 N 次	流保持打开，可继续发送
`onCompleted()`	正常结束流，通知客户端所有消息已发送完毕	恰好 1 次（正常路径）	流关闭，不可再调用 onNext
`onError(Throwable)`	异常终止流，传递 gRPC 错误状态给客户端	恰好 1 次（异常路径）	流关闭，客户端收到错误码

客户端接收服务端流时，通常使用阻塞式或响应式 API 逐条读取：

// Java 客户端阻塞式读取示例
SensorServiceGrpc.SensorServiceBlockingStub stub = 
    SensorServiceGrpc.newBlockingStub(channel);

SensorQueryRequest request = SensorQueryRequest.newBuilder()
        .setSensorId("sensor-01")
        .setStartTimestamp(1700000000L)
        .setEndTimestamp(1700003600L)
        .build();

Iterator<SensorReading> readings = stub.getSensorReadings(request);
while (readings.hasNext()) {
    SensorReading reading = readings.next();
    System.out.printf("时间: %d, 温度: %.2f°C, 湿度: %.2f%%%n",
            reading.getTimestamp(),
            reading.getTemperature(),
            reading.getHumidity());
}

  性能要点： 服务端流模式下，每条 onNext() 调用都会触发一次 gRPC 框架内部的序列化和 HTTP/2 DATA 帧写入。如果单条消息体积很小（如几十字节），建议在业务层做批量聚合，将多条数据合并为一条消息发送，以减少帧开销和系统调用次数。但需注意单条 gRPC 消息默认最大为 4MB，可通过 MaxMessageSize 配置调整。

服务端流的结束信号由 onCompleted() 触发，gRPC 框架会在 HTTP/2 层面发送一个带有 END_STREAM 标志的 HEADERS 帧（包含 gRPC 状态码和可选尾随元数据）。客户端收到该帧后，hasNext() 返回 false，流式读取循环自然终止。如果服务端调用 onError()，则客户端会收到对应的 gRPC 状态码（如 INTERNAL、DEADLINE_EXCEEDED）并抛出异常。