动态Kafka消费者与AOP

我有几个动态卡夫卡消费者(基于部门ID等),你可以找到下面的代码。

基本上,我想记录每一次的时间。 onMessage() 方法调用,因此我创建了 @LogExecutionTime 方法自定义自定义注释并将其添加到 onMessage() 方法。 但是我的 logExecutionTime() 属于 LogExecutionTimeAspect 即使我的电话 onMessage() 当有关于主题的消息时调用,其他一切都正常工作。

你能帮我弄清楚我遗漏了什么吗 LogExecutionTimeAspect 上课让它开始工作?

日志执行时间:

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface LogExecutionTime {
}

LogExecutionTimeAspect类:

@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;
    }
}

部门信息消费者类别:

@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);
    }
}

应用程序启动程序类:

@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) ,但不起作用。