首页 > 其他分享 >学会 CompletableFuture：让你的代码免受阻塞之苦！

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

时间：2023-10-23 17:26:20浏览次数：57  
标签：线程 return 之苦 免受 CompletableFuture static 方法 public

来源：https://juejin.cn/post/6844904024332828685

写在前面

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

  • CompletableFuture的使用
  • CompletableFure异步和同步的性能测试
  • 已经有了Future为什么仍需要在JDK1.8中引入CompletableFuture
  • CompletableFuture的应用场景
  • 对CompletableFuture的使用优化

场景说明

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

  • 店铺类：Shop.java
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类。

推荐一个开源免费的 Spring Boot 实战项目：

https://github.com/javastacks/spring-boot-best-practice

CompletableFuture使用

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

  • CompletableFuture的创建：

    说明：

    • 两个重载方法之间的区别 => 后者可以传入自定义Executor，前者是默认的，使用的ForkJoinPool

    • supplyAsync和runAsync方法之间的区别 => 前者有返回值，后者无返回值

    • Supplier是函数式接口，因此该方法需要传入该接口的实现类，追踪源码会发现在run方法中会调用该接口的方法。因此使用该方法创建CompletableFuture对象只需重写Supplier中的get方法，在get方法中定义任务即可。又因为函数式接口可以使用Lambda表达式，和new创建CompletableFuture对象相比代码会简洁不少

    • 使用new方法

      CompletableFuture<Double> futurePrice = new CompletableFuture<>();

    • 使用CompletableFuture#completedFuture静态方法创建

      public static <U> CompletableFuture<U> completedFuture(U value) {
    return new CompletableFuture<U>((value == null) ? NIL : value);
}

      参数的值为任务执行完的结果，一般该方法在实际应用中较少应用

    • 使用 CompletableFuture#supplyAsync静态方法创建 supplyAsync有两个重载方法：

      //方法一
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}
//方法二
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                   Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

    • 使用CompletableFuture#runAsync静态方法创建 runAsync有两个重载方法

      //方法一
public static CompletableFuture<Void> runAsync(Runnable runnable) {
    return asyncRunStage(asyncPool, runnable);
}
//方法二
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

  • 结果的获取： 对于结果的获取CompltableFuture类提供了四种方式

    //方式一
public T get()
//方式二
public T get(long timeout, TimeUnit unit)
//方式三
public T getNow(T valueIfAbsent)
//方式四
public T join()

    说明：

    示例：

    • get()和get(long timeout, TimeUnit unit) => 在Future中就已经提供了，后者提供超时处理，如果在指定时间内未获取结果将抛出超时异常
    • getNow => 立即获取结果不阻塞，结果计算已完成将返回结果或计算过程中的异常，如果未计算完成将返回设定的valueIfAbsent值
    • join => 方法里不会抛出异常
public class AcquireResultTest {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
      //getNow方法测试
      CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
          try {
              Thread.sleep(60 * 1000 * 60 );
          } catch (InterruptedException e) {
              e.printStackTrace();
          }

          return "hello world";
      });

      System.out.println(cp1.getNow("hello h2t"));

      //join方法测试
      CompletableFuture<Integer> cp2 = CompletableFuture.supplyAsync((()-> 1 / 0));
      System.out.println(cp2.join());

      //get方法测试
      CompletableFuture<Integer> cp3 = CompletableFuture.supplyAsync((()-> 1 / 0));
      System.out.println(cp3.get());
  }
}

说明：

  • 第一个执行结果为hello h2t，因为要先睡上1分钟结果不能立即获取
  • join方法获取结果方法里不会抛异常，但是执行结果会抛异常，抛出的异常为CompletionException
  • get方法获取结果方法里将抛出异常，执行结果抛出的异常为ExecutionException
  • 异常处理： 使用静态方法创建的CompletableFuture对象无需显示处理异常，使用new创建的对象需要调用completeExceptionally方法设置捕获到的异常，举例说明：
CompletableFuture completableFuture = new CompletableFuture();
new Thread(() -> {
   try {
       //doSomething，调用complete方法将其他方法的执行结果记录在completableFuture对象中
       completableFuture.complete(null);
   } catch (Exception e) {
       //异常处理
       completableFuture.completeExceptionally(e);
    }
}).start();

同步方法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以声明式的方式优雅的处理这些需求。而且在Future编程中想要拿到Future的值然后拿这个值去做后续的计算任务，只能通过轮询的方式去判断任务是否完成这样非常占CPU并且代码也不优雅，用伪代码表示如下：

while(future.isDone()) {
    result = future.get();
    doSomrthingWithResult(result);
}

但CompletableFuture提供了API帮助我们实现这样的需求

其他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)

说明：

  • BiFunction<? super T,? super U,? extends V> fn参数 => 定义对结果的处理
  • Executor executor参数 => 自定义线程池
  • 以async结尾的方法将会在一个新的线程中执行组合操作

示例：

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);
}

说明：

  • Function<? super T,? extends U> fn参数 => 对前一个CompletableFuture 计算结果的转化操作
  • Executor executor参数 => 自定义线程池
  • 以async结尾的方法将会在一个新的线程中执行组合操作 示例：
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);
}

说明：

  • Consumer<? super T> action参数 => 对前一个CompletableFuture计算结果的操作
  • Executor executor参数 => 自定义线程池
  • 同理以async结尾的方法将会在一个新的线程中执行组合操作 示例：
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);
}

说明：

  • Function<? super T, ? extends CompletionStage<U>> fn参数 => 当前CompletableFuture计算结果的执行
  • Executor executor参数 => 自定义线程池
  • 同理以async结尾的方法将会在一个新的线程中执行组合操作 示例：
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);
  }

说明：

  • CompletionStage<? extends U> other参数 => 新的CompletableFuture的计算结果
  • BiFunction<? super T,? super U,? extends V> fn参数 => 定义了两个CompletableFuture对象完成计算后如何合并结果，该参数是一个函数式接口，因此可以使用Lambda表达式
  • Executor executor参数 => 自定义线程池
  • 同理以async结尾的方法将会在一个新的线程中执行组合操作

示例：

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);
}

说明：

  • allOf => 所有的CompletableFuture都执行完后执行计算。
  • anyOf => 任意一个CompletableFuture执行完后就会执行计算

示例：

  • allOf方法测试
public class AllOfTest {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
      CompletableFuture<Void> future1 = CompletableFuture.supplyAsync(() -> {
          System.out.println("hello");
          return null;
      });
      CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
          System.out.println("world"); return null;
      });
      CompletableFuture<Void> result = CompletableFuture.allOf(future1, future2);
      System.out.println(result.get());
  }
}

allOf方法没有返回值，适合没有返回值并且需要前面所有任务执行完毕才能执行后续任务的应用场景

  • anyOf方法测试
public class AnyOfTest {
  private static Random random = new Random();
  public static void main(String[] args) throws ExecutionException, InterruptedException {
      CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
          randomSleep();
          System.out.println("hello");
          return "hello";});
      CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
          randomSleep();
          System.out.println("world");
          return "world";
      });
      CompletableFuture<Object> result = CompletableFuture.anyOf(future1, future2);
      System.out.println(result.get());
 }

  private static void randomSleep() {
      try {
          Thread.sleep(random.nextInt(10));
      } catch (InterruptedException e) {
          e.printStackTrace();
      }
  }
}

两个线程都会将结果打印出来，但是get方法只会返回最先完成任务的结果。该方法比较适合只要有一个返回值就可以继续执行其他任务的应用场景

注意点

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

CompletableFuture的应用场景

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

优化空间

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的核数。CPU密集型任务线程池大小配置为CPU核心数就可以了，但是IO密集型，线程池的大小由CPU数量 * CPU利用率 * (1 + 线程等待时间/线程CPU时间)确定。而CompletableFuture的应用场景就是IO密集型任务，因此默认的ForkJoinPool一般无法达到最佳性能，我们需自己根据业务创建线程池。

近期热文推荐：

1.1,000+ 道 Java面试题及答案整理(2022最新版)

2.劲爆！Java 协程要来了。。。

3.Spring Boot 2.x 教程，太全了！

4.别再写满屏的爆爆爆炸类了，试试装饰器模式，这才是优雅的方式！！

5.《Java开发手册（嵩山版）》最新发布，速速下载！

觉得不错，别忘了随手点赞+转发哦！

标签：线程,return,之苦,免受,CompletableFuture,static,方法,public
From： https://www.cnblogs.com/javastack/p/17782960.html

相关文章

  • 滚珠螺杆应如何存放避免受损
    滚珠螺杆是一种高精度的机械零件，保存或使用不当，会直接损坏，影响生产效率，因此我们在使用时需要注意以下事项：1、避免垂直放置：没有施加预压的螺杆垂直放置时，螺母会因自重而从螺杆轴上脱荐下来，因此使用时要注意避免垂直放置。如果无法避免垂直放置，可以参考垂直使用方法进行操作。2、注意......
  • CompletableFuture异步优化代码
    CompletableFuture异步编排优化代码我们在项目开发中，有可能遇到一个接口需要调用N个服务的接口。比如用户请求获取订单信息，需要调用用户信息、商品信息、物流信息等接口，最后再汇总数据统一返回。如果使用串行的方法按照顺序挨个调用接口，这样接口的响应的速度就很慢。如果并行调用......
  • 研发必会-异步编程利器之CompletableFuture(含源码 中)
    微信公众号访问地址：研发必会-异步编程利器之CompletableFuture(含源码中)近期热推文章：    1、springBoot对接kafka,批量、并发、异步获取消息,并动态、批量插入库表;    2、SpringBoot用线程池ThreadPoolTaskExecutor异步处理百万级数据;    3、基于Redis的Geo实现附......
  • 保护您的Shopify站点免受封禁和关联
    Shopify是一种流行的电商平台，但要确保您的Shopify站点不被封禁和关联，需要采取一些预防措施。本文将介绍一些方法，帮助您保护和维护您的Shopify站点的稳定性和安全性。一、避免被封禁的方法1. 遵守平台政策：Shopify有一系列规定和政策，包括禁止销售某些商品的规定、反垃圾邮件政策等。......
  • 并发_CompletableFuture
              ......
  • 并发编程系列-CompletableFuture
    利用多线程来提升性能，实质上是将顺序执行的操作转化为并行执行。仔细观察后，你还会发现在顺序转并行的过程中，一定会牵扯到异步化。举个例子，现在下面这段示例代码是按顺序执行的，为了优化性能，我们需要将其改为并行执行。那具体的实施方法是什么呢？//以下两个方法都是耗时操作doBizA();......
  • 从 5s 到 0.5s！CompletableFuture 异步任务优化技巧，确实优雅！
    一个接口可能需要调用N个其他服务的接口，这在项目开发中还是挺常见的。举个例子：用户请求获取订单信息，可能需要调用用户信息、商品详情、物流信息、商品推荐等接口，最后再汇总数据统一返回。如果是串行（按顺序依次执行每个任务）执行的话，接口的响应速度会非常慢。考虑到这些接口之间......
  • CompletableFuture 异步多线程D优雅！
    一个示例回顾Future一些业务场景我们需要使用多线程异步执行任务，加快任务执行速度。JDK5新增了Future接口，用于描述一个异步计算的结果。虽然Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用Future.get()的方式阻塞调用线程，或者使用轮......
  • 多账号管理：如何保护多个Wish店铺免受关联风险
    在如今的电商市场上，开设多个Wish店铺已成为许多商家实现成功的策略之一。Wish作为全球知名的电商平台，拥有庞大的用户群体，为商家提供了一个广阔的销售平台。然而，随之而来的是开店铺的风险，特别是关联风险。在这篇博客中，将分享一些方法和技巧，帮助商家提高Wish店铺的销量，并保护多个店铺......
  • 解决代码使用CompletableFuture做异步时spring-cloud-starter-sleuth的日志追踪号为空
    产生问题原因就是异步调用，导致spanId和traceId丢失了@Async注解的异步调用是没问题的前提使用spring-cloud-starter-sleuthjar包版本2.2.8.RELEASE关于追踪号的xml配置为<pattern>%yellow(%date{yyyy-MM-ddHH:mm:ss.SSS})[%X{X-B3-TraceId:-},%X{X-B3-SpanId:-},%X{X-B......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99