使用@KafkaListener时，能否仅依据消息Key触发消费逻辑？

方案一：使用Spring Kafka的RecordFilterStrategy（推荐） #

通过自定义过滤器，在消息到达监听方法前就过滤掉不符合条件的消息，避免进入方法做冗余判断，效率更高。

1. 自定义过滤器Bean：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.stereotype.Component;

@Component
public class InitiateKeyFilter implements RecordFilterStrategy<String, MessageModel> {
    @Override
    public boolean filter(ConsumerRecord<String, MessageModel> record) {
        // 返回true代表过滤该消息，false则保留——也就是只留下key为Initiate的消息
        return !"Initiate".equals(record.key());
    }
}

2. 在@KafkaListener中绑定过滤器：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    @Autowired
    private InitiateKeyFilter initiateKeyFilter;

    @KafkaListener(
        topics = "test_topic", 
        groupId = "group_id",
        filter = "initiateKeyFilter" // 指定过滤器Bean的名称
    )
    public void consumerListener(ConsumerRecord<String, MessageModel> message) throws IOException {
        businessFunctionality(message.value()); // 无需再做判断，进来的都是符合条件的消息
    }

    private void businessFunctionality(MessageModel value) {
        // 业务逻辑实现
    }
}

方案二：使用@KafkaListener的properties配置（轻量方式） #

如果不想单独创建过滤器类，可直接在注解里通过properties结合SpEL表达式实现过滤：

@KafkaListener(
    topics = "test_topic", 
    groupId = "group_id",
    properties = {
        "filter.record=!#record.key().equals('Initiate')" // 表达式含义：过滤掉key不是Initiate的消息
    }
)
public void consumerListener(ConsumerRecord<String, MessageModel> message) throws IOException {
    businessFunctionality(message.value());
}

方案对比 #

• 方案一的优势是过滤逻辑可复用，适合多个监听方法需要相同规则的场景，代码结构更清晰。
• 方案二更轻量，适合单一监听方法的简单过滤需求，无需额外创建类。

这两种方案都比原方法更优：过滤逻辑提前到了消息监听容器层面，不符合条件的消息不会进入业务方法，减少了不必要的方法调用和判断，性能更优。

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