KafkaListener和u listener@KafkaListener无法使用方面注释解析“null”

Question

我使用@KafkaListener，我需要一个动态的主题名，所以我使用SpEL'\uu listener'来实现这一点

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

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

它工作得非常好。

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

@MyCustomAnnotationToRecordPerformance@KafkaListener（主题 = "#{__ listener.my道具}"）公共无效监听器Kafka（@Payload MyObject myObject）

这里是aspect类

@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之前解决侦听器。

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

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

@enableKafka导入KafkaListenerConfigurationSelector，这是DeferredConfigurationSelector。

下面是对这门课的评论

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。但我觉得这样做很奇怪。

提前感谢你的帮助。

Answer 1

多亏了加里的帮助，我找到了解决办法。一旦我们有了方面，类就会被代理，CGLIB对象中的属性就会变为null。我们需要调用getter以获得来自原始对象的值，而不是来自代理对象的值

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

因此，解决方案就是为我的私人计算机创建一个公共getter

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

谢谢大家。

Answer 2

我刚刚用@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会很有帮助。