gRPC Java长连接流并行请求发送及多流管理方案咨询

前置场景 #

我正在使用双向Streaming RPC API，MyRequest和MyResponse均为流式传输，接口定义如下：

service MyStreamedService {
  rpc myOperation(stream MyRequest) returns (stream MyResponse)
}

以下是封装gRPC流的简化类实现：

public class MyStreamWrapper implements StreamObserver<MyResponse> {
  public MyStreamWrapper(ManagedChannel myChannel) {
    myStub = MyStreamedServiceGrpc.newStub(myChannel);
    // 创建流并通过StreamObserver长期持有流引用
    myStream = myStub.myOperation(this);
  }
  
  @Override
  public void onNext(MyResponse r) {
    // 处理响应逻辑未展示
  }
  
  @Override
  public void onError(Throwable t) {
    // 限流逻辑未展示，不限流会占用大量CPU
    // 创建新流
    myStream = myStub.myOperation(this);
  }
  
  @Override
  public void onCompleted() {
    // 服务端已调用StreamObserver<MyRequest>.onCompleted
    // 用异步API创建新流
    myStream = myStub.myOperation(this);
  }
  
  // 业务场景：多线程需要异步发送请求
  public void send(MyRequest r) {
    synchronized(myStream) {
      myStream.onNext(r);
    }
  }
}

问题解答 #

问题1：为什么send方法中访问myStream需要加同步锁？同一条流并行发送无序请求必须做线程同步的原因是什么？如果每个请求都会被封装为带有独立stream-id的HTTP2 DATA帧，该限制是否是gRPC Java客户端实现独有的？ #

回答：gRPC的StreamObserver接口本身明确要求串行调用，并非线程安全，该限制是所有语言gRPC实现的通用规范，不是Java独有。
你存在一个认知误区：同一条gRPC双向流对应唯一的HTTP/2 stream ID，该流上的所有请求都会复用同一个ID发送帧，并不是每个请求分配独立的stream ID。如果多线程并行调用onNext，会导致请求序列化、帧拼接顺序混乱，甚至生成不符合HTTP/2规范的帧，因此必须加锁保证同一时间只有一个线程写入同一条流。
另外你当前的同步实现存在明显bug：onError和onCompleted中修改myStream引用时没有加锁，send方法仅锁当前的myStream实例，无法保护引用替换过程，可能出现空指针、多线程并行写入新流实例的问题。

问题2：当线程从send方法返回时，能保证执行到哪一步？ #

回答：最低仅能保证请求已写入gRPC客户端的缓冲区，没有更高的交付保证。gRPC客户端默认会攒帧批量发送以提升性能，因此返回时请求大概率还未发送到网络，更不可能保证代理、服务端接收或处理。如果客户端缓冲区已满，onNext会直接抛出异常，不会阻塞等待缓冲区空闲。

问题3：目前客户端线程已做同步，同一时间仅会调用一次onNext，这种情况下客户端是否还会压垮服务端？上述send方法是否会通过阻塞客户端线程实现背压？高流量场景下遇到"INTERNAL: RST_STREAM closed stream. HTTP/2 error code: PROTOCOL_ERROR"报错是什么原因？ #

回答：仅串行写入流完全无法避免压垮服务端，也没有任何背压能力：只要客户端发送速度持续高于服务端处理速度，服务端的TCP接收窗口、HTTP/2流控制窗口都会被占满，后续帧无法发送，客户端缓冲区也会被打满，严重时就会触发协议层错误。
你遇到的PROTOCOL_ERROR大概率和两个问题有关：一是前面提到的同步实现bug，流重建时可能出现多线程并行写入新流，导致帧顺序异常触发协议错误；二是高流量下流重建没有加退避逻辑，频繁重建流导致服务端协议层解析异常。

问题4：考虑到stream创建成本很低，维护并复用myStream这种做法是否不常见？ #

回答：长生命周期流复用的做法不算少见，适合对延迟、吞吐要求极高，且请求需要在同一条流上保序的场景，但你的实现缺失了流保活、错误重试退避、旧流请求幂等性兜底等必要逻辑，直接出错立刻重建流反而容易引发惊群效应打垮服务端。
如果你的业务中每个请求是独立的，不需要同一条流保序，那更推荐使用单请求对应单条短生命周期流的模式，该方案开发成本低、异常处理简单，是当前的主流做法。

问题5：由于单条流只能指向一个服务端，是否需要在现有类基础上新增逻辑，在channel创建时生成N条myStream，send请求时轮询使用这些流？也可以选择动态创建新流，再通过大小为N的信号量限制使用流的线程数，该方案是否可行？ #

回答：两种方案都可行，可根据业务场景选择：

• 固定N条长生命周期流轮询的方案，适合请求需要保序、对延迟要求极高的场景，需要额外实现流健康检测、失效流自动替换的逻辑，避免部分流压力过大或失效后请求失败。
• 动态创建流+信号量限制并发数的方案灵活性更高，不需要处理流的保活和健康检测，流的创建成本极低完全可以接受，更适配绝大多数业务场景，只要设置合理的并发上限就能很好地实现背压，避免压垮服务端。

内容的提问来源于stack exchange，提问作者Diarmuid Leonard