Java并行流指北

举报
javalover123 发表于 2023/07/06 23:36:27 2023/07/06
【摘要】 Java并行流,方便了 并发操作,但是不注意可能会导致问题。如 最大线程数,怎么控制并发数,类加载器,线程上下文变化,ForkJoinPool 的 execute、submit、invoke 方法的区别 等。

一、前言

  • Java并行流,方便了 并发操作,但是不注意可能会导致问题。
  • 如 最大线程数,怎么控制并发数,类加载器,线程上下文变化,ForkJoinPool 的 execute、submit、invoke 方法的区别 等。
  • 注意:本文以 openjdk 11.0.10 为例,没有特殊说明时,都是指 ForkJoinPool.commonPool()

二、注意点

1. 并行度

  • ***并行度 不等于 最大线程数(maximumPoolSize)***,下图 commonPool 有49个线程,但是 并行度为1
  • 默认的 并行度为 CPU核数 - 1,最小为 1
  • 可通过 -Djava.util.concurrent.ForkJoinPool.common.parallelism=数量 设置
    image.png

2. 容器里面的并行度

  • 下图中,/sys/fs/cgroup/cpu/cpu.cfs_quota_us 除以 /sys/fs/cgroup/cpu/cpu.cfs_period_us = cpu核数
  • 不等于 nproc,更不等于 获得宿主机的 lscpu | grep ‘CPU(s):’
    image.png

3. 最大线程数

  • 并行度 不等于 最大线程数(maximumPoolSize)
  • 即使 并行度 parallelism 为1,还有 备用线程(maximumPoolSize、COMMON_MAX_SPARES)
  • commonPool 默认 256,自定义 ForkJoinPool() 默认 32767。这样看,比较少会出现 线程数不够的情况。
    image.png

4. 并发太大,压垮后端

  • 假如 ForkJoinPool.commonPool() 线程比较多,并行流集合的元素也比较多时,给下游较大压力
  • jstack pid | grep -c commonPool

5. 线程上下文变化

如:获取不到用户信息了,可以获取到用户信息以后,传到并行流使用

final String deviceUdid = RequestUtils.getDeviceUdid();
data.parallelStream().forEach(d -> {
	// use deviceUdid instead of RequestUtils.getDeviceUdid() do something
});

6. ForkJoinPool 的 execute、submit、invoke 方法的区别

  • 有些简单的任务,不想单独创建线程池,可以用 ForkJoinPool.commonPool()
  • execute():异步执行,没有返回值,不能等待执行完成
  • submit():异步执行,返回 ForkJoinTask,需增加 .join() 等待完成
  • invoke():等于 submit() + join()

7. spring boot使用Java并行流发送kafka消息报错

  • 类加载器不一样,详见 spring boot 使用 Java 并行流发送 kafka 消息报错
  • 使用 spring-boot-maven-plugin 打包以后,依赖在 jar里面自定义位置(BOOT-INF/lib/),使用 org.springframework.boot.loader.LaunchedURLClassLoader 加载
  • ForkJoinPool.commonPool 默认使用 DefaultForkJoinWorkerThreadFactory,用的 系统ClassLoader,所以 并行流加载不到依赖的 class
  • 可通过 -Djava.util.concurrent.ForkJoinPool.common.threadFactory 设置 自定义线程工厂,使用当前 ClassLoader 解决
    image.png

8. 自定义并行流线程池

参考 concurrency - Custom thread pool in Java 8 parallel stream - Stack Overflow

  • 方案一(各种情况都有效)
CompletableFuture.runAsync(list.parallelStream().forEach(), new ForkJoinPool(2)).join()
  • 方案二(部分场景似乎没有效果)
// 第4个参数 asyncMode,默认 false,设置为 true 适用于 FIFO
ForkJoinPool forkJoinPool = new ForkJoinPool(2, pool -> new ForkJoinWorkerThread(pool) {
}, null, false);
forkJoinPool.invoke(() -> list.parallelStream().forEach());

9. 控制并发数

  • 可考虑把 集合切分成需要的份数,然后 parallelStream()
List<String> list = List.of("a", "b", "c");
CollUtil.split(list, list.size() / 2 + 1).parallelStream().forEach(b -> {
    b.stream().forEach(System.out::println);
});

10. 顺序消费

  • 如 forEachOrdered 会导致没有并发效果
  • 需要并行,还要使用输入顺序的,可考虑把 集合切分成需要的份数,然后 parallelStream()

三、总结

本文遵守【CC BY-NC】协议,转载请保留原文出处及本版权声明,否则将追究法律责任。
本文首先发布于 https://www.890808.xyz/ ,其他平台需要审核更新慢一些。

javalover123

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。