Golang Kafka消费者无消息时CPU占用100%问题求助

问题描述 #

基于confluent-kafka-go实现了带Worker Pool的Kafka Consumer，启动后无消息流入时容器CPU占用接近100%，无法达到预期的处理性能。核心代码如下：

Worker及Worker Pool定义 #

type Worker struct {
    consumer *Consumer
    events   chan *kafka.Message
}

func (w *Worker) start() {
    for event := range w.events {
        w.consumer.handledEventMap[callBackKey] += 1
        w.consumer.HandleMessage(event)
    }
}

type WorkerPool struct {
    workers []*Worker
    queue   chan *kafka.Message
}

func NewWorkerPool(consumer *Consumer, numWorkers int) *WorkerPool {
    pool := &WorkerPool{
        workers: make([]*Worker, numWorkers),
        queue:   make(chan *kafka.Message),
    }

    for i := 0; i < numWorkers; i++ {
        pool.workers[i] = &Worker{
            consumer: consumer,
            events:   pool.queue,
        }
        go pool.workers[i].start()
    }

    return pool
}

func (pool *WorkerPool) Stop() {
    close(pool.queue)
}

消费方法实现 #

func (c *Consumer) Consume() {
    numWorkers := 10
    pool := NewWorkerPool(c, numWorkers)
    defer pool.Stop()
    ticker := time.NewTicker(time.Second * 60)

    for {
        select {
        case event := <-c.consumer.Events():
            switch e := event.(type) {
            case *kafka.Message:
                pool.queue <- e // Add the event to the worker pool's queue
            case kafka.Error:
                logrus.WithError(e).Warning("an error occurred while reading from topic")
            }
        case <-ticker.C:
            logrus.Infof("Consumer %s received events in past 60 seconds: %v", c.consumerName, c.handledEventMap)
            c.handledEventMap = make(map[string]int)
        }
    }
}

问题根源 #

1. 未处理非消息/错误事件导致CPU空转 #

confluent-kafka-go的Consumer会持续发送多种事件，包括kafka.PartitionEOF（分区末尾事件）。当没有新消息时，消费者会不断推送PartitionEOF事件，而代码中未处理这类事件，导致主循环疯狂执行空操作，直接拉满CPU。

2. 无缓冲任务队列限制并发性能 #

pool.queue是无缓冲channel，当Worker处理速度跟不上消息流入速度时，主消费线程会阻塞在pool.queue <- e，无法继续读取Kafka事件，完全抵消了Worker Pool的并发优势。

3. 并发不安全的事件统计 #

多个Worker Goroutine同时修改c.handledEventMap，没有任何同步机制，会导致统计数据错误，甚至出现panic。

4. 不完善的Worker Pool停止逻辑 #

仅关闭任务队列无法等待所有Worker完成当前任务，可能导致部分消息未处理就退出。

修复方案 #

1. 处理所有事件类型，避免CPU空转 #

在事件处理分支中添加默认case，忽略非关键事件：

switch e := event.(type) {
case *kafka.Message:
    pool.queue <- e
case kafka.Error:
    logrus.WithError(e).Warning("an error occurred while reading from topic")
case kafka.PartitionEOF:
    // 可选：记录分区末尾日志，避免空转
    logrus.Debugf("reached end of partition %v", e)
default:
    // 忽略其他无关事件
}

2. 给任务队列添加缓冲 #

根据预期消息吞吐量设置合适的缓冲大小，避免主线程阻塞：

func NewWorkerPool(consumer *Consumer, numWorkers int, queueSize int) *WorkerPool {
    pool := &WorkerPool{
        workers: make([]*Worker, numWorkers),
        queue:   make(chan *kafka.Message, queueSize), // 设置缓冲
    }
    // ... 其余代码不变
}

// 在Consume中初始化时指定缓冲大小，比如1000
pool := NewWorkerPool(c, numWorkers, 1000)

3. 修复并发安全问题 #

给handledEventMap添加互斥锁：

type Consumer struct {
    // ... 原有字段
    handledEventMap map[string]int
    mapMutex        sync.Mutex
}

// 在Worker中修改统计时加锁
func (w *Worker) start() {
    for event := range w.events {
        w.consumer.mapMutex.Lock()
        w.consumer.handledEventMap[callBackKey] += 1
        w.consumer.mapMutex.Unlock()
        w.consumer.HandleMessage(event)
    }
}

4. 完善Worker Pool停止逻辑 #

添加等待组，确保所有Worker完成当前任务后再退出：

type WorkerPool struct {
    workers []*Worker
    queue   chan *kafka.Message
    wg      sync.WaitGroup
}

func NewWorkerPool(consumer *Consumer, numWorkers int, queueSize int) *WorkerPool {
    pool := &WorkerPool{
        workers: make([]*Worker, numWorkers),
        queue:   make(chan *kafka.Message, queueSize),
    }

    pool.wg.Add(numWorkers)
    for i := 0; i < numWorkers; i++ {
        pool.workers[i] = &Worker{
            consumer: consumer,
            events:   pool.queue,
        }
        go func() {
            defer pool.wg.Done()
            pool.workers[i].start()
        }()
    }

    return pool
}

func (pool *WorkerPool) Stop() {
    close(pool.queue)
    pool.wg.Wait() // 等待所有Worker完成
}

内容的提问来源于stack exchange，提问作者Omri. B