Java8 CompletableFuture组合式的编程(笔记)

    * 实现异步API


public double getPrice(String product) {
    return calculatePrice(product);
}

/**
 * 同步计算商品价格的方法
 *
 * @param product 商品名称
 * @return 价格
 */
private double calculatePrice(String product) {
    delay();
    return random.nextDouble() * product.charAt(0) + product.charAt(1);
}
/**
 * 模拟计算,查询数据库等耗时
 */
public static void delay() {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

    * 将同步方法装换为异步方法

/**
 * 异步计算商品的价格.
 *
 * @param product 商品名称
 * @return 价格
 */
public Future<Double> getPriceAsync(String product) {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        double price = calculatePrice(product);
        futurePrice.complete(price);
    }).start();
    return futurePrice;
}

使用异步API 模拟客户端
Shop shop = new Shop("BestShop");
long start = System.nanoTime();
Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
long incocationTime = (System.nanoTime() - start) / 1_000_000;
System.out.println("执行时间:" + incocationTime + " msecs");
try {
    Double price = futurePrice.get();
    System.out.printf("Price is %.2f%n", price);
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}
long retrievalTime = (System.nanoTime() - start) / 1_000_000;
System.out.println("retrievalTime:" + retrievalTime + " msecs");

* 错误处理

     上述代码,如果没有意外,可以正常工作,但是如果价格计算过程中生产了错误会怎样呢?非常不幸,这种情况下你会得到一个相当糟糕的结果:用于提示错误的异常会限制在视图计算商品的价格的当前线程的范围内,最终会杀死该线程,而这会导致等待get方法放回结果的客户端永久的被阻塞,
     客户端可以使用重载的get方法,它给定了一个超时时间,这是一种推荐做法!
     为了让客户端能了解商店无法提供请求商品价格的原因.我们对代码优化,!
/**
 * 异步计算商品的价格.
 *
 * @param product 商品名称
 * @return 价格
 */
public Future<Double> getPriceAsync(String product) {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        try {
            double price = calculatePrice(product);
            futurePrice.complete(price);
        } catch (Exception e) {
            //否则就抛出异常,完成这次future操作
            futurePrice.completeExceptionally(e);
        }
    }).start();
    return futurePrice;
}


    * 使用工厂方法supplyAsync创建CompletableFuture

/**
 * 异步计算商品的价格.
 *
 * @param product 商品名称
 * @return 价格
 */
public Future<Double> getPriceAsync(String product) {
   /* CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        try {
            double price = calculatePrice(product);
            futurePrice.complete(price);
        } catch (Exception e) {
            //否则就抛出异常,完成这次future操作
            futurePrice.completeExceptionally(e);
        }
    }).start();
    return futurePrice;*/

    return CompletableFuture.supplyAsync(() -> calculatePrice(product));
}


    * 让代码免受阻塞之苦

案例:最佳价格查询器
private static List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
        new Shop(":LetsSaveBig"),
        new Shop("MyFavoriteShop"),
        new Shop("BuyItAll"));

/**
 * 最佳价格查询器
 *
 * @param product 商品
 * @return
 */
public static List<String> findprices(String product) {
    return shops
            .stream()
            .map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
            .collect(Collectors.toList());
}

验证findprices的正确性和执行性能
long start = System.nanoTime();
System.out.println(findprices("myPhones27s"));
long duration = (System.nanoTime() - start) / 1_000_000;
System.out.println("Done in " + duration+" msecs");

执行结果:

* 使用平行流对请求进行并行操作

/**
 * 最佳价格查询器(并行流)
 *
 * @param product 商品
 * @return
 */
public static List<String> parallelFindprices(String product) {
    return shops
            .parallelStream()
            .map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
            .collect(Collectors.toList());
}

同样的测试代码:

相当不错,看起来这是个简单有效的主意,对4个不同商店的查询实现了并行.所有完成操作的总耗时只有1秒多一点,让我们尝试使用CompletableFuture,将findprices方法中对不同商店的同步调用替换为异步调用.

    * 使用CompletableFuture发起异步请求

/**
 * 最佳价格查询器(异步调用实现)
 * @param product 商品
 * @return
 */
public static List<String> asyncFindprices(String product) {
    //使用这种方式,你会得到一个List<CompletableFuture<String>>,列表中的每一个CompletableFuture对象在计算完成后都包含商店的String类型的名称.
    //但是,由于你用CompletableFuture实现了asyncFindprices方法要求返回一个List<String>.你需要等待所有的future执行完毕,将其包含的值抽取出来,填充到列表中才能返回
    List<CompletableFuture<String>> priceFuture = shops
            .stream()
            .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))))
            .collect(Collectors.toList());
    //为了实现这个效果,我门可以向最初的List<CompletableFuture<String>>施加第二个map操作,对list中的每一个future对象执行join操作,一个接一个地等待他们允许结束,join和get方法
    //有相同的含义,不同的在于join不会抛出任何检测到的异常
    return priceFuture
            .stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
}

相同的测试代码:

结果让我们失望了.我们采用异步调用新版方法,要比并行流慢了一倍.

    * 寻找更好的方案

经过我增加商店数量,然后使用三种方式反复的测试,发现了一个问题,并行流和异步调用的性能不分伯仲,究其原因都一样,它们内部采用的是同样的通用线程池,默认都使用固定数目的线程,具体线程数取决于Runtime.getRuntime.availableProcessors()放回值,然而,.CompletableFuture具有一定的优势,因为它允许你对执行器进行配置,尤其是线程池的大小,让它以适合应用需求的方式进行配置,满足程序的要求,而这是并行流API无法提供的.

    * 使用定制的执行器

private static final Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100));
/**
 * 最佳价格查询器(异步调用实现,自定义执行器)
 *
 * @param product 商品
 * @return
 */
public static List<String> asyncFindpricesThread(String product) {
    List<CompletableFuture<String>> priceFuture = shops
            .stream()
            .map(shop -> CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor))
            .collect(Collectors.toList());
    return priceFuture
            .stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
}

经过测试处理5个商店 是1秒多,处理9个商店也是1秒多

并行--使用流还是CompletableFutures?
     目前为止,我们已经知道对集合进行并行计算有两种方式,要么将其转化为并行流,利用map这样的操作开展工作,要么枚举出集合中的每一个元素,创建新的线程,在CompletableFuture内对其进行操作,后者提供了更多的灵活性,你可以调整线程池大小,二者能帮助你确保整体计算机不会因为线程都在等待I/O而发生阻塞
     我们使用这些API的建议如下:

    1. 如果你进行的是计算密集型的操作,并且没有I/O,那么推荐使用Stream接口,因为实现简单,同时效率也可能是最高的
    2. 反之,如果你并行的工作单元还涉及等待I/O的操作(包括网络连接等待).那么使用CompletableFuture是灵活性更好,你可以像前面讨论的那样,依据等待/计算,或者W/C的比率设定需要使用的线程数,