架设我是一个店主,提供了商品价格查询api 传过来商品名字,我就返回价格

public double getPrice(String product) {
//一些耗时的操作 比如查询数据库
        return calculatePrice(product);
 }

CompletableFuture

明显这是一个阻塞式的调用,可以改进为异步的↓

  public Future<Double> getPriceAsync(String product) {

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

        Thread thread = new Thread(() -> {
            double price = calculatePrice(product);
            //结果可以通过futurePrice来get
            futurePrice.complete(price);
        });

        thread.start();
        return futurePrice;
    }

不直接返回结果而是一个Future对象
这样客户就可以这样调用

Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
//做写其他的事,直到必须用到价格
 double price = futurePrice.get(1,TimeUnit.SECONDS);//等待一秒,必须拿到价格

completeExceptionally

如果我这边查询价格的方法出了错,希望把异常报给客户,改进

 public Future<Double> getPrice(String product) {
        CompletableFuture<Double> futurePrice = new CompletableFuture<>();
        new Thread( () -> {
                    try {
                        double price = calculatePrice(product);
                        futurePrice.complete(price);
                    } catch (Exception ex) {
                        //这句不同,放入异常,抛给调用方
                        futurePrice.completeExceptionally(ex);
                    }
        }).start();
        return futurePrice;
    }

CompletableFuture.supplyAsync

我发现CompletableFuture本身提供了工厂方法包装了上面这坨,功能一样的简短版改进↓

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

调用方改进

架设我不是店主,所以getPrice没得改进,还是第一个阻塞的版本
我要查这些店list

 private final List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
                                                   new Shop("LetsSaveBig"),
                                                   new Shop("MyFavoriteShop"),
                                                   new Shop("BuyItAll"),
                                                   new Shop("ShopEasy"));

parallelStream

适合计算密集型,都是计算不用怎么等

  //并行流 几个核就几个线程
    public List<String> findPricesParallel(String product) {
        return shops.parallelStream()
                .map(shop -> shop.getName() + " price is " + shop.getPrice(product))
                .collect(Collectors.toList());
    }

Executor 定制

自定义线程数
但是发现,大量时间都在等待商店响应,而不是计算,线程数超过cpu核数更加高效,为了自定义线程数,用自制的执行器,改用CompletableFuture
定制执行器

  ThreadFactory threadFactory=new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        }
    };
 //指定线程池大小 为店的个数
    private final Executor executor = Executors.newFixedThreadPool(shops.size(),threadFactory);
   //CompletableFuture 先拿回 List<CompletableFuture<String>> 再阻塞保证拿到每个String 达成返回 List<String>
    //定制了线程池
    public List<String> findPricesFuture(String product) {

        List<CompletableFuture<String>> priceFutures =
                shops.stream()
                .map(shop -> {
                    //耗时任务
                    Supplier<String> stringSupplier = () -> shop.getName() + " price is "
                            + shop.getPrice(product);
                   //除了任务之外 多一个执行器入参
                    return CompletableFuture.supplyAsync(stringSupplier, executor);
                })
                .collect(Collectors.toList());

        List<String> prices = priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
        return prices;
    }

thenCompose

第一个CompletableFuture作为第二个CompletableFuture的输入

现在有个新需求,商店不止会返回价格,还会返回折扣,我拿着折扣号和价格,再去折扣中心计算一下返回的才是最后实际价格

 public Stream<CompletableFuture<String>> findPricesStream(String product) {

        return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor))
                //对上一个future的结果额外加工
                .map(future -> future.thenApply(Quote::parse))
                //thenCompose方法,用第一个future当入参,得下一个future
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)));
    }

//使用
    public List<String> findPricesFuture(String product) {

//调用上面那个
        List<CompletableFuture<String>> priceFutures = findPricesStream(product)
                .collect(Collectors.<CompletableFuture<String>>toList());

//真的取数据
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

thenAccept

改进,不等所有价格都查到了,返回一个就显示一个

 public void printPricesStream(String product) {
        //开始时间
        long start = System.nanoTime();

        Stream<CompletableFuture<String>> pricesStream = findPricesStream(product);

        CompletableFuture[] futures = pricesStream
                //thenAccept 来一个就处理一个,打印显示出来  不等了
                .map(f -> f.thenAccept(s -> System.out.println(s + " (done in " + ((System.nanoTime() - start) / 1_000_000) + " msecs)")))
                .toArray(size -> new CompletableFuture[size]);

        //等全部结束,有个返回就行的话 改成anyOf
        CompletableFuture.allOf(futures).join();

//打印:提醒一下全部结束了
        System.out.println("All shops have now responded in " + ((System.nanoTime() - start) / 1_000_000) + " msecs");
    }

thenCombine

要把查到的欧元换成美元,需要去查汇率,查汇率不需要依赖其他结果,一边拿到欧元价格,一边查到欧元美元汇率,得到最终美元价格,需要合并2个独立的CompletableFuture,得到一个CompletableFuture

  CompletableFuture<Double> futurePriceInUSD = 
//第一个异步
                CompletableFuture.supplyAsync(() -> shop.getPrice(product))
                .thenCombine(
//参数1 是一个CompletableFuture
                    CompletableFuture.supplyAsync( () ->  ExchangeService.getRate(Money.EUR, Money.USD)),
//参数2 是怎么融合2个异步结果
                    (price, rate) -> price * rate
                );

java8 CompletableFuture