夯实 kafka 系列|第六章：自界说注解 @EvalEventListener 开发

夯实 kafka 系列|第六章：自界说注解 @EvalEventListener 开发



1.媒介

在文章 《夯实 kafka 系列|第五章：基于 kafka 分布式事故框架 eval-event》 留了一个功能未实现

• @EvalEventListener 监听分布式事故
  

本文来讨论这个自界说注解怎样开发。
源码已更新到 github

• https://github.com/huajiexiewenfeng/eval-event
  

2.需求

通过 @EvalEventListener  监听到分布式事故 MyEvent(自界说事故)

1. @EvalEventListener
2. public void onEvalEvent(MyEvent event){
3.     ...
4. }

复制代码

3.思绪

3.1 @kafkaListener 注解的实现

我们先看看 kafka 官方怎样实现 @kafkaListener，然后参照这个来举行实现。
3.1.1 KafkaListenerAnnotationBeanPostProcessor（重点）

• 作用：在 Spring 容器初始化时扫描全部 Bean，辨认带有 @KafkaListener 注解的方法。
  
• 流程：
  
  
  • 遍历全部 Bean 的类方法。
    
  • 通过反射查抄方法是否标注 @KafkaListener。
    
  • 将符合条件的监听方法封装为 MethodKafkaListenerEndpoint。
    
  
  

3.1.2 监听器端点注册

• 将分析后的端点信息注册到 KafkaListenerEndpointRegistry（管理全部监听容器的中央注册表）。
  

3.1.3 监听器容器创建

• 容器工厂 ConcurrentKafkaListenerContainerFactory
  

• 作用：根据 @KafkaListener 的设置创建 ConcurrentMessageListenerContainer。
  

3.1.4 动态署理天生（重点）

• 对带有 @KafkaListener 的方法天生动态署理，使其具备消息处理惩罚本领。
  

3.1.5 总结

一共分为三步

• 在 Bean 的初始化后生命周期阶段，找到全部标注 @KafkaListener 的方法和类
  
• 整理好全部须要的元数据信息（Bean Method topic 等等），用于动态署理+斲丧 kafka 消息
  
• 动态署理，实验类的方法，即标注 @KafkaListener 的方法
  

3.2 参考实现

我们大抵实现流程如下：

• EvalEventListenerAnnotationBeanPostProcessor 实现 BeanPostProcessor
  
• 遍历全部 Bean 的类方法，通过反射查抄方法是否标注 @EvalEventListener
  
• 找到 topic 和 class 映射关系
  
• 启动线程，拉取 topic 消息
  
• 找到 @EvalEventListener 对应的 class，以及 method
  
• 反射实验目的类的 method 方法
  

4.开发

3.1 @EvalEventListener 注解

1. package com.csdn.event.kafka.annotation;
3. import java.lang.annotation.*;
5. @Target({ElementType.METHOD})
6. @Retention(RetentionPolicy.RUNTIME)
7. @Documented
8. public @interface EvalEventListener {
9.     /**
10.      * 指定监听的事件类型（默认根据方法参数推断）
11.      */
12.     Class<?>[] eventType() default {};
13. }

复制代码

3.2 EvalEventAnnotationDefinition 注解界说

用于存储 Bean Method topic 等等信息

1. public class EvalEventAnnotationDefinition {
2.     private String topic;
4.     // 事件的目标类
5.     private Class<?> targetClass;
7.     // 事件的类型
8.     private Class<?> eventClass;
10.     private Method method;
11.     ...
12. }

复制代码

3.3 EvalEventListenerAnnotationBeanPostProcessor 初始化后处理惩罚器

• 找到标注 @EvalEventListener 类和方法
  
• 循环实验 processEvalEventListener 方法
  
  
  
  • 封装 EvalEventAnnotationDefinition 信息
    
  • 启动 EvalEventListenerAnnotationThread 线程
    
  
  

1. /**
2. * This class processes beans after their initialization to find methods annotated with
3. * {@link EvalEventListener} and register them as event listeners.
4. */
5. @Component
6. public class EvalEventListenerAnnotationBeanPostProcessor<T extends EvalEvent> implements BeanPostProcessor, ApplicationContextAware {
8.     ...
10.     @Override
11.     public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
12.         Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
13.         Map<Method, Set<EvalEventListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
14.                 (MethodIntrospector.MetadataLookup<Set<EvalEventListener>>) method -> {
15.                     Set<EvalEventListener> listenerMethods = findListenerAnnotations(method);
16.                     return (!listenerMethods.isEmpty() ? listenerMethods : null);
17.                 });
18.         if (annotatedMethods.isEmpty()) {
19.             // No methods found
20.             logger.debug("No @EvalEventListener annotations found on bean '{}'", beanName);
21.         } else {
22.             // Non-empty set of methods
23.             for (Map.Entry<Method, Set<EvalEventListener>> entry : annotatedMethods.entrySet()) {
24.                 Method method = entry.getKey();
25.                 for (EvalEventListener listener : entry.getValue()) {
26.                     processEvalEventListener(listener, method, bean);
27.                 }
28.             }
29.             logger.debug(" {} @KafkaListener methods processed on bean '{}'"
30.                     , annotatedMethods.size(), beanName + ": " + annotatedMethods);
31.         }
32.         return bean;
33.     }
35.     private void processEvalEventListener(EvalEventListener listener, Method method, Object bean) {
36.         // topic
37.         Class<?>[] eventTypeArr = listener.eventType();
38.         Class<?> eventType = eventTypeArr.length > 0 ? eventTypeArr[0] : null;
39.         if (eventType == null) {
40.             // 从方法参数中获取
41.             eventType = method.getParameterTypes()[0];
42.         }
43.         if (eventType == null) {
44.             throw new IllegalStateException("Event type not specified for @EvalEventListener on method: " + method);
45.         }
46.         try {
47.             Object o = eventType.newInstance();
48.             if (!(o instanceof EvalEvent)) {
49.                 throw new IllegalStateException("Event type must be a subclass of EvalEvent: " + eventType);
50.             }
51.             EvalEvent event = (EvalEvent) o;
52.             EvalEventAnnotationDefinition evalEventDefinition = new EvalEventAnnotationDefinition(event.getTopic(), bean.getClass(), eventType, method);
53.             EvalEventListenerAnnotationThread<T> evalEventListenerThread =
54.                     new EvalEventListenerAnnotationThread<>(evalEventDefinition, eventKafkaConsumerFactory, applicationContext);
55.             evalEventListenerThread.start();
56.         } catch (Exception e) {
57.             throw new RuntimeException(e);
58.         }
59.     }
60. ...
61. }

复制代码

3.4 EvalEventListenerAnnotationThread 斲丧逻辑

• 创建KafkaConsumer
  
• 拉取 kafka 消息
  
• 动态署理实验目的类的方法
  
• ack 提交
  

1. public class EvalEventListenerAnnotationThread<T extends EvalEvent> extends Thread {
3. ...
5.     @Override
6.     public void run() {
7.         // 1. 创建KafkaConsumer
8.         KafkaConsumer<String, ?> consumer;
9.         try {
10.             consumer = eventKafkaConsumerFactory.buildKafkaConsumer(evalEventAnnotationDefinition.getTargetClass());
11.             List<String> topicList = new ArrayList<>();
12.             topicList.add(evalEventAnnotationDefinition.getTopic());
13.             consumer.subscribe(topicList);
14.         } catch (Exception e) {
15.             log.error("KafkaConsumer构造失败", e);
16.             e.printStackTrace();
17.             return;
18.         }
19.         // 2. 消费消息
20.         try {
21.             while (true) {
22.                 try {
23.                     // 3. 拉取消息
24.                     ConsumerRecords<String, ?> records = consumer.poll(
25.                             Duration.ofMillis(500));
26.                     if (records.isEmpty()) {
27.                         continue;
28.                     }
29.                     // 4. 处理消息
30.                     dispatch(records);
31.                     // 5. 使用异步提交规避阻塞
32.                     consumer.commitAsync();
33.                 } catch (Exception e) {
34.                     log.error("消息处理异常", e);
35.                 }
36.             }
37.         } finally {
38.             try {
39.                 // 6.最后一次提交使用同步阻塞式提交
40.                 consumer.commitSync();
41.             } finally {
42.                 consumer.close();
43.             }
44.         }
45.     }
47.     private void dispatch(ConsumerRecords<String, ?> records) {
48.         for (ConsumerRecord<String, ?> record : records) {
49.             try {
50.                 Object data = record.value();
51.                 if (data == null) {
52.                     log.warn("接收到空消息记录，跳过处理");
53.                     continue;
54.                 }
56.                 // 转换事件数据为指定类型
57.                 T event = (T) ConvertUtil.convertEvent(data, evalEventAnnotationDefinition.getEventClass());
58.                 if (event == null) {
59.                     log.error("事件数据转换失败，跳过处理");
60.                     continue;
61.                 }
63.                 Class<?> targetClass = evalEventAnnotationDefinition.getTargetClass();
64.                 Method method = evalEventAnnotationDefinition.getMethod();
66.                 if (targetClass == null || method == null) {
67.                     log.error("目标类或方法为空，无法处理事件");
68.                     continue;
69.                 }
71.                 // 从 Spring 容器获取 bean 实例
72.                 Object target;
73.                 try {
74.                     // 先尝试从 Spring 容器获取
75.                     target = applicationContext.getBean(targetClass);
76.                 } catch (NoSuchBeanDefinitionException e) {
77.                     // 如果容器中没有，尝试创建新实例
78.                     log.warn("Spring 容器中未找到 {} 的实例，将创建新实例", targetClass.getSimpleName());
79.                     target = targetClass.getDeclaredConstructor().newInstance();
80.                     // 可选：如果需要依赖注入，可以在这里自动装配
81.                     autowireBean(target);
82.                 }
83.                 method.setAccessible(true);
84.                 // 调用目标方法
85.                 method.invoke(target, event);
87.             } catch (InstantiationException | IllegalAccessException e) {
88.                 log.error("创建目标类实例失败: {}", e.getMessage(), e);
89.             } catch (NoSuchMethodException e) {
90.                 log.error("找不到目标类的无参构造方法: {}", e.getMessage(), e);
91.             } catch (InvocationTargetException e) {
92.                 log.error("方法调用失败: {}", e.getTargetException().getMessage(), e.getTargetException());
93.             } catch (Exception e) {
94.                 log.error("事件处理失败: {}", e.getMessage(), e);
95.             }
96.         }
97.     }
99. ...
101. }

复制代码

5.测试

5.1 测试代码

1. @Component
2. public class UserAnnotationEventListener  {
3.     @EvalEventListener
4.     public void onEvent(UserCreatedEvent event) {
5.         System.out.println("Received event: " + JSONObject.toJSONString(event));
6.     }
7. }

复制代码

5.2 源码调试

断点1 EvalEventListenerAnnotationBeanPostProcessor

• 扫描到 UserAnnotationEventListener 以及对应的方法
  

断点2

• 封装 EvalEventAnnotationDefinition
  

断点3 EvalEventListenerAnnotationThread

• 拉取 kafka 消息
  

断点4 EvalEventListenerAnnotationThread 动态署理实验方法

5.3 测试结果