如何为Kafka不同消息类型配置Spring-Kafka独立监听器？

完全可以实现，核心是利用Kafka和Spring-Kafka自带的过滤机制，让不同监听器直接只接收对应类型的消息，无需先全量接收再过滤。以下是具体实现方案：

1. 基于@KafkaListener的SpEL表达式过滤 #

这是最简单的方式，直接在监听器注解中通过SpEL表达式匹配消息体的messageType字段：

@Component
public class TypeAListener {

    @KafkaListener(
            topics = "your-target-topic",
            groupId = "group-type-a",
            condition = "#p0.payload.messageType == 'A'"
    )
    public void handleTypeAMessage(Message<YourMessageEntity> message) {
        YourMessageEntity payload = message.getPayload();
        // 处理类型A的业务逻辑
    }
}

同理，创建类型B的监听器：

@Component
public class TypeBListener {

    @KafkaListener(
            topics = "your-target-topic",
            groupId = "group-type-b",
            condition = "#p0.payload.messageType == 'B'"
    )
    public void handleTypeBMessage(Message<YourMessageEntity> message) {
        YourMessageEntity payload = message.getPayload();
        // 处理类型B的业务逻辑
    }
}

关键说明 #

• YourMessageEntity是你定义的消息实体类，需与Kafka中JSON消息结构对应，同时要确保Spring-Kafka配置了JsonMessageConverter，能自动将JSON反序列化为实体对象。
• 每个监听器使用独立的groupId，避免消息分配混乱或重复消费。

2. 自定义消息过滤器（复杂场景适用） #

如果过滤逻辑涉及多条件判断，可以实现RecordFilterStrategy接口，将过滤逻辑封装为独立组件：

@Component
public class TypeARecordFilter implements RecordFilterStrategy<ConsumerRecord<String, YourMessageEntity>> {

    @Override
    public boolean filter(ConsumerRecord<String, YourMessageEntity> record) {
        // 返回true表示过滤该消息，false表示保留并交给监听器处理
        return !"A".equals(record.value().getMessageType());
    }
}

接着配置专属的监听器容器工厂：

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, YourMessageEntity> typeAFactory(
            ConsumerFactory<String, YourMessageEntity> consumerFactory,
            TypeARecordFilter typeARecordFilter) {
        ConcurrentKafkaListenerContainerFactory<String, YourMessageEntity> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setRecordFilterStrategy(typeARecordFilter);
        return factory;
    }
}

最后在监听器中指定使用该工厂：

@KafkaListener(
        topics = "your-target-topic",
        groupId = "group-type-a",
        containerFactory = "typeAFactory"
)
public void handleTypeAMessage(YourMessageEntity message) {
    // 处理类型A的业务逻辑
}

3. Kafka客户端拦截器（性能优先方案） #

如果消息量较大，推荐在Kafka客户端层面过滤，减少网络传输和Spring层面的无效处理。自定义消费者拦截器：

public class MessageTypeInterceptor implements ConsumerInterceptor<String, YourMessageEntity> {

    @Override
    public ConsumerRecord<String, YourMessageEntity> onConsume(ConsumerRecord<String, YourMessageEntity> record) {
        // 返回null表示Kafka跳过该消息，否则交给监听器
        return "A".equals(record.value().getMessageType()) ? record : null;
    }

    // 以下方法默认实现即可
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {}
    @Override
    public void configure(Map<String, ?> configs) {}
    @Override
    public void close() {}
}

然后在消费者工厂配置中添加拦截器：

@Configuration
public class KafkaConfig {

    @Bean
    public ConsumerFactory<String, YourMessageEntity> typeAConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-address");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-type-a");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // 配置拦截器
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MessageTypeInterceptor.class.getName());
        // 指定Json反序列化的目标类型
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, YourMessageEntity.class.getName());
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, YourMessageEntity> typeAFactory(
            ConsumerFactory<String, YourMessageEntity> typeAConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, YourMessageEntity> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(typeAConsumerFactory);
        return factory;
    }
}

核心注意点 #

• 不同类型的监听器必须使用独立的groupId，保证消息能正确分配到对应监听器。
• 确保消息反序列化配置正确，否则无法读取messageType字段进行过滤。
• 简单场景用SpEL表达式即可，复杂逻辑用自定义过滤器，高吞吐量场景优先选客户端拦截器。

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