java - 使用 AOP 的动态 Kafka 消费者
问题描述
我有几个动态的 Kafka 消费者(基于部门 ID 等),您可以在下面找到代码。
基本上,我想记录每个onMessage()
方法调用所花费的时间,因此我创建了一个@LogExecutionTime
方法级别的自定义注释并将其添加到onMessage()
method 。但是,即使每当有关于该主题的消息时都会调用 my 并且其他一切正常,但我logExecutionTime()
的 of永远不会被调用。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;
}
}
DepartmentsMessageConsumer 类:
@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
:
/**
* 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 一起工作。
那我建议你考虑分开 aDepartmentsMessageConsumer
和ConcurrentMessageListenerContainer
. 我的意思是把它ConcurrentMessageListenerContainer
移到单独的@Configuration
类中。是一个很好的ApplicationLauncher
候选人。使它成为一个@Bean
并依赖于你DepartmentsMessageConsumer
的注入。关键是您需要给 AOP 一个机会来检测您DepartmentsMessageConsumer
的@PostConstruct
.
推荐阅读
- sql-server - SQL Server - 查看物化视图列表(快照)
- java - 使用 javapoet 构建类时返回正确的类型对象
- c# - c#查看调用Angular组件中断,但直接调用Angular工作正常
- ios - 允许 iOS http 实时流和流
- python - 我可以使用 df1 中的一列和 df2 单元格中的任何值之一连接两个数据框吗?
- javascript - Uncaught (in promise) dom-to-image
- c++ - Visual Studio中编译wiringPi.h的错误
- python - 通过组合文本熊猫制作新的数据框
- mysql - 如果一列被更改,如何更新另一列?
- tensorflow - 在 TensorFlow 中提取地标周围的多个补丁