我有几个基于部门ID等的动态Kafka消费者,下面是代码。
基本上,我想记录每个onMessage()
方法调用所花费的时间,因此我创建了一个@LogExecutionTime
方法级别的自定义注释,并将其添加到onMessage()
方法中。
但是,即使我的onMessage()
在主题上有消息时被调用并且其他所有内容都正常工作,我的LogExecutionTimeAspect
的logExecutionTime()
从未被调用。
请问,我在LogExecutionTimeAspect
类中缺少什么,以便它开始工作?
LogExecutionTime:
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface LogExecutionTime {
}
LogExecutionTimeAspect class:
@Aspect
@Component
public class LogExecutionTimeAspect {
@Around("within(com.myproject..*) && @annotation(LogExecutionTime)")
public Object logExecutionTime(ProceedingJoinPoint joinPoint) throws Throwable {
long startTime = System.currentTimeMillis();
Object object = joinPoint.proceed();
long endTime = System.currentTimeMillis();
System.out.println(" Time taken by Listener ::"+(endTime-startTime)+"ms");
return object;
}
}
DepartmentsMessageConsumer class:
@Component
public class DepartmentsMessageConsumer implements MessageListener {
@Value(value = "${spring.kafka.bootstrap-servers}" )
private String bootstrapAddress;
@PostConstruct
public void init() {
Map<String, Object> consumerProperties = new HashMap<>();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "DEPT_ID_HERE");
ContainerProperties containerProperties =
new ContainerProperties("com.myproj.depts.topic");
containerProperties.setMessageListener(this);
DefaultKafkaConsumerFactory<String, Greeting> consumerFactory =
new DefaultKafkaConsumerFactory<>(consumerProperties,
new StringDeserializer(),
new JsonDeserializer<>(Department.class));
ConcurrentMessageListenerContainer container =
new ConcurrentMessageListenerContainer<>(consumerFactory,
containerProperties);
container.start();
}
@Override
@LogExecutionTime
public void onMessage(Object message) {
ConsumerRecord record = (ConsumerRecord) message;
Department department = (Department)record.value();
System.out.println(" department :: "+department);
}
}
ApplicationLauncher类:
@SpringBootApplication
@EnableKafka
@EnableAspectJAutoProxy
@ComponentScan(basePackages = { "com.myproject" })
public class ApplicationLauncher extends SpringBootServletInitializer {
public static void main(String[] args) {
SpringApplication.run(ApplicationLauncher.class, args);
}
}
编辑:
我尝试过使用@EnableAspectJAutoProxy(exposeProxy=true)
,但没有起作用。
@EnableAspectJAutoProxy(exposeProxy=true)
没有起作用,我知道我可以使用拦截器,但我想让它与AOP一起工作。 - developerConcurrentMessageListenerContainer
的创建提取到单独的配置中,并公开为Bean,现在我收到了java.lang.IllegalArgumentException: A org.springframework.kafka.listener.GenericMessageListener
实现必须被提供的错误提示。 - developercontainerProperties.setMessageListener(this);
,但是这是错误的。你需要将你的DepartmentsMessageConsumer
作为一个接口注入到concurrentMessageListenerContainer()
中,例如:@Qualifier("departmentsMessageConsumer") MessageListener messageListener
。 - Artem Bilanthis
,这是错误的。需要使用由Spring容器创建的相同监听器bean。现在已经解决了,非常感谢您再次帮助我。 - developer