Future接口

Java5引入Future接口，它的设计初衷是对将来某个时刻会发生的结果进行建模。获取另一个线程中的计算结果。

public static void main(String[] args) {
    ExecutorService executor = Executors.newCachedThreadPool();
    Future<Double> future = executor.submit(new Callable<Double>() {
        @Override
        public Double call() throws Exception {
            return doSomeLongComputation();
        }
    });

    doSomethingElse();

    try {
        Double result = future.get(1, TimeUnit.SECONDS);
        System.out.println("result:" + result);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    } catch (TimeoutException e) {
        e.printStackTrace();
    }

    executor.shutdown();
}

Future接口的局限性

1、建两个异步计算合并成一个，这两个异步计算相互独立，同时第二个又以来第一个的计算结果。

2、等待Future集合中的所有任务都完成

3、尽等待集合中执行最开的任务完成

4、应对Future的完成事件（即当Future的完成事件发生时会收到通知，并且使用Future计算的结果进行下一步操作）

使用CompletableFuture构建异步应用

同步API：你调用某个方法，调用方法在被调用方法的执行过程中等待，被调用方法执行结束返回，调用方法取得被调用方法的返回并且继续执行。

异步API：你调用某个方法，被调用方法会直接返回，或者至少被调用方法计算完成之前，将它剩余的计算任务交由另一个线程去做，该线程和调用方法是异步的。执行剩余计算的线程会将计算结果返回给调用方，返回的方式要么通过回调函数，要么是调用方再次执行一个等待，知道计算完成的方法调用。

实现异步API

public class Shop {

    private Random random = new Random();
    private String shopName;

    public Shop(String shopName){
        this.shopName = shopName;
    }

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

    private double calculatePrice(String product){
        delay();

        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    public static void delay(){
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Shop shop = new Shop("BestShop");
        double price = shop.getPrice("huawei mate60");
        System.out.println("price:" + price);
    }
    
}

getPrice变成异步getPriceAsync

public Future<Double> getPriceAsync(String product){
    CompletableFuture<Double> futurePrice = new CompletableFuture<Double>();
    new Thread( () -> {
        double price = calculatePrice(product);
        futurePrice.complete(price);
    }).start();

    return futurePrice;
}

    public static void main(String[] args) {
        Shop shop = new Shop("BestShop");
        Future<Double> futurePrice = shop.getPriceAsync("huawei mate60");
        double price = 0;
        try {
            price = futurePrice.get(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        System.out.println("price:" + price);
    }

使用工厂方法创建CompletableFuture

public Future<Double> getPriceAsync(String product){
    
    CompletableFuture<Double> futurePrice = CompletableFuture.supplyAsync( () -> calculatePrice(product));
    
    return futurePrice;
}

让你的代码免受阻塞之苦

public static void main(String[] args) {
    long start = System.nanoTime();
    List<String> priceList = findPrices("mate60");

    priceList.forEach( obj -> System.out.println(obj));
    System.out.println("duration : " + (System.nanoTime() - start)/1000000);


}

public static List<String> findPrices(String product){
    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavorite"),
            new Shop("BuyItAll")
    );

    List<String> priceList = shops.stream()
            .map( shop -> String.format("%s price is %.2f", shop.getShopName(), shop.getPrice(product)))
            .collect(toList());

    return priceList;
}

BestPrice price is 135.37
LetsSaveBig price is 149.27
MyFavorite price is 145.88
BuyItAll price is 204.46
duration : 4077

使用并行流对请求进行异步

public static List<String> findPrices(String product){
    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavorite"),
            new Shop("BuyItAll")
    );

    List<String> priceList = shops.parallelStream()
            .map( shop -> String.format("%s price is %.2f", shop.getShopName(), shop.getPrice(product)))
            .collect(toList());

    return priceList;
}

BestPrice price is 158.87
LetsSaveBig price is 172.89
MyFavorite price is 117.87
BuyItAll price is 126.37
duration : 1071

使用CompletableFuture发起异步请求

public static List<String> findPrices(String product){
    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavorite"),
            new Shop("BuyItAll")
    );

    List<CompletableFuture<String>> futurePrice = shops.stream()
            .map( shop -> CompletableFuture.supplyAsync(
                    () -> String.format("%s price is %.2f", shop.getShopName(), shop.getPrice(product))
            ))
            .collect(toList());

    return futurePrice.stream()
            .map(CompletableFuture::join)
            .collect(toList());
}

BestPrice price is 126.59
LetsSaveBig price is 114.19
MyFavorite price is 193.87
BuyItAll price is 104.76
duration : 1069

并行流和CompletableFuture他们内部使用相同的通用线程池，默认都使用固定数目的线程，具体线程数目取决于Runtime.getRuntime().availableProcessors()的返回值，然而，CompletableFuture具有一定的优势，可以对执行器executor进行配置。

线程池大小计算规则

线程数=处理器核的数目 * CPU利用率 * 等待时间和计算时间的比率

使用定制执行器

public static List<String> findPrices(String product){
    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavorite"),
            new Shop("BuyItAll")
    );

    ExecutorService executor = Executors.newFixedThreadPool(10);

    List<CompletableFuture<String>> futurePrice = shops.stream()
            .map( shop -> CompletableFuture.supplyAsync(
                    () -> String.format("%s price is %.2f", shop.getShopName(), shop.getPrice(product))
            , executor))
            .collect(toList());

    List<String> prictList = futurePrice.stream()
            .map(CompletableFuture::join)
            .collect(toList());

    executor.shutdown();

    return prictList;
}

BestPrice price is 138.58
LetsSaveBig price is 178.11
MyFavorite price is 190.15
BuyItAll price is 130.21
duration : 1072

并行流还是CompletableFuture

• 如果你进行的是计算密集型的操作，并且没有I/O，那么推荐使用Stream接口并行流，因为实现简单，同时效率也可能是最高的
• 反之，如果你并行的工作单元还涉及等待I/O的操作，那么使用CompletableFuture灵活性更好，可以根据实际情况设置线程数目

对多个异步任务进行流水线操作

在计算商品价格中增加折扣服务。

public class Discount {

    public enum Code {
        NONE(0), SILVER(5), GOLD(10), PLATINUM(15), DIAMOND(20);
        private final int percentage;
        Code(int percentage){
            this.percentage = percentage;
        }
    }

    public static String applyDiscount(Quote quote){
        return quote.getShopName() + " price is " + Discount.apply(quote.getPrice(), quote.getDiscountCode());
    }

    private static double apply(double price, Code code){
        delay();
        return price * (100 - code.percentage) / 100;
    }

    public static void delay(){
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

public class Quote {
    private final String shopName;
    private final double price;
    private final Discount.Code discountCode;

    public Quote(String shopName, double price, Discount.Code discountCode) {
        this.shopName = shopName;
        this.price = price;
        this.discountCode = discountCode;
    }

    public static Quote parse(String s){
        String[] split = s.split(":");
        String shopName = split[0];
        double price = Double.parseDouble(split[1]);
        Discount.Code discountCode = Discount.Code.valueOf(split[2]);
        return new Quote(shopName, price, discountCode);
    }

    public String getShopName() {
        return shopName;
    }

    public double getPrice() {
        return price;
    }

    public Discount.Code getDiscountCode() {
        return discountCode;
    }
}

public class Shop {

    private Random random = new Random();
    private String shopName;

    public Shop(String shopName){
        this.shopName = shopName;
    }

    public String getPrice(String product){
        double price = calculatePrice(product);
        Discount.Code code = Discount.Code.values()[random.nextInt(Discount.Code.values().length)];

        return String.format("%s:%.2f:%s", shopName, price, code);
    }

    public Future<Double> getPriceAsync(String product){

        CompletableFuture<Double> futurePrice = CompletableFuture.supplyAsync( () -> calculatePrice(product));

        return futurePrice;
    }

    private double calculatePrice(String product){
        delay();

        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    public String getShopName() {
        return shopName;
    }

    public static void delay(){
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        long start = System.nanoTime();
        List<String> priceList = findPrices("huawei mate60");
        priceList.forEach(str -> System.out.println(str));

        System.out.println("duration : " + (System.nanoTime() - start)/1000000);


    }

    public static List<String> findPrices(String product){
        List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
                new Shop("LetsSaveBig"),
                new Shop("MyFavorite"),
                new Shop("BuyItAll")
        );

        return shops.stream()
                .map(shop -> shop.getPrice(product))
                .map(Quote::parse)
                .map(Discount::applyDiscount)
                .collect(toList());
    }

}

BestPrice price is 135.5495
LetsSaveBig price is 142.624
MyFavorite price is 143.9915
BuyItAll price is 130.696
duration : 8102

构造同步和异步

public static List<String> findPrices(String product){
    List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavorite"),
            new Shop("BuyItAll")
    );

    ExecutorService executor = Executors.newFixedThreadPool(8);

    List<CompletableFuture<String>> priceFuture = shops.stream()
            .map(shop -> CompletableFuture.supplyAsync( () -> shop.getPrice(product), executor))
            .map(future -> future.thenApply(Quote::parse))
            .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync( () -> Discount.applyDiscount(quote), executor)))
            .collect(toList());

    return priceFuture.stream().map(CompletableFuture::join)
            .collect(toList());

}

BestPrice price is 167.47
LetsSaveBig price is 94.0
MyFavorite price is 180.77
BuyItAll price is 162.624
duration : 2082

• thenApply()：当前线程执行结束，返回结果，以结果为参数执行Function方法，并且返回CompletableFuture

• thenCompose()：对两个异步操作进行流水线，第一个操作完成时，将其结果作为参数传递给第二个操作。你可以建立两个CompletableFuture对象，对第一个CompletableFuture对象调用thenCompse，并向其传递一个函数。当第一个CompletableFuture执行完成后，它的结果作为该函数的参数，这个函数的返回值是以第一个CompletableFuture的返回做输入的第二个CompletableFuture对象。

• thenCombine()：它接受名为BiFunction的第二个参数，这个参数定义了当两个CompletableFuture对象完成计算后，两个CompletableFuture的计算结果作为参数，传递给BiFunction
• thenAccept()：CompletableFuture执行完成后，以返回值为参数
• CompletableFuture.anyOf(CompletableFuture数组)：返回第一个执行完成的CompletableFuture对象