Go微服务中Kafka连接数异常：如何记录连接的建立与关闭？

问题背景 #

我使用github.com/segmentio/kafka-go库与Kafka Broker建立连接，订阅者和生产者的配置如下：

订阅者配置 #

c := kafka_go.ReaderConfig{
    Brokers:     []string{BootstrapEndpoint},
    GroupID:     GroupID,
    Topic:       Topic,
    MaxAttempts: MaxAttempts,
    MinBytes:    10e3, //10KB
    MaxBytes:    10e6, //10MB
}

c.Dialer = &kafka_go.Dialer{
    Timeout:       10 * time.Second,
    DualStack:     true,
    SASLMechanism: plain.Mechanism{Username: Username, Password: Password},
    TLS:           &tls.Config{},
}

r := kafka_go.NewReader(c)

// 启动订阅者
func Start() {
    go func() {
        defer s.reader.Close()

        for {
            metadata := map[string]interface{}{}
            msg, err := s.reader.FetchMessage(s.ctx)

            // 业务处理逻辑
            // ...

            s.reader.CommitMessages(ctx, msg)
        }
    }()
}

生产者配置 #

transport := &kafka_go.Transport{
    TLS: &tls.Config{},
    SASL: plain.Mechanism{
        Username: "username",
        Password: Password,
    },
}

kafkaWriter := &kafka_go.Writer{
    Addr:         kafka_go.TCP(Url),
    Topic:        Topic,
    Balancer:     &kafka_go.RoundRobin{},
    Async:        false,
    BatchTimeout: time.Second,
    MaxAttempts:  5,
    Transport:    transport,
}

kafkaWriter.WriteMessages(
    ctx, kafka_go.Message{Key: []byte(id), Value: bytesData},
)

实际运行中发现打开的TCP连接数远高于预期（预期订阅者和生产者各仅打开1个TCP连接），请问有没有办法在Go微服务中记录连接的打开与关闭操作？

解决方案 #

1. 包装kafka-go的Dialer，追踪连接全生命周期 #

kafka-go的Dialer负责创建新连接，我们可以包装它的DialContext方法，同时包装返回的net.Conn来记录关闭事件：

type LoggingDialer struct {
    *kafka_go.Dialer
}

func (d *LoggingDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
    log.Printf("[Kafka连接] 正在创建: %s://%s", network, address)
    conn, err := d.Dialer.DialContext(ctx, network, address)
    if err != nil {
        log.Printf("[Kafka连接] 创建失败: %s://%s, 错误: %v", network, address, err)
        return conn, err
    }
    // 包装连接以记录关闭动作
    wrappedConn := &LoggingConn{Conn: conn, addr: address}
    log.Printf("[Kafka连接] 创建成功: %s://%s, 连接标识: %p", network, address, wrappedConn)
    return wrappedConn, nil
}

type LoggingConn struct {
    net.Conn
    addr string
}

func (c *LoggingConn) Close() error {
    err := c.Conn.Close()
    log.Printf("[Kafka连接] 已关闭: %s, 连接标识: %p", c.addr, c)
    return err
}

使用方式：替换原Dialer实例

c.Dialer = &LoggingDialer{
    Dialer: &kafka_go.Dialer{
        Timeout:       10 * time.Second,
        DualStack:     true,
        SASLMechanism: plain.Mechanism{Username: Username, Password: Password},
        TLS:           &tls.Config{},
    },
}

2. 自定义Transport监控生产者连接 #

对于生产者的Transport，可以包装其RoundTrip方法，追踪请求对应的连接使用情况：

type LoggingTransport struct {
    *kafka_go.Transport
}

func (t *LoggingTransport) RoundTrip(req *kafka_go.Request) (*kafka_go.Response, error) {
    log.Printf("[Kafka请求] 发送到Broker: %s, 请求类型: %s", req.Addr, req.ApiKey)
    resp, err := t.Transport.RoundTrip(req)
    if err != nil {
        log.Printf("[Kafka请求] 失败: %s, 错误: %v", req.Addr, err)
    }
    return resp, err
}

同时给Transport的Dialer配置上面的LoggingDialer，确保生产者的连接创建/关闭都被记录。

3. 用标准工具辅助排查 #

• 实时查看连接状态：用netstat或ss命令过滤进程的活跃连接：

  netstat -anp | grep <你的进程PID> | grep ESTABLISHED
  

• 启用pprof分析连接：导入net/http/pprof暴露Profile数据，查看活跃连接详情：

  import _ "net/http/pprof"
  
  // 启动pprof服务
  go func() {
      log.Println(http.ListenAndServe("localhost:6060", nil))
  }()
  

  访问http://localhost:6060/debug/pprof/netconn可查看当前所有网络连接的状态。

4. 排查连接数超预期的根源 #

kafka-go的Reader默认会为每个分配到的分区创建单独连接，生产者也会根据Broker数量、负载策略创建多个连接，这可能是你连接数超预期的核心原因。通过上面的日志可以明确连接是因分区分配、重连还是其他逻辑创建的。

内容的提问来源于stack exchange，提问作者Ron Weasley