keycloak~EventListenerProvider初始化kafka引出的类加载问题

种地  金牌会员 | 2023-7-19 17:37:45 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题

主题 960|帖子 960|积分 2880

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;
  6.   public KcKafkaConsumer(KeycloakSessionFactory keycloakSessionFactory, Properties properties, String topic)
  7.       throws ClassNotFoundException {
  8.     this.keycloakSessionFactory = keycloakSessionFactory;
  9.     this.kafkaConsumer = new KafkaConsumer<>(properties);
  10.     this.kafkaConsumer.subscribe(Collections.singleton(topic));
  11.   }
  12. @Override
  13.   public void run() {
  14.     try {
  15.       while (!closed.get()) {
  16.         ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
  17.         // 处理Kafka消息
  18.         for (ConsumerRecord<String, String> record : records) {
  19.           System.out.println("Topic:" + record.topic() + ",Received message: " + record.value());
  20.           //TODO: 处理Kafka消息的具体逻辑
  21.         }
  22.       }
  23.     } finally {
  24.       kafkaConsumer.close();
  25.     }
  26.   }
  27.   public void shutdown() {
  28.     closed.set(true);
  29.     kafkaConsumer.close();
  30.   }
  31. }
复制代码
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

种地

金牌会员
这个人很懒什么都没写!

标签云

快速回复 返回顶部 返回列表