3. gRPC 通信机制：一元RPC、服务端流式RPC、客户端流式RPC、双向流式RPC

gRPC 基于 HTTP/2 协议，定义了四种核心通信模式。在分布式事务压测场景中，不同模式对应不同的数据交换特征：请求-响应延迟、吞吐量模型以及背压（Backpressure）行为均存在显著差异。理解这四种模式是设计压测脚本、配置负载模型以及分析性能瓶颈的前提。

3.1 四种通信模式概述

下表从消息流方向、典型事务场景和压测关注点三个维度进行对比：

3.2 一元 RPC（Unary RPC）

一元 RPC 是最基础的请求-响应模式。客户端发送一个请求消息，服务端返回一个响应消息。在 HTTP/2 层面，它对应一个 Stream 上的 HEADERS + DATA 帧交换。

// Proto 定义示例
service TransactionService {
  rpc Commit (CommitRequest) returns (CommitResponse);
}

message CommitRequest {
  string transaction_id = 1;
  repeated string participant_ids = 2;
}

message CommitResponse {
  bool success = 1;
  string error_message = 2;
}

压测关键参数：

• 连接数： 每个 HTTP/2 连接可承载多个并发 Stream，但受 MAX_CONCURRENT_STREAMS 限制。
• 请求大小： 事务负载大小直接影响序列化/反序列化耗时和网络带宽。
• 超时设置： 分布式事务中，超时会导致回滚，压测时需区分“业务超时”与“网络超时”。

3.3 服务端流式 RPC（Server Streaming RPC）

客户端发送一个请求，服务端返回一个消息流。适用于服务端需要持续推送数据的场景，例如事务状态变更通知。

// Proto 定义示例
service TransactionService {
  rpc WatchStatus (WatchRequest) returns (stream StatusEvent);
}

message WatchRequest {
  string transaction_id = 1;
}

message StatusEvent {
  string phase = 1;
  int64 timestamp = 2;
  string detail = 3;
}

压测关注点：

• 流控（Flow Control）： HTTP/2 基于窗口的流控机制会限制服务端发送速率。压测时需监控 WINDOW_UPDATE 帧频率。
• 消息批处理： 服务端通常将多条消息合并为一个 DATA 帧发送，批处理大小影响吞吐。
• 客户端消费能力： 如果客户端消费慢，背压会传导至服务端，导致服务端发送缓冲区积压。

3.4 客户端流式 RPC（Client Streaming RPC）

客户端发送一个消息流，服务端在接收完所有消息后返回一个响应。适用于批量数据上传场景，例如批量事务日志上报。

// Proto 定义示例
service TransactionService {
  rpc BatchUpload (stream TransactionLog) returns (UploadSummary);
}

message TransactionLog {
  string log_id = 1;
  bytes payload = 2;
}

message UploadSummary {
  int32 total_received = 1;
  int32 failed_count = 2;
}

压测关注点：

• 发送速率控制： 客户端发送过快会导致服务端内存压力，发送过慢则无法打满吞吐。
• 消息边界： 每条消息的序列化大小影响 HTTP/2 DATA 帧的分片。
• 服务端合并延迟： 服务端通常需要等待所有消息到达后才返回响应，压测需记录“最后一条消息发送时间”到“收到响应”的延迟。

3.5 双向流式 RPC（Bidirectional Streaming RPC）

客户端和服务端可以独立发送消息流，两者之间是全双工通信。在分布式事务中，这最接近协调器与参与者之间的交互模型：协调器发送准备请求，参与者流式返回准备结果。

// Proto 定义示例
service TransactionService {
  rpc TwoPhaseCommit (stream CoordinatorMessage) returns (stream ParticipantMessage);
}

message CoordinatorMessage {
  oneof msg {
    PrepareRequest prepare = 1;
    CommitRequest commit = 2;
    AbortRequest abort = 3;
  }
}

message ParticipantMessage {
  oneof msg {
    PrepareResponse prepare_ack = 1;
    CommitResponse commit_ack = 2;
  }
}

压测关注点：

• 全双工吞吐： 需要同时测量上行和下行吞吐量，两者可能不对称。
• 消息顺序性： 同一 Stream 上的消息顺序由应用层保证，压测需验证乱序到达时的处理逻辑。
• 资源开销： 每个双向流会占用一个 HTTP/2 Stream，大量并发流可能导致服务端 Stream 耗尽。

3.6 四种模式对比总结

以下 SVG 图展示了四种通信模式在分布式事务压测中的消息流特征与资源模型：

3.7 压测脚本适配要点

针对四种模式，压测脚本需要做以下差异化适配：

1. 一元 RPC： 使用同步阻塞 Stub 或异步回调 Stub，关注连接池配置和超时重试策略。
2. 服务端流式 RPC： 必须实现 StreamObserver 的 onNext()、onError()、onCompleted() 回调，并在 onNext() 中记录消息到达时间戳。
3. 客户端流式 RPC： 使用 StreamObserver 的 onNext() 发送消息，注意控制发送间隔避免 OOM。发送完成后调用 onCompleted()。
4. 双向流式 RPC： 使用 StreamObserver 同时处理读写，建议为读写分别分配独立线程池，避免相互阻塞。

// 双向流式 RPC 压测脚本核心片段（Java + gRPC）
// 注意：实际压测中需配合线程池和计数器管理并发流

ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
    .usePlaintext()
    .maxInboundMessageSize(10 * 1024 * 1024)
    .build();

TransactionServiceGrpc.TransactionServiceStub asyncStub = TransactionServiceGrpc.newStub(channel);

// 创建双向流
StreamObserver<CoordinatorMessage> requestObserver = asyncStub.twoPhaseCommit(
    new StreamObserver<ParticipantMessage>() {
        @Override
        public void onNext(ParticipantMessage msg) {
            // 记录响应时间戳
            responseTimestamps.add(System.nanoTime());
        }

        @Override
        public void onError(Throwable t) {
            errorCount.incrementAndGet();
        }

        @Override
        public void onCompleted() {
            completedCount.incrementAndGet();
        }
    });

// 发送准备请求
CoordinatorMessage prepareMsg = CoordinatorMessage.newBuilder()
    .setPrepare(PrepareRequest.newBuilder().setTransactionId("tx-001"))
    .build();
requestObserver.onNext(prepareMsg);

// 发送提交请求
CoordinatorMessage commitMsg = CoordinatorMessage.newBuilder()
    .setCommit(CommitRequest.newBuilder().setTransactionId("tx-001"))
    .build();
requestObserver.onNext(commitMsg);

// 结束流
requestObserver.onCompleted();