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

一、背景

• 公司业务分 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)