spring kafka能否实现消息过滤

作者：袖梨 2026-06-15

是的，Spring Kafka 可以实现消息过滤。在 Spring Kafka 中，你可以使用 KafkaMessageListenerContainer 和 MessageListenerAdapter 来处理接收到的消息。为了实现消息过滤，你可以在 MessageListenerAdapter 的实现类中编写自定义的逻辑来过滤消息。

以下是一个简单的示例：

首先，创建一个实现 ConsumerAwareErrorHandler 接口的类，用于处理接收到的错误消息：

import org.springframework.kafka.listener.ConsumerAwareErrorHandler;import org.springframework.kafka.listener.Message;public class CustomErrorHandler implements ConsumerAwareErrorHandler {@Overridepublic void handle(Exception thrownException, Message message, ConsumerRecord<?, ?> data) {// 在这里编写你的错误处理逻辑}}

创建一个实现 MessageListener 接口的类，用于处理接收到的消息：

import org.springframework.kafka.listener.MessageListener;import org.springframework.messaging.handler.annotation.Header;import org.springframework.stereotype.Component;@Componentpublic class CustomMessageListener implements MessageListener {@Overridepublic void onMessage(Message message) {// 在这里编写你的消息过滤和处理逻辑String payload = new String(message.getPayload());String key = message.getKey();// 示例：根据消息头或消息体进行过滤if (shouldFilter(payload)) {// 处理过滤后的消息} else {// 忽略过滤后的消息}}private boolean shouldFilter(String payload) {// 在这里编写你的过滤逻辑return payload.contains("filtered");}}

在你的 KafkaListenerEndpoint 配置类中，将 CustomMessageListener 与 KafkaMessageListenerContainer 关联起来：

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.kafka.annotation.KafkaListenerConfigurer;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.config.KafkaListenerEndpointRegistrar;import org.springframework.kafka.config.KafkaListenerEndpointRegistry;import org.springframework.kafka.core.ConsumerFactory;import org.springframework.kafka.core.DefaultKafkaConsumerFactory;import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import org.springframework.kafka.listener.config.MethodKafkaListenerEndpoint;import org.springframework.kafka.listener.config.MethodKafkaListenerEndpointRegistrar;import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;import org.springframework.kafka.support.serializer.JsonDeserializer;import java.util.HashMap;import java.util.Map;@Configurationpublic class KafkaListenerConfig implements KafkaListenerConfigurer {@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);return factory;}@Overridepublic void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {Map<String, Object> props = new HashMap<>();// 配置你的消费者属性，如 groupId、bootstrapServers 等// ...registrar.registerEndpoint(new MethodKafkaListenerEndpoint<>("custom-topic","customMethod",getClass().getClassLoader(),String.class,String.class,props));}@Beanpublic KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry(KafkaListenerEndpointRegistrar registrar) {KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();registrar.afterPropertiesSet();registry.start();return registry;}}

在你的 CustomMessageListener 实现类中，使用 @KafkaListener 注解指定要监听的主题和组：

import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;@Componentpublic class CustomMessageListener implements MessageListener {@Override@KafkaListener(topics = "custom-topic", groupId = "custom-group")public void onMessage(Message message) {// 在这里编写你的消息过滤和处理逻辑}}