Java web

CompletableFuture：让你的代码免受阻塞之苦

2019-12-19  本文已影响0人  何甜甜在吗

提高应用性能的时候很容易就会想起异步，异步去处理一些任务这样主线程可以尽快响应。

写在前面

通过阅读本篇文章你将了解到：

场景说明

查询所有商店某个商品的价格并返回，并且查询商店某个商品的价格的API为同步
一个Shop类，提供一个名为getPrice的同步方法

public class Shop {
    private Random random = new Random();
    /**
     * 根据产品名查找价格
     * */
    public double getPrice(String product) {
        return calculatePrice(product);
    }

    /**
     * 计算价格
     *
     * @param product
     * @return
     * */
    private double calculatePrice(String product) {
        delay();
        //random.nextDouble()随机返回折扣
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    /**
     * 通过睡眠模拟其他耗时操作
     * */
    private void delay() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

查询商品的价格为同步方法，并通过sleep方法模拟其他操作。这个场景模拟了当需要调用第三方API，但第三方提供的是同步API，在无法修改第三方API时如何设计代码调用提高应用的性能和吞吐量，这时候可以使用CompletableFuture类

CompletableFuture使用

Completable是Future接口的实现类，在JDK1.8中引入

同步方法Pick异步方法查询所有店铺某个商品价格

店铺为一个列表：

private static List<Shop> shopList = Arrays.asList(
        new Shop("BestPrice"),
        new Shop("LetsSaveBig"),
        new Shop("MyFavoriteShop"),
        new Shop("BuyItAll")
);

同步方法：

private static List<String> findPriceSync(String product) {
    return shopList.stream()
            .map(shop -> String.format("%s price is %.2f",
                    shop.getName(), shop.getPrice(product)))  //格式转换
            .collect(Collectors.toList());
}

异步方法：

private static List<String> findPriceAsync(String product) {
    List<CompletableFuture<String>> completableFutureList = shopList.stream()
            //转异步执行
            .map(shop -> CompletableFuture.supplyAsync(
                    () -> String.format("%s price is %.2f",
                            shop.getName(), shop.getPrice(product))))  //格式转换
            .collect(Collectors.toList());

    return completableFutureList.stream()
            .map(CompletableFuture::join)  //获取结果不会抛出异常
            .collect(Collectors.toList());
}

性能测试结果：

Find Price Sync Done in 4141
Find Price Async Done in 1033

异步执行效率提高四倍

为什么仍需要CompletableFuture

在JDK1.8以前，通过调用线程池的submit方法可以让任务以异步的方式运行，该方法会返回一个Future对象，通过调用get方法获取异步执行的结果：

private static List<String> findPriceFutureAsync(String product) {
    ExecutorService es = Executors.newCachedThreadPool();
    List<Future<String>> futureList = shopList.stream().map(shop -> es.submit(() -> String.format("%s price is %.2f",
            shop.getName(), shop.getPrice(product)))).collect(Collectors.toList());

    return futureList.stream()
            .map(f -> {
                String result = null;
                try {
                    result = f.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

                return result;
            }).collect(Collectors.toList());
}

既生瑜何生亮，为什么仍需要引入CompletableFuture？
对于简单的业务场景使用Future完全没有，但是想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务的值等等，使用Future提供的那点API就囊中羞涩，处理起来不够优雅，这时候还是让CompletableFuture以声明式的方式优雅的处理这些需求

其他API介绍

whenComplete计算结果的处理：

对前面计算结果进行处理，无法返回新值
提供了三个方法：

//方法一
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
//方法二
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
//方法三
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

说明：

示例：

public class WhenCompleteTest {
    public static void main(String[] args) {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "hello");
        CompletableFuture<String> cf2 = cf1.whenComplete((v, e) ->
                System.out.println(String.format("value:%s, exception:%s", v, e)));
        System.out.println(cf2.join());
    }
}

thenApply转换：

将前面计算结果的的CompletableFuture传递给thenApply，返回thenApply处理后的结果。可以认为通过thenApply方法实现CompletableFuture<T>至CompletableFuture<U>的转换。白话一点就是将CompletableFuture的计算结果作为thenApply方法的参数，返回thenApply方法处理后的结果
提供了三个方法：

//方法一
public <U> CompletableFuture<U> thenApply(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(null, fn);
}

//方法二
public <U> CompletableFuture<U> thenApplyAsync(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(asyncPool, fn);
}

//方法三
public <U> CompletableFuture<U> thenApplyAsync(
    Function<? super T,? extends U> fn, Executor executor) {
    return uniApplyStage(screenExecutor(executor), fn);
}

说明：

public class ThenApplyTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenApplyTest::randomInteger).thenApply((i) -> i * 8);
        System.out.println(result.get());
    }

    public static Integer randomInteger() {
        return 10;
    }
}

这里将前一个CompletableFuture计算出来的结果扩大八倍

thenAccept结果处理：

thenApply也可以归类为对结果的处理，thenAccept和thenApply的区别就是没有返回值
提供了三个方法：

//方法一
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
    return uniAcceptStage(null, action);
}

//方法二
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
    return uniAcceptStage(asyncPool, action);
}

//方法三
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
                                               Executor executor) {
    return uniAcceptStage(screenExecutor(executor), action);
}

说明：

public class ThenAcceptTest {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(ThenAcceptTest::getList).thenAccept(strList -> strList.stream()
                .forEach(m -> System.out.println(m)));
    }

    public static List<String> getList() {
        return Arrays.asList("a", "b", "c");
    }
}

将前一个CompletableFuture计算出来的结果打印出来

thenCompose异步结果流水化：

thenCompose方法可以将两个异步操作进行流水操作
提供了三个方法：

//方法一
public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(null, fn);
}

//方法二
public <U> CompletableFuture<U> thenComposeAsync(
    Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(asyncPool, fn);
}

//方法三
public <U> CompletableFuture<U> thenComposeAsync(
    Function<? super T, ? extends CompletionStage<U>> fn,
    Executor executor) {
    return uniComposeStage(screenExecutor(executor), fn);
}

说明：

public class ThenComposeTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenComposeTest::getInteger)
                .thenCompose(i -> CompletableFuture.supplyAsync(() -> i * 10));
        System.out.println(result.get());
    }

    private static int getInteger() {
        return 666;
    }

    private static int expandValue(int num) {
        return num * 10;
    }
}

执行流程图：


thenCombine组合结果：

thenCombine方法将两个无关的CompletableFuture组合起来，第二个Completable并不依赖第一个Completable的结果
提供了三个方法：

//方法一
public <U,V> CompletableFuture<V> thenCombine( 
    CompletionStage<? extends U> other,
    BiFunction<? super T,? super U,? extends V> fn) {
    return biApplyStage(null, other, fn);
}
  //方法二
  public <U,V> CompletableFuture<V> thenCombineAsync(
      CompletionStage<? extends U> other,
      BiFunction<? super T,? super U,? extends V> fn) {
      return biApplyStage(asyncPool, other, fn);
  }

  //方法三
  public <U,V> CompletableFuture<V> thenCombineAsync(
      CompletionStage<? extends U> other,
      BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
      return biApplyStage(screenExecutor(executor), other, fn);
  }

说明：

示例：

public class ThenCombineTest {
    private static Random random = new Random();
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(ThenCombineTest::randomInteger).thenCombine(
                CompletableFuture.supplyAsync(ThenCombineTest::randomInteger), (i, j) -> i * j
        );

        System.out.println(result.get());
    }

    public static Integer randomInteger() {
        return random.nextInt(100);
    }
}

将两个线程计算出来的值做一个乘法在返回
执行流程图：


allOf&anyOf组合多个CompletableFuture：

方法介绍：

//allOf
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
    return andTree(cfs, 0, cfs.length - 1);
}
//anyOf
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
    return orTree(cfs, 0, cfs.length - 1);
}

说明：

示例：

注意点

很多方法都提供了异步实现【带async后缀】，但是需小心谨慎使用这些异步方法，因为异步意味着存在上下文切换，可能性能不一定比同步好。如果需要使用异步的方法，先做测试，用测试数据说话！！！

CompletableFuture的应用场景

存在IO密集型的任务可以选择CompletableFuture，IO部分交由另外一个线程去执行。Logback、Log4j2异步日志记录的实现原理就是新起了一个线程去执行IO操作，这部分可以以CompletableFuture.runAsync(()->{ioOperation();})的方式去调用，有关Logback异步日志记录的原理可以参考这篇文章Logback异步日志记录

优化空间

supplyAsync执行任务底层实现：

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}
static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<U> d = new CompletableFuture<U>();
    e.execute(new AsyncSupply<U>(d, f));
    return d;
}

底层调用的是线程池去执行任务，而CompletableFuture中默认线程池为ForkJoinPool

private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

ForkJoinPool线程池的大小取决于CPU的核数。之前写的为什么阿里巴巴要禁用Executors创建线程池？文章中提及过，CPU密集型任务线程池大小配置为CPU核心数就可以了，但是IO密集型，线程池的大小由CPU数量 * CPU利用率 * (1 + 线程等待时间/线程CPU时间)确定。而CompletableFuture的应用场景就是IO密集型任务，因此默认的ForkJoinPool一般无法达到最佳性能，我们需自己根据业务创建线程池

最后附：示例代码，欢迎Fork与Star

附往期文章：欢迎你的阅读、点赞、评论

并发相关
1.为什么阿里巴巴要禁用Executors创建线程池？
2.自己的事情自己做，线程异常处理

设计模式相关：
1. 单例模式，你真的写对了吗？
2. (策略模式+工厂模式+map)套餐 Kill 项目中的switch case

JAVA8相关：
1. 使用Stream API优化代码
2. 亲，建议你使用LocalDateTime而不是Date哦

数据库相关：
1. mysql数据库时间类型datetime、bigint、timestamp的查询效率比较
2. 很高兴！终于踩到了慢查询的坑

高效相关：
1. 撸一个Java脚手架，一统团队项目结构风格

日志相关：
1. 日志框架，选择Logback Or Log4j2？
2. Logback配置文件这么写，TPS提高10倍

工程相关：
1. 闲来无事，动手写一个LRU本地缓存
2. Redis实现点赞功能模块
3. JMX可视化监控线程池
4. 权限管理 【SpringSecurity篇】
5. Spring自定义注解从入门到精通
6. java模拟登陆优酷
7. QPS这么高，那就来写个多级缓存吧
8. java使用phantomjs进行截图

其他：
1. 使用try-with-resources优雅关闭资源
2. 老板，用float存储金额为什么要扣我工资

上一篇 下一篇

猜你喜欢

热点阅读