第4章 gRPC实战

第4章：gRPC实战：四种流式RPC模式

好，咱们今天来点真家伙。gRPC 的四种通信模式，说白了就是解决「客户端和服务端怎么聊天」的问题。我刚开始接触 gRPC 时，总觉得流式 RPC 很神秘，后来在项目中用了一次实时数据推送，才真正体会到它的威力。

4.1 一元RPC：最基础的请求-响应

一元 RPC 是最简单的模式。客户端发一个请求，服务端回一个响应。就像你打电话问客服：「我的订单到哪了？」客服查一下告诉你：「已经发货了。」

我个人习惯把一元 RPC 当作入门练习。它的 proto 定义长这样：

service OrderService {
  rpc GetOrder(OrderRequest) returns (OrderResponse);
}

message OrderRequest {
  string order_id = 1;
}

message OrderResponse {
  string status = 1;
  string tracking_info = 2;
}

实现起来也很直接。服务端代码：

class OrderServiceImpl extends OrderServiceGrpc.OrderServiceImplBase {
    @Override
    public void getOrder(OrderRequest request, 
                         StreamObserver<OrderResponse> responseObserver) {
        String orderId = request.getOrderId();
        // 模拟查询数据库
        OrderResponse response = OrderResponse.newBuilder()
            .setStatus("已发货")
            .setTrackingInfo("SF1234567890")
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}

小提示：一元 RPC 适合查询类操作，比如获取用户信息、查询订单状态。我在项目中用它做登录验证，简单可靠。

4.2 服务端流式RPC：服务端主动推送

服务端流式 RPC 是什么场景？客户端发一个请求，服务端可以返回多条数据。就像你问客服：「给我看看这个月的所有订单。」客服一条一条念给你听。

我记得有一次做实时监控系统，需要把服务器日志推送给客户端。一元 RPC 显然不行，因为日志是源源不断的。服务端流式 RPC 正好派上用场。

proto 定义：

service LogService {
  rpc SubscribeLogs(LogRequest) returns (stream LogEntry);
}

message LogRequest {
  string log_level = 1;  // INFO, WARN, ERROR
}

message LogEntry {
  string timestamp = 1;
  string message = 2;
  string level = 3;
}

服务端实现：

class LogServiceImpl extends LogServiceGrpc.LogServiceImplBase {
    @Override
    public void subscribeLogs(LogRequest request,
                              StreamObserver<LogEntry> responseObserver) {
        String level = request.getLogLevel();
        // 模拟持续推送日志
        for (int i = 0; i < 100; i++) {
            LogEntry entry = LogEntry.newBuilder()
                .setTimestamp(System.currentTimeMillis() + "")
                .setMessage("日志内容 #" + i)
                .setLevel(level)
                .build();
            responseObserver.onNext(entry);
            try {
                Thread.sleep(1000);  // 每秒推送一条
            } catch (InterruptedException e) {
                responseObserver.onError(e);
                return;
            }
        }
        responseObserver.onCompleted();
    }
}

注意：服务端流式 RPC 一定要处理好背压问题。我曾经遇到过服务端推送太快，客户端处理不过来，导致内存溢出。建议在客户端使用流控机制。

4.3 客户端流式RPC：客户端批量上传

客户端流式 RPC 正好反过来。客户端可以发多条数据，服务端只回一条。比如上传一个大文件，你可以分块发送，最后服务端告诉你「上传成功」。

说白了，这就是「你一句我一句，但最后我只回一句」。我在做图片上传服务时用过这个模式，客户端把图片分片发送，服务端组装后返回一个 URL。

proto 定义：

service FileUploadService {
  rpc UploadFile(stream FileChunk) returns (UploadResponse);
}

message FileChunk {
  bytes data = 1;
  int32 chunk_index = 2;
  string file_name = 3;
}

message UploadResponse {
  string file_url = 1;
  int64 file_size = 2;
}

客户端实现：

public void uploadFile(String filePath) {
    FileUploadServiceGrpc.FileUploadServiceStub stub = 
        FileUploadServiceGrpc.newStub(channel);
    
    StreamObserver<UploadResponse> responseObserver = new StreamObserver<>() {
        @Override
        public void onNext(UploadResponse response) {
            System.out.println("上传成功，URL: " + response.getFileUrl());
        }
        @Override
        public void onError(Throwable t) {
            System.err.println("上传失败: " + t.getMessage());
        }
        @Override
        public void onCompleted() {
            System.out.println("上传流程结束");
        }
    };
    
    StreamObserver<FileChunk> requestObserver = stub.uploadFile(responseObserver);
    
    try {
        File file = new File(filePath);
        FileInputStream fis = new FileInputStream(file);
        byte[] buffer = new byte[1024 * 64];  // 64KB 一块
        int bytesRead;
        int index = 0;
        
        while ((bytesRead = fis.read(buffer)) != -1) {
            FileChunk chunk = FileChunk.newBuilder()
                .setData(ByteString.copyFrom(buffer, 0, bytesRead))
                .setChunkIndex(index++)
                .setFileName(file.getName())
                .build();
            requestObserver.onNext(chunk);
        }
        requestObserver.onCompleted();
    } catch (Exception e) {
        requestObserver.onError(e);
    }
}

经验之谈：客户端流式 RPC 适合大数据量上传。我建议 chunk 大小设为 64KB 或 128KB，太小了网络开销大，太大了内存压力大。

4.4 双向流式RPC：真正的实时对话

双向流式 RPC 是最灵活的模式。客户端和服务端可以随时发消息，就像两个人聊天，谁都可以先说，谁都可以随时打断。

你想想看，什么场景需要这个？在线客服、实时协作编辑、游戏对战。我记得做即时通讯系统时，双向流式 RPC 简直是天选之子。

proto 定义：

service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user_id = 1;
  string content = 2;
  int64 timestamp = 3;
}

服务端实现：

class ChatServiceImpl extends ChatServiceGrpc.ChatServiceImplBase {
    @Override
    public StreamObserver<ChatMessage> chat(
            StreamObserver<ChatMessage> responseObserver) {
        
        return new StreamObserver<ChatMessage>() {
            @Override
            public void onNext(ChatMessage message) {
                System.out.println("收到消息: " + message.getContent());
                
                // 处理消息，比如转发给其他用户
                ChatMessage reply = ChatMessage.newBuilder()
                    .setUserId("server")
                    .setContent("已收到: " + message.getContent())
                    .setTimestamp(System.currentTimeMillis())
                    .build();
                responseObserver.onNext(reply);
            }
            
            @Override
            public void onError(Throwable t) {
                System.err.println("连接异常: " + t.getMessage());
            }
            
            @Override
            public void onCompleted() {
                System.out.println("客户端断开连接");
                responseObserver.onCompleted();
            }
        };
    }
}

  核心要点：双向流式 RPC 的难点在于状态管理。每个连接对应一个 StreamObserver，你需要维护这些观察者的集合。我曾经用 ConcurrentHashMap 管理在线用户，每个用户对应一个 StreamObserver，实现消息广播。

4.5 四种模式对比

模式	客户端	服务端	典型场景	我踩过的坑
一元 RPC	发1次	回1次	查询、登录	超时设置不合理
服务端流式	发1次	回多次	日志推送、监控	背压导致 OOM
客户端流式	发多次	回1次	文件上传、批量提交	chunk 大小没调好
双向流式	发多次	回多次	聊天、实时协作	连接管理混乱

4.6 实战建议

嗯，这里我要多说几句。选哪种模式，不是看哪个更高级，而是看业务需求。

能一元就别流式：流式 RPC 的复杂度比一元高一个数量级。除非你真的需要持续推送或批量上传，否则老老实实用一元。
注意连接生命周期：流式 RPC 的长连接需要心跳检测。我习惯在应用层加一个 ping-pong 机制，每 30 秒检测一次。
错误处理要到位：流式 RPC 的 onError 回调一定要实现。我曾经偷懒没写，结果连接断了都不知道，数据丢了半天才发现。
性能监控不能少：gRPC 默认有拦截器机制，可以统计每个 RPC 的耗时、成功率。我在项目中用这个做 SLA 监控，效果很好。

避坑指南：我曾经在生产环境遇到过一个问题——双向流式 RPC 的服务端线程数暴涨。后来发现是客户端频繁重连，每次重连都创建新的 StreamObserver，旧的又没有及时清理。解决方案是用一个定时任务，定期清理超时的连接。

好了，四种模式都讲完了。说白了，gRPC 的流式 RPC 就是给微服务通信加上了「实时」的能力。下一章咱们聊聊 gRPC 的拦截器和认证，那才是真正让 gRPC 变得安全可靠的关键。

📚 课程目录