使用AOP实现动态Kafka消费者

3

我有几个基于部门ID等的动态Kafka消费者,下面是代码。

基本上,我想记录每个onMessage()方法调用所花费的时间,因此我创建了一个@LogExecutionTime方法级别的自定义注释,并将其添加到onMessage()方法中。 但是,即使我的onMessage()在主题上有消息时被调用并且其他所有内容都正常工作,我的LogExecutionTimeAspectlogExecutionTime()从未被调用。

请问,我在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),但没有起作用。

1个回答

1

您应该考虑在@EnableAspectJAutoProxy上打开此选项:

/**
 * Indicate that the proxy should be exposed by the AOP framework as a {@code ThreadLocal}
 * for retrieval via the {@link org.springframework.aop.framework.AopContext} class.
 * Off by default, i.e. no guarantees that {@code AopContext} access will work.
 * @since 4.3.1
 */
boolean exposeProxy() default false;

另一方面,有一种类似于这样的东西,将比AOP更好:
/**
 * A plugin interface that allows you to intercept (and possibly mutate) records received by the consumer. A primary use-case
 * is for third-party components to hook into the consumer applications for custom monitoring, logging, etc.
 *
 * <p>
 * This class will get consumer config properties via <code>configure()</code> method, including clientId assigned
 * by KafkaConsumer if not specified in the consumer config. The interceptor implementation needs to be aware that it will be
 * sharing consumer config namespace with other interceptors and serializers, and ensure that there are no conflicts.
 * <p>
 * Exceptions thrown by ConsumerInterceptor methods will be caught, logged, but not propagated further. As a result, if
 * the user configures the interceptor with the wrong key and value type parameters, the consumer will not throw an exception,
 * just log the errors.
 * <p>
 * ConsumerInterceptor callbacks are called from the same thread that invokes {@link org.apache.kafka.clients.consumer.KafkaConsumer#poll(long)}.
 * <p>
 * Implement {@link org.apache.kafka.common.ClusterResourceListener} to receive cluster metadata once it's available. Please see the class documentation for ClusterResourceListener for more information.
 */
public interface ConsumerInterceptor<K, V> extends Configurable {

更新

@EnableAspectJAutoProxy(exposeProxy=true) 没有生效,我知道可以使用拦截器,但我想让它与AOP一起工作。

然后我建议您考虑将 DepartmentsMessageConsumerConcurrentMessageListenerContainer 分开。我的意思是将 ConcurrentMessageListenerContainer 移动到单独的 @Configuration 类中。 ApplicationLauncher 是一个很好的选择。将其作为 @Bean 并依赖于您的 DepartmentsMessageConsumer 进行注入。重点是需要给AOP一个机会来操纵您的 DepartmentsMessageConsumer,但是在 @PostConstruct 中,这太早了,无法从Kafka实例化和开始消费。


@EnableAspectJAutoProxy(exposeProxy=true) 没有起作用,我知道我可以使用拦截器,但我想让它与AOP一起工作。 - developer
请在我的答案中查找更新。 - Artem Bilan
我已将ConcurrentMessageListenerContainer的创建提取到单独的配置中,并公开为Bean,现在我收到了java.lang.IllegalArgumentException: A org.springframework.kafka.listener.GenericMessageListener实现必须被提供的错误提示。 - developer
我已经将最新的代码更新到问题本身。 - developer
你仍然在那里使用 containerProperties.setMessageListener(this);,但是这是错误的。你需要将你的 DepartmentsMessageConsumer 作为一个接口注入到 concurrentMessageListenerContainer() 中,例如:@Qualifier("departmentsMessageConsumer") MessageListener messageListener - Artem Bilan
非常感谢,我明白了。我之前使用了this,这是错误的。需要使用由Spring容器创建的相同监听器bean。现在已经解决了,非常感谢您再次帮助我。 - developer

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接