Skip to content

Discussion 2756

Jump to bottom
alanxtl edited this page Sep 20, 2026 · 1 revision

双向流 BidiStreamForClient 的问题

讨论目录 · GitHub 原讨论

  • 编号：#2756
  • 分类：Q&A
  • 状态：已关闭 / 有已采纳答案
  • 创建：2024-12-03T10:53:29Z
  • 更新：2026-03-16T12:34:25Z
  • 作者：zhangcunli

社区原始记录；其中的建议、未决提案和版本描述需结合 版本与升级 核对。快照日期：2026-09-20。

已采纳答案

答案原文 · 作者：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  
}

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

讨论正文

  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 连接似乎并没有被复用

评论与回复

评论 1 · No-SilverBullet

永久链接 · 2024-12-13T03:22:22Z

回复 1.1 · zhangcunli

永久链接 · 2024-12-17T02:55:19Z

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

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

评论 2 · AlexStocks

永久链接 · 2024-12-22T12:13:22Z

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

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

回复 2.1 · AlexStocks

永久链接 · 2024-12-22T12:17:58Z

https://github.com/apache/dubbo-go-pixiu/issues/645

评论 3 · zhangcunli

永久链接 · 2024-12-25T07:48:15Z

写了个最小验证 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 池吗？这样用起来有些麻烦。

回复 3.1 · AlexStocks

永久链接 · 2024-12-29T12:24:14Z

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

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

回复 3.2 · zhangcunli

永久链接 · 2024-12-29T12:46:02Z

@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 即可

回复 3.3 · Similarityoung

永久链接 · 2025-02-15T10:29:09Z

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

一、双向流通信的核心机制 在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  
}

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

首页 · 专题导读 · 版本与升级 · 来源 · 历史档案

资料快照：2026-09-20

Apache Dubbo-go

完整资料库

维护

Clone this wiki locally