ToB企服应用市场:ToB评测及商务社交产业平台

标题: keycloak~EventListenerProvider初始化kafka引出的类加载问题 [打印本页]

---

作者: 种地    时间: 2023-7-19 17:37
标题: keycloak~EventListenerProvider初始化kafka引出的类加载问题
EventListenerProvider初始

keycloak提供的事件处理机制，可以通过实现EventListenerProvider接口来实现自定义的事件处理逻辑。在keycloak启动时，会通过ServiceLoader机制加载所有的EventListenerProvider实现类，并将其注册到keycloak的事件处理机制中。

• 构造方法，在每个keycloak后台操作时，它都会重新构建实例
  
• OnEvent方法，在事件发生时执行，不会出现类加载问题，因为这样类已经被加载了
  

EventListenerProviderFactory

EventListenerProviderFactory是进行事件处理器的生产工厂，用于创建EventListenerProvider实例。在keycloak启动时，会通过ServiceLoader机制加载所有的EventListenerProviderFactory实现类，并将其注册到keycloak的事件处理机制中。

• init方法：keycloak启动时会执行，用于初始化EventListenerProviderFactory实例，可以在此方法中进行一些初始化操作。
  
• postInit方法：keycloak启动时会执行，在init方法之后，会执行这个方法
  
• create方法：在kc后台开启这个EventListenerProviderFactory之后，每次请求都会执行这个create方法，对于它生产的provider对象，可能考虑使用单例的方式， 避免每次请求都创建一个新的对象
  
• close方法：在keycloak程序关闭后或者当前事件被注册时，这个方法才会执行
  

问题

• 问题描述：在EventListenerProviderFactory的init方法中，通过kafka发送消息，会出现类加载问题，因为在keycloak启动时，kafka的类的加载器还没有被加载，所以会出现类加载问题。
  
• 解决：需要将类加载器这块，修改成当前类加载器去加载对应的文件，如下代码解决了类无法加载的问题
  

1.   @Override
2.   public void postInit(KeycloakSessionFactory keycloakSessionFactory) {
3.     try {
4.       this.executorService = Executors.newFixedThreadPool(2);
5.       Properties kafkaProperties = new Properties();
6.       kafkaProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
7.           ConfigFactory.getInstance().getStrPropertyValue("kafka.host"));
8.       kafkaProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "kc-ListenerProviderFactory");
9.       kafkaProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
10.       kafkaProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
11.       kafkaProperties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
12.       kafkaProperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
13.       kafkaProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
14.       kafkaProperties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
15.       // 需要使用当前类加载器，否则会出现无法加载StringDeserializer的情况
16.       Class<?> stringDeserializerClass =
17.           getClass().getClassLoader().loadClass("org.apache.kafka.common.serialization.StringDeserializer");
18.       kafkaProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, stringDeserializerClass);
19.       kafkaProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, stringDeserializerClass);
20.       this.kafkaConsumerAdd = new KcKafkaConsumer(keycloakSessionFactory,kafkaProperties, "black_list_add");
21.       executorService.submit(kafkaConsumerAdd);
22.       this.kafkaConsumerRemove=new KcKafkaConsumer(keycloakSessionFactory,kafkaProperties, "black_list_remove");
23.       executorService.submit(kafkaConsumerRemove);
24.     } catch (ClassNotFoundException e) {
25.       throw new RuntimeException(e);
26.     }
27.   }

复制代码

• 对于kafka-clients实现消费者的话，代码还是比较简单的
  

1. public class KcKafkaConsumer implements Runnable {
2.   private static final Logger logger = Logger.getLogger(ConfigFactory.class);
3.   private final AtomicBoolean closed = new AtomicBoolean(false);
4.   private final KeycloakSessionFactory keycloakSessionFactory;
5.   private KafkaConsumer<String, String> kafkaConsumer;
7.   public KcKafkaConsumer(KeycloakSessionFactory keycloakSessionFactory, Properties properties, String topic)
8.       throws ClassNotFoundException {
9.     this.keycloakSessionFactory = keycloakSessionFactory;
10.     this.kafkaConsumer = new KafkaConsumer<>(properties);
11.     this.kafkaConsumer.subscribe(Collections.singleton(topic));
12.   }
14. @Override
15.   public void run() {
16.     try {
18.       while (!closed.get()) {
19.         ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
20.         // 处理Kafka消息
21.         for (ConsumerRecord<String, String> record : records) {
22.           System.out.println("Topic:" + record.topic() + ",Received message: " + record.value());
23.           //TODO: 处理Kafka消息的具体逻辑
24.         }
25.       }
26.     } finally {
27.       kafkaConsumer.close();
28.     }
29.   }
30.   public void shutdown() {
31.     closed.set(true);
32.     kafkaConsumer.close();
33.   }
34. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4