文章/答案/技术大牛

发布

社区首页 >专栏 >04-CompletableFuture异步线程性能

04-CompletableFuture异步线程性能

彼岸舞

发布于 2022-05-10 03:14:52

4030

文章被收录于专栏：java开发的那点事java开发的那点事

场景

场景	方法
任务少, 不频繁	直接使用线程
任务数稳定,频繁	使用线程池

线程池

优点

不用频繁的创建和销毁线程
不需要担心OOM
直接往任务队列添加任务即可
核心线程忙不过来,可以自动增加到最大线程数

构造参数

核心线程数
最大线程数
空闲活跃时长
时长单位
阻塞队列
线程工厂
拒绝策略
- 直接丢弃
- 替换最后一个
- 抛异常
- 谁提交的任务谁执行
- --- 自行扩展

工具类

Executors

方法	描述
newSingleThreadExecutor	创建一个单线程的线程池
newCachedThreadPool	创建一个无上限的线程池(Integer.MAX)
newFixedThreadPool	创建一个固定线程数的线程池

需求

小白和他的朋友门,连续输了10几把游戏, 决定去餐厅吃饭了,3个人,直接点了10盘菜,决定化悲愤为食量

实现

编写代码

先将之前的公共方法抽成一个工具类

package com.dance;

import java.util.StringJoiner;

public class SmallTool {
    /**
     * 休眠方法
     * @param millis 毫秒
     */
    public static void sleep(long millis){
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 打印方法
     * @param text 文本
     */
    public static void print(String text){
        String str = new StringJoiner("\t|\t")
                .add(String.valueOf(System.currentTimeMillis()))
                .add(String.valueOf(Thread.currentThread().getId()))
                .add(Thread.currentThread().getName())
                .add(text)
                .toString();
        System.out.println(str);
    }
}

创建菜类

package com.dance;

import java.util.concurrent.TimeUnit;

/**
 * 菜
 */
public class Dish {
    /**
     * 菜名
     */
    private final String name;
    /**
     * 用时(秒)
     */
    private final Integer productionTime;

    public Dish(String name, Integer productionTime) {
        this.name = name;
        this.productionTime = productionTime;
    }

    /**
     * 做菜
     */
    public void make(){
        SmallTool.sleep(TimeUnit.SECONDS.toMillis(productionTime));
        SmallTool.print(name + "制作完毕 来吃我吧!");
    }
}

编写过程

@Test
public void testOne(){

    SmallTool.print("小白和小伙伴门 进餐厅点菜");

    long startTime = System.currentTimeMillis();

    ArrayList<Dish> dishes = new ArrayList<>();

    // 点菜
    for (int i = 1; i <= 10; i++) {
        dishes.add(new Dish("菜" + i, 1));
    }
    // 做菜
    dishes.forEach(dish -> CompletableFuture.runAsync(dish::make).join());

    SmallTool.print("菜都做好了, 上桌 " + (System.currentTimeMillis() - startTime));

}

执行结果

1649519841265    |    1    |    main    |    小白和小伙伴门 进餐厅点菜
1649519842281    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜1制作完毕 来吃我吧!
1649519843286    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜2制作完毕 来吃我吧!
1649519844294    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜3制作完毕 来吃我吧!
1649519845300    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜4制作完毕 来吃我吧!
1649519846307    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜5制作完毕 来吃我吧!
1649519847313    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜6制作完毕 来吃我吧!
1649519848319    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜7制作完毕 来吃我吧!
1649519849326    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜8制作完毕 来吃我吧!
1649519850333    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜9制作完毕 来吃我吧!
1649519851339    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜10制作完毕 来吃我吧!
1649519851343    |    1    |    main    |    菜都做好了, 上桌 10075

好像没什么问题, 但是这样的话, 一个一个调用join,硬是把多线程玩成了单线程~

代码改造

@Test
public void testTwo(){

    SmallTool.print("小白和小伙伴门 进餐厅点菜");

    long startTime = System.currentTimeMillis();

    ArrayList<Dish> dishes = new ArrayList<>();

    // 点菜
    for (int i = 1; i <= 10; i++) {
        dishes.add(new Dish("菜" + i, 1));
    }

    ArrayList<CompletableFuture<Void>> completableFutures = new ArrayList<>();

    // 做菜 将所有线程引用收集
    dishes.forEach(dish -> completableFutures.add(CompletableFuture.runAsync(dish::make)));

    // 将所有线程统一join
    CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join();

    SmallTool.print("菜都做好了, 上桌 " + (System.currentTimeMillis() - startTime));

}

执行结果

1649520172283    |    1    |    main    |    小白和小伙伴门 进餐厅点菜
1649520173305    |    30    |    ForkJoinPool.commonPool-worker-31    |    菜7制作完毕 来吃我吧!
1649520173305    |    27    |    ForkJoinPool.commonPool-worker-23    |    菜3制作完毕 来吃我吧!
1649520173305    |    32    |    ForkJoinPool.commonPool-worker-3    |    菜9制作完毕 来吃我吧!
1649520173305    |    28    |    ForkJoinPool.commonPool-worker-27    |    菜5制作完毕 来吃我吧!
1649520173305    |    26    |    ForkJoinPool.commonPool-worker-9    |    菜4制作完毕 来吃我吧!
1649520173305    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜2制作完毕 来吃我吧!
1649520173305    |    25    |    ForkJoinPool.commonPool-worker-5    |    菜1制作完毕 来吃我吧!
1649520173305    |    29    |    ForkJoinPool.commonPool-worker-13    |    菜6制作完毕 来吃我吧!
1649520173305    |    33    |    ForkJoinPool.commonPool-worker-21    |    菜10制作完毕 来吃我吧!
1649520173305    |    31    |    ForkJoinPool.commonPool-worker-17    |    菜8制作完毕 来吃我吧!
1649520173335    |    1    |    main    |    菜都做好了, 上桌 1049

哇咔咔, 不得了呀, 原本10秒的事情, 居然只用了一秒

使用Stream优化代码

@Test
public void testTwo(){

    SmallTool.print("小白和小伙伴门 进餐厅点菜");

    long startTime = System.currentTimeMillis();

    /*
        1: 生成1 - 10 的数字
        2: 创建10盘菜
        3: 提交runAsync 并且执行make
        4: 转换为数组
        5: 执行allOf
        6: 执行统一join
     */
    CompletableFuture.allOf(IntStream.range(1, 10)
            .mapToObj(i -> new Dish("菜" + i, 1))
            .map(dish -> CompletableFuture.runAsync(dish::make))
            .toArray(CompletableFuture[]::new)).join();

    SmallTool.print("菜都做好了, 上桌 " + (System.currentTimeMillis() - startTime));

}

瞬间一大片代码变成了一句, emmm, 我则么没有这么吊,当然执行结果是一样的

需求进化

如果小白现在突然想点20盘菜呢?

需求点:

任务巨多, 如何保证性能
如何观察任务, 调度情况
线程复用问题

实现

编写代码

其实就是将上一个例子的10改为20而已

执行结果

1649521040000    |    1    |    main    |    小白和小伙伴门 进餐厅点菜
1649521041025    |    37    |    ForkJoinPool.commonPool-worker-29    |    菜14制作完毕 来吃我吧!
1649521041025    |    30    |    ForkJoinPool.commonPool-worker-31    |    菜6制作完毕 来吃我吧!
1649521041025    |    36    |    ForkJoinPool.commonPool-worker-11    |    菜13制作完毕 来吃我吧!
1649521041025    |    27    |    ForkJoinPool.commonPool-worker-9    |    菜4制作完毕 来吃我吧!
1649521041025    |    34    |    ForkJoinPool.commonPool-worker-7    |    菜11制作完毕 来吃我吧!
1649521041025    |    31    |    ForkJoinPool.commonPool-worker-17    |    菜9制作完毕 来吃我吧!
1649521041025    |    35    |    ForkJoinPool.commonPool-worker-25    |    菜12制作完毕 来吃我吧!
1649521041025    |    24    |    ForkJoinPool.commonPool-worker-19    |    菜2制作完毕 来吃我吧!
1649521041025    |    28    |    ForkJoinPool.commonPool-worker-27    |    菜5制作完毕 来吃我吧!
1649521041025    |    25    |    ForkJoinPool.commonPool-worker-5    |    菜1制作完毕 来吃我吧!
1649521041025    |    26    |    ForkJoinPool.commonPool-worker-23    |    菜3制作完毕 来吃我吧!
1649521041025    |    29    |    ForkJoinPool.commonPool-worker-13    |    菜7制作完毕 来吃我吧!
1649521041025    |    38    |    ForkJoinPool.commonPool-worker-15    |    菜15制作完毕 来吃我吧!
1649521041025    |    33    |    ForkJoinPool.commonPool-worker-21    |    菜10制作完毕 来吃我吧!
1649521041025    |    32    |    ForkJoinPool.commonPool-worker-3    |    菜8制作完毕 来吃我吧!
1649521042040    |    30    |    ForkJoinPool.commonPool-worker-31    |    菜16制作完毕 来吃我吧!
1649521042040    |    28    |    ForkJoinPool.commonPool-worker-27    |    菜18制作完毕 来吃我吧!
1649521042040    |    26    |    ForkJoinPool.commonPool-worker-23    |    菜19制作完毕 来吃我吧!
1649521042040    |    25    |    ForkJoinPool.commonPool-worker-5    |    菜17制作完毕 来吃我吧!
1649521042048    |    1    |    main    |    菜都做好了, 上桌 2040

可以看的出来, 执行的线程重复了, 用时2ms, 为什么呢? 核心池的最大是15, 应为这个和你电脑的CPU核心数有关, 我电脑是8核16线程的, ForkJoinPool的最大线程数默认应该是最大线程数-1

我们看一下

@Test
public void testForkJoinPool(){
    // 电脑支持的最大线程数
    System.out.println(Runtime.getRuntime().availableProcessors());
    // 通用池 当前大小
    System.out.println(ForkJoinPool.commonPool().getPoolSize());
    // 通用池最大线程数
    System.out.println(ForkJoinPool.getCommonPoolParallelism());
}

结果

16
0
15

这个时候就需要用到线程池来解决了,之前CompeletableFuture的方法总结中带Async的后缀的方法, 其实都是可以多传入一个参数的,那就是指定线程池, 如果不指定,默认使用的线程池就是ForkJoinPool.commonPool从名字也可以看出,这是ForkJoin的池

改进一

将通用池的线程数设置为合适大小

@Test
public void testThree(){

    // -Djava.util.concurrent.ForkJoinPool.common.parallelism=20
    System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");

    SmallTool.print("小白和小伙伴门 进餐厅点菜");

    long startTime = System.currentTimeMillis();

    CompletableFuture.allOf(IntStream.range(1, 20)
            .mapToObj(i -> new Dish("菜" + i, 1))
            .map(dish -> CompletableFuture.runAsync(dish::make))
            .toArray(CompletableFuture[]::new)).join();

    SmallTool.print("菜都做好了, 上桌 " + (System.currentTimeMillis() - startTime));

}

执行结果

1649522125623    |    1    |    main    |    小白和小伙伴门 进餐厅点菜
1649522126645    |    35    |    ForkJoinPool.commonPool-worker-25    |    菜12制作完毕 来吃我吧!
1649522126645    |    34    |    ForkJoinPool.commonPool-worker-39    |    菜11制作完毕 来吃我吧!
1649522126645    |    28    |    ForkJoinPool.commonPool-worker-59    |    菜5制作完毕 来吃我吧!
1649522126645    |    41    |    ForkJoinPool.commonPool-worker-5    |    菜18制作完毕 来吃我吧!
1649522126645    |    29    |    ForkJoinPool.commonPool-worker-45    |    菜6制作完毕 来吃我吧!
1649522126645    |    40    |    ForkJoinPool.commonPool-worker-19    |    菜17制作完毕 来吃我吧!
1649522126645    |    24    |    ForkJoinPool.commonPool-worker-51    |    菜2制作完毕 来吃我吧!
1649522126645    |    26    |    ForkJoinPool.commonPool-worker-9    |    菜4制作完毕 来吃我吧!
1649522126645    |    25    |    ForkJoinPool.commonPool-worker-37    |    菜1制作完毕 来吃我吧!
1649522126645    |    37    |    ForkJoinPool.commonPool-worker-61    |    菜14制作完毕 来吃我吧!
1649522126645    |    27    |    ForkJoinPool.commonPool-worker-23    |    菜3制作完毕 来吃我吧!
1649522126645    |    33    |    ForkJoinPool.commonPool-worker-53    |    菜10制作完毕 来吃我吧!
1649522126645    |    32    |    ForkJoinPool.commonPool-worker-3    |    菜9制作完毕 来吃我吧!
1649522126645    |    42    |    ForkJoinPool.commonPool-worker-55    |    菜19制作完毕 来吃我吧!
1649522126645    |    36    |    ForkJoinPool.commonPool-worker-11    |    菜13制作完毕 来吃我吧!
1649522126645    |    30    |    ForkJoinPool.commonPool-worker-31    |    菜7制作完毕 来吃我吧!
1649522126645    |    38    |    ForkJoinPool.commonPool-worker-47    |    菜15制作完毕 来吃我吧!
1649522126645    |    39    |    ForkJoinPool.commonPool-worker-33    |    菜16制作完毕 来吃我吧!
1649522126645    |    31    |    ForkJoinPool.commonPool-worker-17    |    菜8制作完毕 来吃我吧!
1649522126649    |    1    |    main    |    菜都做好了, 上桌 1023

可以看到,又回到1ms了, 但是这个值到底要设置为多少才合适呢?

答案是都不合适

原因

从名字可以看出ForkJoinPool, 显然这个池并不只为CompeletableFuture服务
只有在启动之前,初始化的时候才可以设置
需要从项目的长期使用量才可以得出

改进二(推荐)

自定义线程池, 这个时候,就又说会上面的第二个参数了,没错 ,那就是Executor

为什么推荐呢?

原因

隔离, 防止影响其他使用ForkJoinPool的代码
方便控制, ForkJoinPool在初始化后, 不可以修改, 但是自定义的线程池可以在任务数量来之后, 通过计算得出线程的数量

这里采用无上限线程池演示

@Test
public void testFour(){

    SmallTool.print("小白和小伙伴门 进餐厅点菜");

    long startTime = System.currentTimeMillis();

    // 创建线程池
    final ExecutorService threadPool = Executors.newCachedThreadPool();

    CompletableFuture.allOf(IntStream.range(1, 20)
            .mapToObj(i -> new Dish("菜" + i, 1))
            .map(dish -> CompletableFuture.runAsync(dish::make, threadPool))
            .toArray(CompletableFuture[]::new)).join();

    // 销毁
    threadPool.shutdown();

    SmallTool.print("菜都做好了, 上桌 " + (System.currentTimeMillis() - startTime));

}

执行结果

1649522607939    |    1    |    main    |    小白和小伙伴门 进餐厅点菜
1649522608962    |    42    |    pool-1-thread-19    |    菜19制作完毕 来吃我吧!
1649522608963    |    26    |    pool-1-thread-3    |    菜3制作完毕 来吃我吧!
1649522608963    |    25    |    pool-1-thread-2    |    菜2制作完毕 来吃我吧!
1649522608963    |    29    |    pool-1-thread-6    |    菜6制作完毕 来吃我吧!
1649522608963    |    31    |    pool-1-thread-8    |    菜8制作完毕 来吃我吧!
1649522608963    |    24    |    pool-1-thread-1    |    菜1制作完毕 来吃我吧!
1649522608963    |    30    |    pool-1-thread-7    |    菜7制作完毕 来吃我吧!
1649522608963    |    39    |    pool-1-thread-16    |    菜16制作完毕 来吃我吧!
1649522608962    |    40    |    pool-1-thread-17    |    菜17制作完毕 来吃我吧!
1649522608963    |    28    |    pool-1-thread-5    |    菜5制作完毕 来吃我吧!
1649522608962    |    27    |    pool-1-thread-4    |    菜4制作完毕 来吃我吧!
1649522608962    |    38    |    pool-1-thread-15    |    菜15制作完毕 来吃我吧!
1649522608962    |    34    |    pool-1-thread-11    |    菜11制作完毕 来吃我吧!
1649522608962    |    32    |    pool-1-thread-9    |    菜9制作完毕 来吃我吧!
1649522608962    |    37    |    pool-1-thread-14    |    菜14制作完毕 来吃我吧!
1649522608962    |    33    |    pool-1-thread-10    |    菜10制作完毕 来吃我吧!
1649522608962    |    36    |    pool-1-thread-13    |    菜13制作完毕 来吃我吧!
1649522608962    |    41    |    pool-1-thread-18    |    菜18制作完毕 来吃我吧!
1649522608962    |    35    |    pool-1-thread-12    |    菜12制作完毕 来吃我吧!
1649522608973    |    1    |    main    |    菜都做好了, 上桌 1028