同一GroupId下多消费者多反序列化器无法匹配正确监听器问题

在同一个Kafka主题topicOne下，配置了两个针对不同事件类型的消费者，各自对应专属反序列化器。接收消息时，反序列化器无法匹配正确的监听器，导致消息无法被消费；重试几次后才会匹配到正确监听器，但会丢失大量消息。当两个监听器属于不同消费组时，功能完全正常。想了解Kafka是如何选择反序列化器并消费消息的。

kafka:
    "FirstEventType":
      "[bootstrap.servers]": ${KAFKA_CONSUMER_BROKER_LIST:localhost:9092}
      "[group.id]": GroupOne
      "[client.id]": firstEventType
      "[auto.offset.reset]": earliest
      "[key.deserializer]": org.apache.kafka.common.serialization.StringDeserializer
      "[value.deserializer]":org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      "[spring.deserializer.value.delegate.class]": com.my.pack.svc.kafka.deserializers.FirstEventTypeDeserializer
      "[messageType]": FirstEventType
      topic: topicOne
    "SecondEventType":
      "[bootstrap.servers]": ${KAFKA_CONSUMER_BROKER_LIST:localhost:9092}
      "[group.id]": GroupOne
      "[client.id]": secondEventType
      "[auto.offset.reset]": earliest
      "[key.deserializer]": org.apache.kafka.common.serialization.StringDeserializer
      "[value.deserializer]": org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      "[spring.deserializer.value.delegate.class]": com.ft.subs.b2cfulfilsvc.kafka.deserializers.SecondEventTypeDeserializer
      "[messageType]": SecondEventType
      topic: topicOne

第一个事件监听器 #

@KafkaListener(
      topics = "topicOne",
      containerFactory = "FirstEventTypeFactory",
      groupId = "GroupOne")
  public void onFirstEventType(
      FirstEventType eventType) {
   // 业务逻辑处理
  }

第二个事件监听器 #

@KafkaListener(
      topics = "topicOne",
      containerFactory = "SecondEventTypeFactory",
      groupId = "GroupOne")
  public void onSecondEventType(
      SecondEventType eventType) {
   // 业务逻辑处理
  }

第一个事件反序列化器 #

@Component
public class FirstEventTypeDeserializer
    extends AbstractMessageDeserializer<FirstEventType> {}

第二个事件反序列化器 #

@Component
public class SecondEventTypeDeserializer
    extends AbstractMessageDeserializer<SecondEventType> {}

抽象反序列化器基类 #

import org.apache.kafka.common.serialization.Deserializer;
// 其他导入包

public abstract class AbstractMessageDeserializer<T> implements Deserializer<T> {
    // 通用反序列化逻辑
}

1. Kafka消费组核心机制 #

Kafka消费组（group.id）是实现消息负载均衡的核心：同一个消费组内的多个消费者，会共同订阅主题的分区，每个分区只能被组内一个消费者消费。当两个监听器设置同一个group.id=GroupOne时，它们属于同一消费组，Kafka会将topicOne的分区分配给其中一个消费者，另一个消费者处于空闲状态。

这就导致：消息发送到某个分区时，只有被分配到该分区的消费者会拉取消息。如果该消费者的反序列化器不匹配消息类型，就会反序列化失败触发重试；只有当分区重新分配（如消费者重启、心跳超时）时，另一个消费者才可能拿到分区，此时才能正确反序列化。这就是“重试几次后匹配正确监听器”的原因，重试过程中失败的消息可能被丢弃或进入死信队列，最终导致消息丢失。

当两个监听器属于不同消费组时，每个消费组会独立订阅主题的所有分区，两个消费者各自拉取全部分区的消息，用自身反序列化器处理对应类型消息，自然无冲突。

2. Spring Kafka反序列化器绑定逻辑 #

在Spring Kafka中，@KafkaListener通过指定的containerFactory绑定对应消费者配置（含反序列化器），每个containerFactory对应一套独立的消费者配置，包括专属反序列化器。但如果两个容器同属一个消费组，它们会参与同一分区分配逻辑，而非各自独立消费。

你的场景中，两个监听器虽用了不同containerFactory，但因同属一个消费组，Kafka只会把分区分配给其中一个容器，另一个容器无法拿到消息。只有分区重新分配时，另一个容器才可能获得分区，此时才能用正确的反序列化器处理消息。

3. 解决方案 #

• 使用不同消费组：最直接的方案，让两个监听器使用不同group.id，每个消费组独立消费主题所有消息，各自用对应反序列化器处理消息，避免分区分配冲突。
• 单一消费者+类型路由：若必须用同一个消费组，可在消费者端统一反序列化（如先反序列化为JSON字符串），再在监听器中根据消息类型路由到不同处理方法，示例：

  @KafkaListener(topics = "topicOne", groupId = "GroupOne")
  public void onMessage(ConsumerRecord<String, String> record) {
      String message = record.value();
      // 根据消息中的type字段判断事件类型
      if (isFirstEventType(message)) {
          FirstEventType event = objectMapper.readValue(message, FirstEventType.class);
          handleFirstEventType(event);
      } else if (isSecondEventType(message)) {
          SecondEventType event = objectMapper.readValue(message, SecondEventType.class);
          handleSecondEventType(event);
      }
  }
  

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