Spring AOP切面处理Kafka消费记录：拦截无效且需获取主题名

一、为什么拦截MessageListener.onMessage的AOP不生效 #

Spring AOP基于动态代理实现，仅能拦截Spring容器中Bean的方法调用，而Kafka消费逻辑中存在两个核心限制：

• MessageListener的实例通常由Spring Kafka内部（如KafkaListenerEndpoint）创建并绑定到消费容器，并非直接作为Spring Bean被管理；
• 消费容器调用onMessage时，是直接调用实例方法，而非通过Spring代理对象，因此Spring AOP无法拦截到该方法。

替代方案：使用Spring Kafka官方的ListenerInterceptor #

这是更适配Kafka消费拦截的原生方案，无需依赖AOP，示例代码如下：

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.listener.ListenerInterceptor;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.kafka.support.KafkaHeaders;

@Slf4j
@Component
public class KafkaRecordInterceptor implements ListenerInterceptor<Object, Object> {

    @Override
    public Message<Object> intercept(Message<Object> message, org.apache.kafka.clients.consumer.Consumer<?, ?> consumer) {
        // 从消息头中获取主题名（含重试主题）
        String topic = message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class);
        log.debug("拦截到消息，主题：{}", topic);
        // 继续处理消息
        return message;
    }
}

在Kafka容器配置中注册拦截器：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConsumerFactory<?, ?> consumerFactory,
            KafkaRecordInterceptor interceptor) {
        ConcurrentKafkaListenerContainerFactory<?, ?> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 绑定拦截器
        factory.setListenerInterceptors(interceptor);
        return factory;
    }
}

二、拦截@KafkaListener方法时获取主题名（含重试主题） #

如果坚持使用AOP拦截@KafkaListener方法，可通过以下两种方式获取主题：

方式1：从方法参数中提取主题信息 #

遍历切面获取的方法参数，从ConsumerRecord或Spring Message中提取主题：

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.messaging.Message;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.stereotype.Component;
import org.apache.kafka.clients.consumer.ConsumerRecord;

@Slf4j
@Aspect
@Component
public class KafkaListenerAspect {

    @Around("@annotation(org.springframework.kafka.annotation.KafkaListener)")
    public Object interceptKafkaListener(ProceedingJoinPoint joinPoint) throws Throwable {
        for (Object arg : joinPoint.getArgs()) {
            if (arg instanceof ConsumerRecord<?, ?> record) {
                String topic = record.topic();
                log.debug("当前消费主题：{}", topic);
                break;
            } else if (arg instanceof Message<?> message) {
                String topic = message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class);
                log.debug("当前消费主题：{}", topic);
                break;
            }
        }
        return joinPoint.proceed();
    }
}

方式2：通过ConsumerContext获取当前消费信息 #

如果方法参数中没有ConsumerRecord或Message，可利用线程绑定的ConsumerContext获取主题：

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.kafka.listener.ConsumerContext;
import org.springframework.stereotype.Component;

@Slf4j
@Aspect
@Component
public class KafkaListenerAspect {

    private final ConsumerContext consumerContext;

    public KafkaListenerAspect(ConsumerContext consumerContext) {
        this.consumerContext = consumerContext;
    }

    @Around("@annotation(org.springframework.kafka.annotation.KafkaListener)")
    public Object interceptKafkaListener(ProceedingJoinPoint joinPoint) throws Throwable {
        // 获取当前消费的主题（仅在消费线程中有效）
        String topic = consumerContext.getConsumer().assignment().iterator().next().topic();
        log.debug("当前消费主题：{}", topic);
        return joinPoint.proceed();
    }
}

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