双向流 BidiStreamForClient 的问题

[zhangcunli]

1. 我在使用 dubbo-go 的时候发现一个问题，请教一下：
   client: dubbo-go: v3.2.0-rc2
   server: dubbo v3.3.0
   registry: zookeeper

2. 双向流模式：CallBidiStream

3. client: dubbo.NewInstance 方式创建客户端

4. 客户端代码： 其中 biStream 类型为：*triple_protocol.BidiStreamForClient
   for ; ; { //1.receive triple server msg err := biStream.Receive(response) if err != nil { break } }

5. 问题 1：biStream 客户端没有关闭长连接的接口；如果服务端不调用 onCompleted，那么 Receive 会永久 block；
   客户端应当具有主动 Close 能力才对，底层使用的是 grpc，而 grpc 客户端是能够主动关闭连接的。

   问题 2：Java 服务端调用 onCompleted() 后，biStream.Receive 返回 EOF
   但是和服务端的 TCP 连接仍然存在；我知道 dubbo client 内部是连接池，但是我测试了一下，TCP 连接似乎并没有被复用

[No-SilverBullet]

[zhangcunli]

please add my wechat,wxfowen, we can discuss your issues on weekly meeting

OK. 已加，最近略忙，提了 discussion 后没有关注，错过了 12.15 的讨论；再讨论的时候发消息给我

[AlexStocks]

zhangcunli:
1 连接 gRPC 不能 close request https://github.com/apache/dubbo-go/issues/2755；
2 client 连接数不断增长，到 1k 多，没有复用；

有待 cunli 提供最小可验证测试方案。

[AlexStocks]

apache/dubbo-go-pixiu#645

[zhangcunli]

写了个最小验证 demo: git@github.com:zhangcunli/triple-stream-test.git
也许是我使用的方式不正确，按我的 demo，结论如下：

测试 1：testReuseCli()： 复用 *client.Client，那么每创建一个 greet.NewGreetService
都会新一个新的 TCP 连接，TCP 连接不会复用；
测试 2：testReuseSvc()：复用greet.NewGreetService，TCP 连接只会有一个

所以，最佳用法是像 testReuseSvc() 中复用 GreetService，然后用户自己维护一个 GreetService 池吗？这样用起来有些麻烦。

[AlexStocks]

@zhangcunli 反馈：1 如何 close request/response 感觉不自然; 2 tcp 连接池复用问题.

测试案例：https://github.com/zhangcunli/triple-stream-test

[zhangcunli]

@zhangcunli 反馈：1 如何 close request/response 感觉不自然; 2 tcp 连接池复用问题.

测试案例：https://github.com/zhangcunli/triple-stream-test

1. 双向流使用问题：
   #dubbo-go: v3.2.0-rc2;
   dubbo-go, 双向流 client 接口如下：关闭流需要调用两个接口CloseRequest、CloseResponse 才行；
   而且如果 server 是 java server(v3.3.0)，则 CloseResponse 依赖 server 端调用 onComplete，否则会卡死

#proto 生成
type GreetService_GreetStreamClient interface {
	Spec() triple_protocol.Spec
	Peer() triple_protocol.Peer
	Send(*GreetStreamRequest) error
	RequestHeader() http.Header
	CloseRequest() error
	Recv() (*GreetStreamResponse, error)
	ResponseHeader() http.Header
	ResponseTrailer() http.Header
	CloseResponse() error
}

如下是 grpc 双向流 client 接口：直接调用 CloseSend() 就可以关闭双向流

type ClientStream interface {
	Header() (metadata.MD, error)
	Trailer() metadata.MD
	CloseSend() error
	Context() context.Context
	SendMsg(m any) error
	RecvMsg(m any) error
}

2.tcp 连接问题，参考上面的 demo 即可

[Similarityoung]

双向流通信的核心机制与资源管理要点解析

一、双向流通信的核心机制
在HTTP/2协议下，双向流的本质是通过d.httpClient.Do(d.request)建立持久化连接实现的。该接口会创建底层TCP连接，同时维护请求/响应双工通道，其中：

1. 请求端管理：

• 通过requestBodyWriter *io.PipeWriter进行请求体写入
• CloseRequest()方法的核心职责：
  • 关闭管道写入端：d.requestBodyWriter.Close()
  • 向服务端发送流结束信号
  • 触发服务端onCompleted回调（Java示例中的响应终止处理）

你可以在 java 的 server 端进行如下操作：

@Override
public void onCompleted() {
      responseObserver.onCompleted();
      System.out.println("biStream completed");
}

1. 响应端管理：

• 通过response.Body进行响应体读取
• CloseResponse()的本质操作：d.response.Body.Close()

二、客户端实现的最佳实践

我仿照 grpc 的 sample 对客户端进行了改写

func TestBiDiStream2(svc greet.GreetService) error {
    // 初始化流通道
    stream, err := svc.GreetStream(context.Background())
    if err != nil {
        return err
    }

    // 异步响应处理协程
    waitc := make(chan struct{})
    go func() {
        defer close(waitc)
        for {
            in, err := stream.Recv()
            if err != nil { // 捕获EOF或其他错误
                fmt.Printf("Recv terminal: %v\n", err)
                return
            }
            fmt.Printf("Response: %s\n", in.Greeting)
        }
    }()

    // 同步发送阶段
    for i := 0; i < 5; i++ {
        if err := stream.Send(&greet.GreetStreamRequest{
            Name: "stream client!"
        }); err != nil {
            return err
        }
    }

    // 优雅关闭流程
    stream.CloseRequest()  // 主动终止发送
    <-waitc               // 等待响应处理完成
    defer stream.CloseResponse() // 最终资源清理

    fmt.Println("Stream closed properly")
    return nil
}

三、资源管理的必要性

为什么需要stream.CloseResponse()，我在相关文章找到了如下解释

为什么需要response.Body.Close()

resp.Body.Close() 做了什么？

如果返回值res的主体未关闭，client 下层的 RoundTripper 接口（一般为 Transport 类型）可能无法重用 res 主体下层保持的 TCP 连接去执行之后的请求。所以它的作用就是用来确保body读干净，释放出该连接

为什么这样做？

连接复用

如果不这么做会发生什么？

第一则是：无法重新使用与服务器的持久 TCP 连接来进行后续的“保持活动”请求，在下次发起HTTP请求的时候，就会重新建立TCP连接

第二如果不关闭当前请求，readLoop 和 writeLoop 两个 goroutine 在 写入请求并获取 response 返回后，并没有跳出 for 循环，而继续阻塞在下一次 for 循环的 select 语句里面，goroutine 一直无法被回收，cpu 和 memory 全部打满。发生goroutine内存泄漏

第三如果请求完成后，对端关闭了连接（对端的HTTP服务器向我发送了FIN），如果这边不调用response.Body.Close()，那么可以看到与这个请求相关的TCP连接的状态一直处于CLOSE_WAIT状态，态，不会被系统回收，则文件描述符不会被释放，出现资源泄漏。

关于客户端管理与TCP连接复用的澄清说明

一、命名误解与技术本质

1. 客户端创建的核心逻辑
   NewGreetService 的命名虽可能引起误解，但其技术本质是创建客户端实例而非直接建立TCP连接。通过源码分析可见：

   svc, err := greet.NewGreetService(cli)  
   // 实际调用链：  
   // => newClientManager(url)  
   //    => 初始化 transport (http.RoundTripper)  

   每个客户端实例内部维护独立的连接池（Transport），负责管理底层TCP连接的复用。

2. 与gRPC设计的对标性
   gRPC的经典实现方式与当前方案高度一致：

   conn := grpc.NewClient(addr)          // 物理连接管理  
   client := pb.NewRouteGuideClient(conn) // 逻辑客户端  
   runRouteChat(client)                  // 复用连接  

   二者的核心差异仅体现在API抽象层级，而非连接管理机制。

二、连接复用的实现机制

1. Transport 的核心作用

   • 作为 http.RoundTripper 接口实现，管理HTTP/2连接池
   • 自动复用空闲TCP连接，减少三次握手开销
   • 通过 MaxIdleConns 等参数控制连接池行为

2. 客户端生命周期管理

   // 正确用法：单例客户端复用  
   client := NewGreetService()  
   for i := 0; i < 10; i++ {  
       runBiDiStream(client) // 复用同一Transport  
   }  

   关键准则：避免重复创建客户端实例，防止产生冗余连接池。