KafkaListener 和 __listener @KafKaListener 无法使用 Aspect Annotation 解析“null”答案

【问题标题】：KafkaListener and __listener @KafKaListener can't resolve 'null' with Aspect AnnotationKafkaListener 和 __listener @KafKaListener 无法使用 Aspect Annotation 解析“null”
【发布时间】：2021-12-17 10:12:12
【问题描述】：

我正在使用@KafkaListener，我需要一个动态主题名称，所以我使用 SpEL '__listener' 来做到这一点

@PostConstruct
public void init() {
    myProps= generateTopicDynamically();
}

@KafkaListener(topics = "#{__listener.myProps}")
public void listenerKafka(@Payload MyObject myObject) {
       //Do something with my event
}

效果很好。

主要问题是当我想添加另一个触发某些方面编程的注释时

@MyCustomAnnotationToRecordPerformance @KafkaListener(topics = "#{__listener.myProps}") public void listenerKafka(@Payload MyObject myObject)

这里是方面类

@Aspect
@Configuration
@Slf4j
public class MyCustomAnnotationToRecordPerformanceAspect {

    @Pointcut("@annotation(MyCustomAnnotationToRecordPerformance)")
    public void annotationMyCustomAnnotationToRecordPerformance() {
    }

    @Around("annotationMyCustomAnnotationToRecordPerformance()")
    public Object doSomething(final ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
        return proceedingJoinPoint.proceed();
    }
}

我有这个问题是因为 Spring 在调用 @PostConstruct 之前尝试解析 __listener。

Caused by: java.lang.IllegalArgumentException: @KafKaListener can't resolve 'null' as a String
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.resolveAsString(KafkaListenerAnnotationBeanPostProcessor.java:648)
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.resolveTopics(KafkaListenerAnnotationBeanPostProcessor.java:520)
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.processListener(KafkaListenerAnnotationBeanPostProcessor.java:419)
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.processKafkaListener(KafkaListenerAnnotationBeanPostProcessor.java:370)
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.postProcessAfterInitialization(KafkaListenerAnnotationBeanPostProcessor.java:298)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyBeanPostProcessorsAfterInitialization(AbstractAutowireCapableBeanFactory.java:431)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:595)
    ... 41 common frames omitted

我尝试调试它我们可以看到很多CGLIB引用，所以bean已经被代理了，但是所有的属性都是空的。所以我认为 Autowired 和 PostConstruct 方法还没有被调用

目前，我尝试延迟管理 @KafkaListener 的处理器，但我无法找到可以在无需重新定义完全 Kafka 配置的情况下更改它的位置

@EnableKafka 导入 KafkaListenerConfigurationSelector 即 DeferredImportSelector。

这里是关于这门课的评论

A {@link DeferredImportSelector} implementation with the lowest order to import a {@link KafkaBootstrapConfiguration} as late as possible.

所以我认为它已经根据评论尽可能延迟了

我用@Transactional 对其进行了测试，我遇到了同样的问题。

@Transactional 
@KafkaListener(topics = "#{__listener.myProps}")
public void listenerKafka(@Payload MyObject myObject)

你有什么想法吗？

我现在看到的唯一选择是将我的班级分成 2 个并创建 2 个 bean。 KafkaListener 方法调用另一个 bean。但我发现必须这样做很奇怪。

提前感谢您的帮助。

【问题讨论】：

GitHub 上的MCVE 会有所帮助。重现问题是解决问题的第一步。此外，@Aspect 应该是@Component，而不是@Configuration。我不知道为什么这么多人会弄错。我不能保证它会帮助你解决问题，但它可能会。无论哪种方式，你都应该改变它。配置代理与组件代理的行为方式不同。
同意 (MCVE) - 对我来说效果很好。
是的，同意@kriegaex，它最初是一个组件，我尝试了不同的想法，忘了回到组件。感谢您的通知。

标签： java spring aop aspectj spring-kafka

【解决方案1】：

我刚刚使用 @Transactional 对其进行了测试，它对我来说可以正常工作 - 我已经确认，当我们到达 @KafkaListener 注释 BPP 时，我们已经有了一个 CGLIB 代理...

@SpringBootApplication
@EnableTransactionManagement
public class So69817946Application {

    public static void main(String[] args) {
        SpringApplication.run(So69817946Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so69817946").partitions(1).replicas(1).build();
    }

}

@Component
class listener {

    public String getTopic() {
        return "so69817946";
    }

    @Transactional
    @KafkaListener(id = "so69817946", topics = "#{__listener.topic}")
    public void listen(String in) {
        System.out.println(in);
    }

}

@Component
class TM extends AbstractPlatformTransactionManager {

    @Override
    protected Object doGetTransaction() throws TransactionException {
        return new Object();
    }

    @Override
    protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
    }

    @Override
    protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
    }

    @Override
    protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
    }

}

so69817946: partitions assigned: [so69817946-0]

而且我可以在调用堆栈中看到事务拦截器。

所以，是的，MCVE 会很有帮助。

【讨论】：

为此浪费了很多时间，我应该先考虑一下。在 Aspect 的情况下，bean 处理器使用代理的 getter，并且无权访问私有属性。我想当我看到你的 getTopic 方法时，我创建了一个 getMyProps() 方法，它可以工作如果你在没有任何 getter 的情况下更改你的代码 private topic="so69817946"，你会重现这个问题。谢谢你给了我缺失的线索。
是的；你必须添加一个吸气剂。您的意思是public 字段吗？ SpEL 无法访问private 字段。 EL1008E: Property or field 'topic' cannot be found - 如果我将其更改为公开，我会遇到 null 问题。
该属性在开始时是公开的以使 SpEL 工作，但对于 Aspect 和 CGLib，只有 getter 工作。我提出并回答了解决方案。真的很感谢。

【解决方案2】：

在 Gary 的帮助下，我找到了解决方案。一旦我们有了方面，类就被代理了，CGLIB 对象中的属性变为空。我们需要调用 getter 来获取原始对象的值，而不是代理对象

SpEL 能够读取将在原始对象上执行的公共 getter，而不是 CGLIB 对象

所以解决方案只是为我的私人创建一个公共 getter

public String getMyProps(){
    return this.myProps;
}

谢谢大家。

【讨论】：