为每个WebSocket客户端创建Kafka Consumer是否合理？遇错求解决方案

问题背景 #

我运行一个实时数据流的 WebSocket 服务器，用 Kafka 做流处理。目前的实现是给每个客户端创建独立的 Kafka Consumer，但当客户端数量超过 5000 时，出现以下错误：

%2|1683792545.811|THREAD|rdkafka#consumer-13806| [thrd:main]: Unable to create broker thread

当前每个客户端创建 Consumer 的代码如下：

func(c *Client) SendDataTOConsumer() {  
    defer func() {
        if r := recover(); r != nil {
            log.Println("PANIC:", r)
        }
    }()
    
    run := true
    config := &kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092", // broker地址
        "group.id": c.Id, // 每个客户端用自己的ID作为消费组
        "auto.offset.reset": "earliest", // 从最早偏移量开始消费
    }

    // 创建Kafka Consumer
    consumer, err := kafka.NewConsumer(config)
    if err != nil {
        log.Println(err)
    }

    defer func ()  {
        consumer.Close()
    }()

    var topic []string
    for key := range c.SubsTopics {
        topic = append(topic, key)
    }

    if err := consumer.SubscribeTopics(topic, nil); err != nil{
        log.Print("订阅主题错误: " ,err)
    }

    // 处理消息
    for run {
        select {
        case unSubscribe := <- c.UnsubChan:
            run = false
            if unSubscribe {
                if err := consumer.Unsubscribe(); err != nil{
                    log.Println("取消订阅错误",err)
                }
            }
        
        default:
            message, err := consumer.ReadMessage(100 * time.Millisecond)
            if err != nil {
                // 错误为信息性，由Consumer自动处理
                continue
            }

            if err := c.Conn.WriteMessage(websocket.TextMessage, message.Value); err != nil {
                // 如果写消息出错，关闭连接、Consumer并停止循环
                if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, websocket.CloseNormalClosure){
                    break
                }
                run = false
            }
        }
    }
}

错误原因分析 #

每个 Kafka Consumer 实例都是重量级资源：它会创建多个后台线程（比如网络 IO 线程、心跳线程、分区协调线程等）。当你创建 5000 个 Consumer 时，系统线程数会急剧膨胀，超出操作系统的线程限制，最终导致无法创建新的 broker 线程（即错误提示的内容）。

另外，每个客户端用独立的 group.id 会导致同一消息被多个消费组重复消费，浪费 Kafka 集群资源，也不符合 Kafka 消费组的设计初衷。

解决方案与最佳实践 #

核心原则：复用 Kafka Consumer 实例，避免每个客户端一个 Consumer #

Kafka Consumer 的设计目标是支持高吞吐量的消息消费，一个 Consumer 实例可以订阅多个主题、处理多个分区的消息，完全不需要为每个客户端单独创建。正确的做法是：

1. 使用少量 Consumer 实例（建议数量 ≤ 主题分区数）

   • 同一个消费组内的 Consumer 会自动分摊主题的分区，确保每个分区只被一个 Consumer 消费，避免重复消费。
   • 一般来说，Consumer 的数量不要超过主题的总分区数，超过的话多余的 Consumer 会处于空闲状态。

2. 维护客户端订阅映射表

   • 在 WebSocket 服务端维护一个全局的映射结构，记录每个客户端订阅的主题，比如：map[string][]string（key 为客户端 ID，value 为订阅的主题列表），或者反过来 map[string][]*Client（key 为主题，value 为订阅该主题的客户端列表）。

3. 消息路由：Consumer 消费后推送给对应客户端

   • Consumer 消费到消息后，根据消息的主题，查找订阅该主题的所有客户端，然后将消息通过 WebSocket 推送给这些客户端。

优化后的代码示例思路 #

// 全局订阅映射：主题 -> 订阅该主题的客户端列表
var topicSubscribers = sync.Map{} // 用sync.Map保证并发安全

// 全局Consumer组，启动固定数量的Consumer实例
func StartKafkaConsumers(topicList []string, consumerCount int) {
    config := &kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "group.id": "websocket-stream-group", // 统一消费组ID
        "auto.offset.reset": "earliest",
    }

    for i := 0; i < consumerCount; i++ {
        go func() {
            consumer, err := kafka.NewConsumer(config)
            if err != nil {
                log.Fatal(err)
            }
            defer consumer.Close()

            if err := consumer.SubscribeTopics(topicList, nil); err != nil {
                log.Fatal("订阅主题错误:", err)
            }

            for {
                msg, err := consumer.ReadMessage(time.Second)
                if err != nil {
                    continue
                }

                // 根据消息主题获取订阅的客户端
                if clients, ok := topicSubscribers.Load(string(msg.Topic)); ok {
                    for _, client := range clients.([]*Client) {
                        // 推送消息给客户端，注意处理并发和连接关闭的情况
                        select {
                        case client.SendChan <- msg.Value:
                        default:
                            // 客户端通道满，可能是连接异常，移除该客户端
                            removeClientFromTopic(client, string(msg.Topic))
                        }
                    }
                }
            }
        }()
    }
}

// 客户端订阅主题时更新映射表
func (c *Client) SubscribeTopic(topic string) {
    clients, _ := topicSubscribers.LoadOrStore(topic, []*Client{})
    newClients := append(clients.([]*Client), c)
    topicSubscribers.Store(topic, newClients)
}

// 客户端取消订阅或断开连接时从映射表移除
func removeClientFromTopic(c *Client, topic string) {
    if clients, ok := topicSubscribers.Load(topic); ok {
        filtered := []*Client{}
        for _, client := range clients.([]*Client) {
            if client.Id != c.Id {
                filtered = append(filtered, client)
            }
        }
        if len(filtered) == 0 {
            topicSubscribers.Delete(topic)
        } else {
            topicSubscribers.Store(topic, filtered)
        }
    }
}

// 客户端的消息处理循环
func (c *Client) HandleWebSocket() {
    defer c.Conn.Close()
    for {
        select {
        case msg := <-c.SendChan:
            if err := c.Conn.WriteMessage(websocket.TextMessage, msg); err != nil {
                // 处理连接错误，移除客户端所有订阅
                c.UnsubscribeAll()
                return
            }
        case <-c.UnsubChan:
            c.UnsubscribeAll()
            return
        }
    }
}

额外注意事项 #

• 并发安全：全局订阅映射表需要用线程安全的结构（如sync.Map），避免并发读写冲突。
• 客户端清理：当客户端断开连接或取消订阅时，要及时从映射表中移除，避免内存泄漏和无效推送。
• 流量控制：如果客户端处理消息速度慢，要避免消息堆积，可以用带缓冲区的通道，或者在通道满时移除异常客户端。
• Consumer 异常处理：要监控 Consumer 的运行状态，当 Consumer 崩溃时自动重启，保证消费不中断。

内容的提问来源于stack exchange，提问作者Suriya Varman N