Go语言Kafka消费者结合缓冲Channel写入Redis管道问题排查

我来帮你拆解下你遇到的两个场景里的问题，都是Go并发编程中channel使用的典型坑：

第一个版本代码的问题 #

你最初的代码里，在处理完Kafka消息后做了这两步：

redisChnl <- e.Value 
<- redisChnl 

这两个操作是同步执行的——刚把消息写入channel，立刻就把它读出来了，相当于channel根本没起到缓冲的作用，所以len(redisChnl)永远是0，因为写入的元素马上就被移除了。

另外，程序冻结的原因是：当Kafka的Poll(100)没有返回消息时，代码会走到<- redisChnl这一步，而此时channel是空的，无缓冲或缓冲为空的channel读取操作会阻塞当前goroutine，所以程序就卡在这一步不动了。

第二个版本代码的问题 #

你改成用goroutine写入channel：

go func() {
    redisChnl <- e.Value
}()

然后发现读取前后channel长度都是2000，原因有两个：

1. 写入速度远大于消费速度：Kafka消费消息的速度比你处理Redis写入的速度快，你刚从channel里读走一个元素，立刻就有新的消息被goroutine写入channel，所以看起来长度没变化。
2. 消费逻辑混乱：你没有专门的goroutine来持续消费channel里的消息，而是零散地在default分支读取，这种方式无法高效处理批量消息，也没法跟上写入的速度。

正确的解决方案 #

我们需要把Kafka消费、消息缓冲、Redis批量写入拆分成独立的并发单元，用channel做通信，同时处理好背压和优雅退出：

修正后的核心代码 #

func main() {
    sigchan := make(chan os.Signal, 1)
    signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

    // 初始化Kafka消费者（你的原有代码不变）
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":        "kafka.com:9093",
        "group.id":                 "testehb",
        "security.protocol":        "ssl",
        "ssl.key.location":         "/Users/key.key",
        "ssl.certificate.location": "/Users/cert.cert",
        "ssl.ca.location":          "/Users/ca.pem",
    })
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
        os.Exit(1)
    }
    defer c.Close()
    fmt.Printf("Created Consumer %v\n", c)
    err = c.SubscribeTopics([]string{"jmx"}, nil)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to subscribe: %s\n", err)
        os.Exit(1)
    }

    redisChnl := make(chan []byte, 2000)
    batchSize := 100 // 批量写入Redis的消息数量

    // 启动专门的Redis消费goroutine，负责批量写入
    go func() {
        pipe := client.Pipeline()
        msgCount := 0
        for msg := range redisChnl {
            // 解析消息生成Redis的key和value
            res := MessageFormat{}
            if unmarshalErr := json.Unmarshal(msg, &res); unmarshalErr != nil {
                fmt.Printf("Unmarshal failed: %v\n", unmarshalErr)
                continue
            }
            host := regexp.MustCompile(`^([^.]+)`).FindString(res.MetricPath)
            pipe.Set(host, string(msg), time.Hour)
            msgCount++

            // 达到批量大小，执行Redis管道
            if msgCount >= batchSize {
                if _, execErr := pipe.Exec(); execErr != nil {
                    fmt.Printf("Pipeline exec failed: %v\n", execErr)
                }
                // 重置管道和计数
                pipe = client.Pipeline()
                msgCount = 0
            }
        }

        // 程序退出前，处理剩余的未批量消息
        if msgCount > 0 {
            if _, execErr := pipe.Exec(); execErr != nil {
                fmt.Printf("Final pipeline exec failed: %v\n", execErr)
            }
        }
    }()

    run := true
    for run {
        select {
        case sig := <-sigchan:
            fmt.Printf("Caught signal %v: terminating\n", sig)
            run = false
            close(redisChnl) // 关闭channel，通知消费goroutine退出
        case ev := <-c.PollChan(): // 用PollChan替代Poll，适配select模型
            switch e := ev.(type) {
            case *kafka.Message:
                // 直接写入缓冲channel，满了会自动阻塞，实现背压
                redisChnl <- e.Value
            case kafka.PartitionEOF:
                fmt.Printf("%% Reached %v\n", e)
            case kafka.Error:
                fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
                run = false
                close(redisChnl)
            default:
                fmt.Printf("Ignored %v\n", e)
            }
        }
    }

    fmt.Printf("Closing consumer\n")
}

关键优化点 #

1. 分离消费和写入逻辑：用专门的goroutine处理Redis批量写入，不会阻塞Kafka消费流程。
2. 用PollChan替代Poll：把Kafka消息事件放到select的case里，避免在default分支里循环阻塞。
3. 自动背压：当缓冲channel满了时，redisChnl <- e.Value会自动阻塞Kafka消费，防止消息堆积过多。
4. 优雅退出：收到终止信号时关闭channel，确保消费goroutine处理完剩余消息再退出。

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