关于后端:spring-boot使用Java并行流发送kafka消息报错

39次阅读

共计 1715 个字符,预计需要花费 5 分钟才能阅读完成。

一、背景

  • 公司业务分 2 个 Kafka,咱们组一个,其余组专用一个
  • 咱们组有 2 个业务在 Java 并行流中发消息到 其余组的 Kafka,一个是 批量治理接口(app 接口专用底层办法,不是批量的,没有用 并行流),另一个是 生产咱们组 Kafka 音讯而后发送。
  • 应用 spring-boot-maven-plugin 打包,公布到生产环境当前,偶然会接到 发送音讯到其余组 Kafka 报错告警,Invalid value org.apache.kafka.common.serialization.StringSerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.StringSerializer could not be found.

二、长期计划

  1. 查了 测试、开发 环境 最近 1 个月的日志,没有呈现过这个谬误
  2. 网上搜寻的后果,应该是 类加载器的问题
  3. 本地 IDEA 开发工具 启动程序、打包 jar 运行,拜访 app 接口,都没有重现
  4. 生产环境第一次当前就失常,并且没有重现的状况下,不敢批改上线,只能先加异样解决

三、起因剖析

  • 类加载器不一样
  • 应用 spring-boot-maven-plugin 打包进去是 fat jar,其中 BOOT-INF/lib/ 寄存依赖 jar,BOOT-INF/classes/ 寄存我的项目的 classes,应用 spring 自定义的 ClassLoader 加载
  • 业务解决应用了 parallelStream 包含发 kafka 音讯,底层应用 ForkJoin 线程池,因为是 JDK 的类,应用 BootClassLoader 加载,BootClassLoader 加载不到 spring 自定义目录 BOOT-INF 的类

四、最终计划

  1. 计划一:不应用 并行
    接口这边批量不会特地大,并且是 操作 Redis,就改回 一般 stream
  2. 计划二:自定义线程池
    生产 Kafka 那边,原来加 parallelStream() 就是因为 每条音讯要做 几个业务解决,串行的话,性能不够(Kafka 分区数公司不容许加太多了,当初 9 个)

留神:倡议应用 自定义 ForkJoinPool + submit() + join()。不能用 execute(),因为它是 异步执行的,也就是说,这条音讯 可能和 前面的音讯同时解决,产生并发问题。

// 自定义 ForkJoinPool,默认的应用 BootClassLoader 加载,有问题
ForkJoinPool bizPool = new ForkJoinPool(6, new ForkJoinWorkerThreadFactory() {
    @Override
    public ForkJoinWorkerThread newThread(ForkJoinPool pool) {return new ForkJoinWorkerThread(pool) {};}
}, null, false);
for (String msg : msgs) {bizPool.submit(() -> list.parallelStream().forEach(item -> {// do something, then send to kafka})).join();}

本文首发于 https://www.890808.xyz/spring-boot-kafka-send-error-with-fork-join/,其余平台须要审核更新慢一些。

五、参考链接

kafka 消费者报错:Class org.apache.kafka.common.serialization.StringDeserializer could not be found._Jaming R 的博客 -CSDN 博客

Kafka Producer – org.apache.kafka.common.serialization.StringSerializer could not be found – Stack Overflow

Fix StringSerializer could not be found when it not in ContextClassLoader by eshizhan · Pull Request #10938 · apache/kafka (github.com)

正文完
 0