java併發編程學習15--CompletableFuture(二)

【模擬情景

上一篇說到每個shop都會提供一個價格查詢的服務，可是如今咱們進行假設：

1. 全部的價格查詢是同步方式提供的
2. shop在返回價格的同時會返回一個折扣碼
3. 咱們須要解析返回的字符串，而且根據折扣碼區獲取折扣後的價格
4. 折扣後的價格計算依然是同步執行的
5. 查詢價格返回的字符串格式爲shopName:price:discountCode("沃爾瑪:200:15")

定義商店對象：Shop.java

public class Shop {

    private String name;
    public Shop(String name){
        this.name = name;
    }
    public String getName(){
        return name;
    }
   
    public String getPriceFormat(String product){
        double price = calculatePrice(product);
        //隨機返回一個折扣碼
        Discount.Code code = Discount.Code.values()[random.nextInt(Discount.Code.values().length)];
        return String.format("%s:%.2f:%s",name,price,code);
    }

    private double calculatePrice(String product){
        delay();
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    private Random random = new Random();
    /**
     * 模擬耗時操做：延遲一秒
     */
    private static void delay(){
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

定義折扣對象：Discount.java

public class Discount {
    public enum Code{

        NONE(0),SILVER(5),GOLD(10),PLATINUM(15),DIAMOND(20);

        private final int percantage;
        Code(int percentage){
            this.percantage = percentage;
        }
    }

    public static String applyDiscount(Quote quote){
        return quote.getShopName() + "prices is " + Discount.apply(quote.getPrice(),quote.getDiscountCode());
    }
    //計算折扣價格
    private static Double apply(double price ,Code code){
        //模擬遠程操做的延遲
        delay();
        return (price * (100 - code.percantage)) / 100;
    }
    private static void delay(){
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

用於封裝解析getPriceFormat的字符串對象：Quote.java

public class Quote {

    private final String shopName;
    private final double price;
    private final Discount.Code discountCode;

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

因而如今的任務就是：

1. 遠程查詢商品價格
2. 將得到的字符串解析成爲Quote對象
3. 根據Quote對象遠程獲取折扣後的價格

如今先看看同步的方式來執行這個操做：

public List<String> findPrices2(String product){
        return shops.stream()
                .map(shop -> shop.getPriceFormat(product))
                .map(Quote::parse)
                .map(Discount::applyDiscount)
                .collect(Collectors.toList());
    }

由於有兩個耗時操做，每一個1秒，耗時毫無疑問20秒以上：

【對多個異步任務進行流水線操做

1. 獲取價格：使用CompletableFuture.supplyAsync()工廠方法便可，一旦運行結束每一個CompletableFuture對象會包含一個shop返回的字符串，這裏記住使用咱們自定義的執行器。

2. 解析報價：通常狀況下解析操做並不涉及到IO處理，所能夠採用同步處理，因此這裏咱們直接使用CompletableFuture對象的thenApply()方法，代表在的帶運算結果後馬上同步處理。

3. 計算折扣價格：這是一個遠程操做，確定是須要異步執行的，因而咱們如今就有了兩次異步處理（1.獲取價格，2.計算折扣）。如今使用級聯的方式將它們串聯起來工做。CompletableFuture提供了thenCompose方法，代表將兩個異步操做進行流水線處理。第一個異步操做的結果會成爲第二個異步操做的入參。使用這樣的方式，即便Future在向不一樣的shop手機報價，主線程依然能夠執行其餘操做，好比響應UI事件。

因而咱們有了以下代碼：

/**
     * 異步查詢
     * 相比並行流的話CompletableFuture更有優點：能夠對執行器配置，設置線程池大小
     */
    @SuppressWarnings("all")
    private final Executor myExecutor = 
      Executors.newFixedThreadPool(Math.min(shops.size(), 100), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        }
    });

    public List<String> findPrices2Async(String product){
       List<CompletableFuture<String>> futurePrices = shops.stream()
                //首先異步獲取價格
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPriceFormat(product),myExecutor))
                //將獲取的字符串解析成對象
                .map(future -> future.thenApply(Quote::parse))
                //使用另外一個異步任務有獲取折扣價格
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote),myExecutor)))
                .collect(Collectors.toList());
       //等待全部異步任務完成
       return futurePrices.stream().map(CompletableFuture::join).collect(Collectors.toList());

運算結果不到3秒：

【整合兩個CompletableFuture對象

咱們剛纔使用thenCompose()將兩個CompletableFuture結合了起來，而且一個CompletableFuture的運算結果將做爲第二個CompletableFuture的入參。可是更多的狀況是兩個不相干的CompletableFuture對象相互結合，而且咱們也不但願第一個任務結束以後纔開始第二個任務。這時可使用thenCombine()。

好比咱們獲取價格的同時也獲取匯率：

遠程獲取匯率方法：

/**
     * 獲取匯率
     */
    public double getRate(String type){
        delay();
        if("$".equals(type)){
            return 0.3;
        }
        if("￥".equals(type)){
            return 0.7;
        }
        return 1;
    }

結合倆個異步操做：

@Test
    public void combine(){
        Shop shop = new Shop("沃爾瑪");
        Future<Double> futurePrice = CompletableFuture.supplyAsync(() -> shop.getPrice("iphoneX"))
                                                      .thenCombine(CompletableFuture.supplyAsync(() -> shop.getRate("$")),
                                                                   (price,rate) -> price * rate);
    }

thenCombine()接受兩個參數：

1. CompletableFuture對象：代表第二個異步操做
2. BiFunction<? super T,? super U,? extends V>接口：兩個異步操做的結果合併處理