需求：给商品打折

1、根据产品名称返回价格
2、查询商品的折扣率
3、计算新价格

源码清单

@Data
public class DiscountShop {
    
    

    private final String name;
    private final Random random;

    public DiscountShop(String name) {
    
    
        this.name = name;
        random = new Random(name.charAt(0) * name.charAt(1) * name.charAt(2));
    }

    // 根据产品名称返回价格折扣信息
    public String getPrice(String product) {
    
    
        double price = calculatePrice(product);
        Discount.Code code = Discount.Code.values()[random.nextInt(Discount.Code.values().length)];
        return name + ":" + price + ":" + code;
    }

    public double calculatePrice(String product) {
    
    
        Util.delay();
        // 依据产品的名称，生成一个随机值作为价格
        return Util.format(random.nextDouble() * product.charAt(0) + product.charAt(1));
    }
}

// 报价服务
@Data
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 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.getDisco
   }
   // 价格打折
   private static double apply(double price, Code code) {
    
    
       Util.delay();
       return Util.format(price * (100 - code.percentage) / 100);
   }
}

实现方案

方案1：串行执行

// 串行支持
public List<String> findPricesSequential(String product) {
    
    
    return discountShops.stream()
            .map(discountShop -> discountShop.getPrice(product))
            .map(Quote::parse)
            .map(Discount::applyDiscount)
            .collect(Collectors.toList());
}

方案2：并行流的方式

// 并行流的方式
// Stream底层依赖的是线程数量固定的通用线程池
public List<String> findPricesParallel(String product) {
    
    
    return discountShops.parallelStream()
            .map(discountShop -> discountShop.getPrice(product))
            .map(Quote::parse)
            .map(Discount::applyDiscount)
            .collect(Collectors.toList());
}

方案3：构造同步和异步操作+返回值

private final Executor executor = Executors.newFixedThreadPool(discountShops.size(), ExecuterThreadFactoryBuilder.build());

// 构造同步和异步操作
public List<String> findPricesFuture(String product) {
    
    
   List<CompletableFuture<String>> priceFutures =
           discountShops.stream()
                   .map(discountShop -> CompletableFuture.supplyAsync(() -> discountShop.getPrice(product), executor))
                   .map(future -> future.thenApply(Quote::parse))// 一般情况下解析操作不涉及任何远程服务，也不会进行任何I/O操作，它几乎可以在第一时间进行，所以能够采用同步操作，不会带来太多的延迟。
                   .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)))
                   .collect(Collectors.<CompletableFuture<String>>toList());

   return priceFutures.stream()
           .map(CompletableFuture::join)// 等待流中的所有Future执行完毕，并提取各自的返回值
           .collect(Collectors.toList());

Java 8的 CompletableFuture API提供了名为thenCompose的方法，thenCompose方法允许你对两个异步操作进行流水线，第一个操作完成时，将其结果作为参数传递给第二个操作。使用这种方式，即使Future在向不同的商店收集报价，主线程还是能继续执行其他重要的操作，比如响应UI事件。thenCompose也提供了一个以Async后缀结尾的版本thenComposeAsync。通常而言，名称中不带Async的方法和它的前一个任务一样，在同一个线程中运行；而名称以Async结尾的方法会将后续的任务提交到一个线程池，所以每个任务是由不同的线程处理的。

方案3：构造同步和异步操作+消费型

调用get或者join方法只会造成阻塞，直到CompletableFuture完成才能继续往下运行。如果你希望尽快将不同商店中的商品价格呈现给你的用户（这是车辆保险或者机票比价网站的典型需求），而不是等所有的数据都完备之后再呈现。
Java 8的CompletableFuture通过thenAccept方法提供了这一功能，它接收CompletableFuture执行完毕后的返回值做参数。thenAccept方法也提供了一个异步版本，名为thenAcceptAsync。异步版本的方法会对处理结果的消费者进行调度，从线程池中选择一个新的线程继续执行，不再由同一个线程完成CompletableFuture的所有任务。

private final Executor executor = Executors.newFixedThreadPool(discountShops.size(), ExecuterThreadFactoryBuilder.build());

// 构造同步和异步操作+消费型
public List<String> printPricesStream(String product) {
    
    
    long start = System.nanoTime();

    CompletableFuture<String>[] priceFutures = discountShops.stream()
            .map(discountShop -> CompletableFuture.supplyAsync(() -> discountShop.getPrice(product), executor))
            .map(future -> future.thenApply(Quote::parse))
            .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)))
            .peek(f -> f.thenAccept(s -> System.out.println(s + " (done in " + ((System.nanoTime() - start) / 1_000_000) + " msecs)")))
            .toArray(size -> new CompletableFuture[size]);
    System.out.println("All shops have now responded in " + ((System.nanoTime() - start) / 1_000_000) + " msecs");

    CompletableFuture.allOf(priceFutures).join();  // 等待所有的任务执行完成

    return null; // 仅为了统一使用stopwatch加的返回值
}

allOf工厂方法接收一个由CompletableFuture构成的数组，数组中的所有CompletableFuture对象执行完成之后，它返回一个CompletableFuture<Void>对象。这意味着，如果你需要等待最初Stream中的所有CompletableFuture对象执行完毕。
然而在另一些场景中，你可能希望只要CompletableFuture对象数组中有任何一个执行完毕就不再等待，你可以使用一个类似的工厂方法anyOf。该方法接收一个CompletableFuture对象构成的数组，返回由第一个执行完毕的CompletableFuture对象的返回值构成的CompletableFuture<Object>。

性能比较

public static void main(String[] args) {
    
    
    StopWatch stopWatch = new StopWatch("性能比较");
    execute("sequential", () -> bestPriceFinder.findPricesSequential("myPhone27S"), stopWatch);
    execute("parallel", () -> bestPriceFinder.findPricesParallel("myPhone27S"), stopWatch);
    execute("composed CompletableFuture", () -> bestPriceFinder.findPricesFuture("myPhone27S"), stopWatch);
    execute("composed printPricesStream", () -> bestPriceFinder.printPricesStream("myPhone27S"), stopWatch);
    StopWatchUtils.logStopWatch(stopWatch);
}

private static void execute(String msg, Supplier<List<String>> s, StopWatch stopWatch) {
    
    
    stopWatch.start(msg);
    System.out.println(s.get());
    stopWatch.stop();
    System.out.println();
}

执行结果
[BestPrice price is 110.93, LetsSaveBig price is 135.58, MyFavoriteShop price is 192.72, BuyItAll price is 184.74, ShopEasy price is 167.28]

[BestPrice price is 117.57, LetsSaveBig price is 174.03, MyFavoriteShop price is 173.77, BuyItAll price is 169.89, ShopEasy price is 176.43]

[BestPrice price is 204.78, LetsSaveBig price is 190.85, MyFavoriteShop price is 128.92, BuyItAll price is 140.31, ShopEasy price is 166.1]

All shops have now responded in 7 msecs
ShopEasy price is 224.23 (done in 2034 msecs)
BuyItAll price is 111.53 (done in 2034 msecs)
MyFavoriteShop price is 119.11 (done in 2034 msecs)
BestPrice price is 127.88 (done in 2034 msecs)
LetsSaveBig price is 147.21 (done in 2034 msecs)
null

性能比较 total cost time = 16226 ms
sequential                               : 10118 ms, 62.36%
parallel                                 : 2035 ms, 12.54%
composed CompletableFuture               : 2031 ms, 12.52%
composed printPricesStream               : 2040 ms, 12.57%

参考

《Java8 实战》第11章　CompletableFuture：组合式异步编程