Goroutine中Kafka重连实现及Go语言Kafka库选型咨询

我希望使用Golang向Kafka写入消息，现有如下代码：

package kafkaK

import (
    "context"
    "github.com/segmentio/kafka-go"
    "github.com/segmentio/kafka-go/snappy"
    "time"
)

var writer *kafka.Writer

func Configure() (w *kafka.Writer, err error) {
    dialer := &kafka.Dialer{
        Timeout:  10 * time.Second,
        ClientID: "123",
    }

    config := kafka.WriterConfig{
        Brokers:          []string{"localhost:9092"},
        Topic:            "test",
        Balancer:         &kafka.LeastBytes{},
        Dialer:           dialer,
        WriteTimeout:     10 * time.Second,
        ReadTimeout:      10 * time.Second,
        CompressionCodec: snappy.NewCompressionCodec(),
    }
    w = kafka.NewWriter(config)
    writer = w
    return w, nil
}

func Push(parent context.Context, key, value []byte) (err error) {
    message := kafka.Message{
        Key:   key,
        Value: value,
        Time:  time.Now(),
    }
    return writer.WriteMessages(parent, message)
}

我通过独立Goroutine向Kafka写入消息，相关代码如下：

func sendMessage(message string) {
    err := kafkaK.Push(context.Background(), nil, []byte(message))
    if err != nil {
        fmt.Println(err)
    }
}

调用方式为：

go sendMessage("message #" + strconv.Itoa(i))

现存在两个问题：

1. 当Kafka不可用时，如何在多Goroutine共用同一Kafka对象的场景下正确实现重连？或许有其他可行方案？
2. 操作Kafka更适合使用哪个Go语言库？

我考虑过使用channel或context实现重连，但作为Go语言新手，暂不清楚具体实现方式。

问题1：多Goroutine下Kafka不可用时的重连方案 #

segmentio/kafka-go的Writer本身有基础重连逻辑，但遇到长时间故障时需要主动干预。由于多Goroutine共用全局实例，必须保证并发安全，以下是具体实现思路：

1. 加互斥锁保护Writer实例 #

在kafkaK包中添加读写互斥锁，避免并发修改冲突：

package kafkaK

import (
    "context"
    "net"
    "sync"
    "github.com/segmentio/kafka-go"
    "github.com/segmentio/kafka-go/snappy"
    "time"
)

var (
    writer *kafka.Writer
    mu     sync.RWMutex // 读多写少场景用读写锁更高效
)

// 原Configure函数逻辑不变

2. 改造Push函数，加入错误判断与重连逻辑 #

写入失败时先判断是否为连接类错误，若是则尝试重建Writer并重试：

func Push(parent context.Context, key, value []byte) error {
    message := kafka.Message{
        Key:   key,
        Value: value,
        Time:  time.Now(),
    }

    // 读锁获取当前writer
    mu.RLock()
    currentWriter := writer
    mu.RUnlock()

    // 首次尝试写入
    err := currentWriter.WriteMessages(parent, message)
    if err == nil {
        return nil
    }

    // 判断是否为连接相关错误
    if isConnectionError(err) {
        // 写锁保护，避免多个Goroutine重复重建
        mu.Lock()
        // 二次检查，防止其他Goroutine已完成重建
        if writer == currentWriter {
            newWriter, confErr := Configure()
            if confErr != nil {
                mu.Unlock()
                return confErr
            }
            writer = newWriter
        }
        mu.Unlock()

        // 用新writer重试一次
        mu.RLock()
        newWriter := writer
        mu.RUnlock()
        return newWriter.WriteMessages(parent, message)
    }

    return err
}

// 识别连接类错误
func isConnectionError(err error) bool {
    // 匹配Kafka内置的连接相关错误码
    if err == kafka.ErrLeaderNotAvailable || err == kafka.ErrBrokerNotAvailable {
        return true
    }
    // 匹配网络超时、拨号失败等错误
    if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
        return true
    }
    // 可根据实际遇到的错误补充更多判断
    return false
}

3. 可选优化：消息缓存与重试机制 #

如果Kafka长时间不可用，可引入本地消息队列暂存失败消息，启动单独Goroutine重试：

var msgQueue = make(chan kafka.Message, 1000) // 根据业务调整缓冲大小

// 程序初始化时调用，启动重试协程
func StartRetryWorker() {
    go func() {
        for msg := range msgQueue {
            err := Push(context.Background(), msg.Key, msg.Value)
            if err != nil {
                // 指数退避后重新入队，避免频繁重试
                time.Sleep(2 * time.Second)
                msgQueue <- msg
            }
        }
    }()
}

// 修改Push函数，失败时将消息放入缓存队列
func Push(parent context.Context, key, value []byte) error {
    // ... 原有写入逻辑 ...
    if err != nil {
        if isConnectionError(err) {
            select {
            case msgQueue <- message:
                return nil // 暂存成功，后续重试
            default:
                return err // 队列满，返回错误
            }
        }
        return err
    }
    return nil
}

问题2：推荐的Go Kafka库 #

目前Go生态中主流的Kafka客户端库有三个，可根据业务场景选择：

1. segmentio/kafka-go #

• 优势：纯Go实现，API简洁易懂，上手成本低；内置压缩、基础重连功能；支持并发写入，适合简单业务。
• 劣势：高级特性（事务、精确一次语义）支持不足；社区活跃度一般。
• 适用场景：新手入门、轻量业务、不需要复杂Kafka特性的场景。

2. Shopify/sarama #

• 优势：纯Go实现，功能全面，支持几乎所有Kafka特性（事务、分区管理、自定义序列化等）；社区活跃，文档完善；大厂生产环境广泛使用。
• 劣势：API相对复杂，学习曲线稍陡。
• 适用场景：复杂生产业务、需要定制化Kafka操作的场景。

3. confluentinc/confluent-kafka-go #

• 优势：基于官方C库librdkafka开发，性能优异、稳定性极高；支持最完整的Kafka高级特性；官方维护，兼容性最好。
• 劣势：依赖C库，安装和跨平台部署相对麻烦；API风格偏向C语言。
• 适用场景：对性能和稳定性要求极高的核心业务、需要用到Kafka高级特性的场景。

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