Golang中Kafka客户端发布消息的实现与优化

这个错误通常是由于Kafka主题分区缺少leader导致的，可能是由于broker故障或网络问题。以下是几种解决方案和优化建议：

1. 增加重试机制和超时配置

p.Writer = &kafka.Writer{
    Addr:     kafka.TCP(addrList...),
    Topic:    topic,
    Balancer: &kafka.LeastBytes{},
    // 增加重试次数
    MaxAttempts: 5,
    // 设置超时时间
    WriteTimeout: 10 * time.Second,
    ReadTimeout:  10 * time.Second,
    // 启用异步写入
    Async: true,
    // 设置批次大小
    BatchSize: 100,
    BatchTimeout: 1 * time.Second,
}

2. 添加错误处理和重试逻辑

func (p *Producer) SendMessage(ctx context.Context, key, value []byte) error {
    message := kafka.Message{
        Key:   key,
        Value: value,
        Time:  time.Now(),
    }
    
    var lastErr error
    for attempt := 1; attempt <= 3; attempt++ {
        err := p.Writer.WriteMessages(ctx, message)
        if err == nil {
            return nil
        }
        
        lastErr = err
        if errors.Is(err, kafka.LeaderNotAvailable) || 
           strings.Contains(err.Error(), "no leader") {
            time.Sleep(time.Duration(attempt) * time.Second)
            continue
        }
        break
    }
    return fmt.Errorf("failed after 3 attempts: %w", lastErr)
}

3. 使用更健壮的连接配置

p.Writer = &kafka.Writer{
    Addr:     kafka.TCP(addrList...),
    Topic:    topic,
    Balancer: &kafka.LeastBytes{},
    // 设置必要的传输层配置
    Transport: &kafka.Transport{
        DialTimeout: 5 * time.Second,
        IdleTimeout: 30 * time.Second,
        TLS:         nil, // 如果需要TLS则配置
        SASL:        nil, // 如果需要SASL认证则配置
    },
    // 设置连接数
    RequiredAcks: kafka.RequireAll,
    // 启用压缩
    Compression: kafka.Snappy,
}

4. 实现健康检查和自动恢复

type KafkaProducer struct {
    writer *kafka.Writer
    mu     sync.RWMutex
}

func (kp *KafkaProducer) HealthCheck() bool {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    // 尝试获取主题元数据
    conn, err := kafka.Dial("tcp", kp.writer.Addr.String())
    if err != nil {
        return false
    }
    defer conn.Close()
    
    partitions, err := conn.ReadPartitions(kp.writer.Topic)
    if err != nil {
        return false
    }
    
    // 检查所有分区是否有leader
    for _, p := range partitions {
        if p.Leader.ID == -1 {
            return false
        }
    }
    return true
}

func (kp *KafkaProducer) Reconnect() error {
    kp.mu.Lock()
    defer kp.mu.Unlock()
    
    if kp.writer != nil {
        kp.writer.Close()
    }
    
    // 重新创建writer
    kp.writer = &kafka.Writer{
        Addr:         kafka.TCP("localhost:9092"),
        Topic:        "your-topic",
        Balancer:     &kafka.LeastBytes{},
        MaxAttempts:  5,
        WriteTimeout: 10 * time.Second,
    }
    return nil
}

5. 监控和日志记录

type MonitoredWriter struct {
    writer *kafka.Writer
    stats  *WriterStats
}

type WriterStats struct {
    TotalMessages    int64
    FailedMessages   int64
    RetryCount       int64
    LastError        error
    LastErrorTime    time.Time
}

func (mw *MonitoredWriter) WriteMessages(ctx context.Context, msgs ...kafka.Message) error {
    start := time.Now()
    err := mw.writer.WriteMessages(ctx, msgs...)
    
    atomic.AddInt64(&mw.stats.TotalMessages, int64(len(msgs)))
    if err != nil {
        atomic.AddInt64(&mw.stats.FailedMessages, int64(len(msgs)))
        mw.stats.LastError = err
        mw.stats.LastErrorTime = time.Now()
        
        // 记录详细错误日志
        log.Printf("Kafka write failed: %v, duration: %v", err, time.Since(start))
    }
    return err
}

这些优化措施可以帮助你更好地处理Kafka连接问题，提高消息发布的可靠性。