1、Future接口

Future接口在Java 5中被引入，设计初衷是对将来某个时刻会发生的结果进行建模。它建模了一种异步计算，返回一个执行运算结果的引用，当运算结束后，这个引用被返回给调用方。在Future中触发那些潜在耗时的操作把调用线程解放出来，让它能继续执行其他有价值的工作，不需要等待耗时的操作完成。
示例：使用Future以异步的方式执行一个耗时的操作

ExecutorService executor = Executors.newCachedThreadPool();
Future<Double> future = executor.submit(new Callable<Double>() { //向ExecutorService提交一个Callable对象 
  public Double call() {
    return doSomeLongComputation();//以异步方式在新线程中执行耗时的操作
  }
});
doSomethingElse();
try {
  Double result = future.get(1, TimeUnit.SECONDS);//获取异步操作结果，如果被阻塞，无法得到结果，在等待1秒钟后退出
} catch (ExecutionException ee) {
  // 计算抛出一个异常
} catch (InterruptedException ie) {
  // 当前线程在等待过程中被中断
} catch (TimeoutException te) {
  // 在Future对象完成之前超时
}

这种编程方式让你的线程可以在ExecutorService以并发方式调用另一个线程执行耗时操作的同时，去执行一些其他任务。如果已经运行到没有异步操作的结果就无法继续进行时，可以调用它的get方法去获取操作结果。如果操作已经完成，该方法会立刻返回操作结果，否则它会阻塞线程，直到操作完成，返回相应的结果。
为了处理长时间运行的操作永远不返回的可能性，虽然Future提供了一个无需任何参数的get方法，但还是推荐使用重载版本的get方法，它接受一个超时的参数，可以定义线程等待Future结果的时间，而不是永无止境地等待下去。

image.png

• 将两个异步计算合并为一个，这两个异步计算之间相互独立，同时第二个又依赖于第一个的结果。
• 等待Future集合中的所有任务都完成。
• 仅等待Future集合中快结束的任务完成，并返回它的结果。
• 通过编程方式完成一个Future任务的执行。
• 应对Future的完成事件（即当Future的完成事件发生时会收到通知，并能使用Future计算的结果进行下一步操作，不只是简单地阻塞等待操作结果）。

2、实现异步API

示例：实现价格查询器
声明依据指定产品名称返回价格的方法：

public class Shop {
  public double getPrice(String product) {
    // 待实现
  }
}

采用delay方法模拟长期运行的方法的执行，它会人为地引入1秒钟的延迟，方法声明如下:

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

在getPrice方法中引入一个模拟的延迟:

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

这个API的使用者调用该方法时，它依旧会被阻塞。为等待同步事件完成而等待1秒钟，这是无法接受的，尤其是考虑到佳价格查询器对网络中的所有商店都要重复这种操作。

1）将同步方法转换为异步方法

首先需要将getPrice转换为getPriceAsync方法，并修改它的返回值：

public Future<Double> getPriceAsync(String product) { ... }

Java 5引入了java.util.concurrent.Future接口表示一个异步计算（即调用线程可以继续运行，不会因为调用方法而阻塞）的结果。这意味着Future是一个暂时还不可知值的处理器，这个值在计算完成后，可以通过调用它的get方法取得。getPriceAsync方法能立刻返回，给调用线程一个机会，能在同一时间去执行其他有价值的计算任务。新的CompletableFuture类提供了大量的方法，让我们有机会以多种可能的方式轻松地实现这个方法：

public Future<Double> getPriceAsync(String product) {
  CompletableFuture<Double> futurePrice = new CompletableFuture<>();//创建CompletableFuture对象，它会包含计算结果
  new Thread(() -> {
    double price = calculatePrice(product);//在另一个线程中以异步方式执行计算 
    futurePrice.complete(price);//需长时间计算的任务结束并得出结果时，设置Future的返回值 
  }).start();
  return futurePrice;//无需等待还没结束的计算，直接返回Future对象
}

上述代码中，创建了一个代表异步计算的CompletableFuture对象实例，它在计算完成时会包含计算的结果。接着调用fork创建了另一个线程去执行实际的价格计算工作，不等该耗时计算任务结束，直接返回一个Future实例。当请求的产品价格终计算得出时，可以使用它的complete方法，结束completableFuture对象的运行，并设置变量的值。
使用异步API：

Shop shop = new Shop("BestShop");
long start = System.nanoTime();
Future<Double> futurePrice = shop.getPriceAsync("my favorite product"); 
long invocationTime = ((System.nanoTime() - start) / 1_000_000); 
System.out.println("Invocation returned after " + invocationTime                                                  + " msecs");
doSomethingElse(); //执行更多任务
try {
  double price = futurePrice.get();//从Future对象中读取价格，如果价格未知，会发生阻塞
  System.out.printf("Price is %.2f%n", price);
} catch (Exception e) {
  throw new RuntimeException(e);
}
long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
System.out.println("Price returned after " + retrievalTime + " msecs");

上述代码中，客户向商店查询了某种商品的价格。由于商店提供了异步API，该调用立刻返回了一个Future对象，通过该对象客户可以在将来的某个时刻取得商品的价格。这种方式下，客户在进行商品价格查询的同时，还能执行一些其他的任务，比如查询其他家商店中商品的价格，不会阻塞在那里等待第一家商店返回请求的结果。最后，如果所有有意义的工作都完成，客户所有要执行的工作都依赖于商品价格时，再调用Future的get方法。执行这个操作后，客户要么获得Future中封装的值（如果异步任务已经完成），要么发生阻塞，直到该异步任务完成，期望的值能够访问。

2）错误处理

如果价格计算过程中产生了错误，这种情况下你会得到一个相当糟糕的结果：用于提示错误的异常会被限制在试图计算商品价格的当前线程的范围内，最终会杀死该线程，而这会导致等待get方法返回结果的客户端永久地被阻塞。
客户端可以使用重载版本的get方法，它使用一个超时参数来避免发生这样的情况。这是值得推荐的做法，应尽量在代码中添加超时判断的逻辑，避免发生类似的问题。这种方法能防止程序永久地等待下去，超时发生时，程序会得到通知发生了TimeoutException。也因为如此，你不会有机会发现计算商品价格的线程内到底发生了什么问题才引发了这样的失效。为了让客户端能了解商店无法提供请求商品价格的原因，需要使用CompletableFuture的completeExceptionally方法将导致CompletableFuture内发生问题的异常抛出。
抛出CompletableFuture内的异常：

public Future<Double> getPriceAsync(String product) {
  CompletableFuture<Double> futurePrice = new CompletableFuture<>();
  new Thread(() -> {
    try {
      double price = calculatePrice(product);
      futurePrice.complete(price);//如果价格计算正常结束，完成Future操作并设置商品价格 
    } catch (Exception ex) {
      futurePrice.completeExceptionally(ex);//否则抛出导致失败的异常，完成这次Future操作 
    }
  }).start();
  return futurePrice;
}

客户端现在会收到一个ExecutionException异常，该异常接收了一个包含失败原因的Exception参数，即价格计算方法初抛出的异常。
使用工厂方法supplyAsync创建CompletableFuture
CompletableFuture类提供了大量精巧的工厂方法，使用这些方法能更容易地完成整个流程，不用担心实现细节。比如，采用supplyAsync方法后，可以用一行语句重写getPriceAsync方法：

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

supplyAsync方法接受一个生产者（Supplier）作为参数，返回一个CompletableFuture对象，该对象完成异步执行后会读取调用生产者方法的返回值。生产者方法会交由ForkJoinPool池中的某个执行线程运行，也可以使用supplyAsync方法的重载版本，传递第二个参数指定不同的执行线程执行生产者方法。一般而言，向CompletableFuture的工厂方法传递可选参数，指定生产者方法的执行线程是可行的。

3、无阻塞执行

1）使用并行流对请求进行并行操作

使用并行流来避免顺序计算：

public List<String> findPrices(String product) {
  return shops.parallelStream()
    .map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
    .collect(toList());
}

2）使用CompletableFuture发起异步请求

使用工厂方法supplyAsync创建CompletableFuture对象：

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

使用这种方式，会得到一个List<CompletableFuture<String>>，列表中的每个CompletableFuture对象在计算完成后都包含商店的String类型的名称。但是，由于用CompletableFutures实现的findPrices方法要求返回一个List<String>，需要等待所有的future执行完毕，将其包含的值抽取出来，填充到列表中才能返回。
为了实现这个效果，可以向List<CompletableFuture<String>>施加第二个map操作，对List中的所有future对象执行join操作，一个接一个地等待它们运行结束。
CompletableFuture类中的join方法和Future接口中的get有相同的含义，并且也声明在Future接口中，它们唯一的不同是join不会抛出任何检测到的异常。使用它不需要使用try/catch语句块，让传递给第二个map方法的Lambda表达式变得臃肿。
使用CompletableFuture实现findPrices方法：

public List<String> findPrices(String product) {
    List<CompletableFuture<String>> priceFutures = shops.stream()
    .map(shop -> CompletableFuture.supplyAsync(() -> shop.getName() + " price is " +shop.getPrice(product)))
    .collect(Collectors.toList()); 
 
    return priceFutures.stream()
      .map(CompletableFuture::join)
      .collect(toList());
}

这里使用了两个不同的Stream流水线，而不是在同一个处理流的流水线上一 个接一个地放置两个map操作。原因是考虑流操作之间的延迟特性，如果你在单一流水线中处理流，发向不同商家的请求只能以同步、顺序执行的方式才会成功。因此，每个创建CompletableFuture对象只能在前一个操作结束后执行查询指定商家的动作、通知join方法返回计算结果。

3）更优方案

并行流的版本工作得非常好，那是因为它能并行地执行四个任务，所以它几乎能为每个商家分配一个线程。如果想要增加第五个商家到商店列表中，非常不幸，并行流版本的程序比之前也多消耗了差不多1秒钟的时间，因为可以并行运行（通用线程池中处于可用状态的）的四个线程现在都处于繁忙状态，都在对前4个商店进行查 询。第五个查询只能等到前面某一个操作完成释放出空闲线程才能继续。
CompletableFuture版本的程序似乎比并行流版本的程序还快一点，但这个版本也不太令人满意。如果试图让代码处理9个商店，并行流版本耗时3143毫秒， 而CompletableFuture版本耗时3009毫秒。它们看起来不相伯仲，究其原因都一样：它们内部采用的是同样的通用线程池，默认都使用固定数目的线程，具体线程数取决于Runtime. getRuntime().availableProcessors()的返回值。然而，CompletableFuture具有一定的优势，因为它允许你对执行器（Executor）进行配置，尤其是线程池的大小，让它以更适合应用需求的方式进行配置，满足程序的要求，而这是并行流API无法提供的。

4）使用定制的执行器

调整线程池的大小：如果线程池中线程的数量过多，最终它们会竞争稀缺的处理器和内存资源，浪费大量的时间在上下文切换上。反之，如果线程的数目过少，处理器的一些核可能就无法充分利用。
线程池大小与处理器的利用率之比可以使用下面的公式进行估算：
Nthreads = NCPU * UCPU * (1 + W/C)
NCPU：是处理器核的数目，可通过Runtime.getRuntime().availableProce- ssors()得到
UCPU：是期望的CPU利用率（该值应该介于0和1之间）
W/C：是等待时间与计算时间的比率
你的应用99%的时间都在等待商店的响应，所以估算出的W/C比率为100。如果期望的CPU利用率是100%，需要创建一个拥有400个线程的线程池。实际操作中，如果你创建的线程数比商店的数目更多，反而是一种浪费。出于这种考虑，建议将执行器使用的线程数，与需要查询的商店数目设定为同一个值，这样每个商店都应该对应一个服务线程。不过，为了避免发生由于商店的数目过多导致服务器超负荷而崩溃，你还是需要设置一个上限，比如100个线程。

private final Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100),//创建一个线程池，线程池中线程数目为100和商店数目二者中较小的一个值
  new ThreadFactory() {
    public Thread newThread(Runnable r) {
      Thread t = new Thread(r);
      t.setDaemon(true);//使用守护线程，这种方式不会阻止程序的关停
      return t;
    }
});

现在创建的是一个由守护线程构成的线程池。Java程序无法终止或者退出一个正在运行中的线程，所以最后剩下的那个线程会由于一直等待无法发生的事件而引发问题。相反，如果将线程标记为守护进程，意味着程序退出时它也会被回收。这二者之间没有性能上的差异。现在，可以将执行器作为第二个参数传递给supplyAsync工厂方法了。比如可以按照下面的方式创建一个可查询指定商品价格的CompletableFuture对象：

CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor);

改进后，使用CompletableFuture方案的程序处理5个商店仅耗时1021秒，处理9个商店耗时1022秒。一般而言，这种状态会一直持续，直到商店的数目达到之前计算的阈值400。
并行时使用流还是CompletableFutures？
对集合进行并行计算有两种方式：要么将其转化为并行流，利用map操作，要么枚举出集合中的每一个元素，创建新的线程，在CompletableFuture内对其进行操作。后者提供了更多的灵活性，你可以调整线程池的大小，而这能帮助你确保整体的计算不会因为线程都在等待I/O而发生阻塞。
使用这些API的建议：

• 如果你进行的是计算密集型的操作，并且没有I/O，那么推荐使用Stream接口，因为实现简单，同时效率也可能是最高的。
• 如果并行的工作单元还涉及等待I/O的操作（包括网络连接等待），那么使用CompletableFuture灵活性更好，可以依据等待/计算，或者W/C的比率设定需要使用的线程数。这种情况不使用并行流的另一个原因是，处理流的流水线中如果发生I/O等待，流的延迟特性会让我们很难判断到底什么时候触发了等待。

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

示例：所有商店都使用一个集中式的折扣服务，该折扣服务提供了五个不同的折扣代码，每个折扣代码对应不同的折扣率。

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;
    }
  }
  // Discount类的具体实现
}

假设所有的商店都统一修改getPrice方法的返回格式。getPrice现在以ShopName:price:DiscountCode的格式返回一个String类型的值。

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", name, price, code);
}
private double calculatePrice(String product) {
  delay();
  return random.nextDouble() * product.charAt(0) + product.charAt(1);
}

1）实现折扣服务

对商店返 回字符串的解析操作封装到了下面的Quote类之中：

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.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;}
}

通过传递shop对象返回的字符串给静态工厂方法parse，可以得到Quote类的一个实例， 它包含了shop的名称、折扣之前的价格，以及折扣代码。
Discount服务还提供了一个applyDiscount方法，它接收一个Quote对象，返回一个字符串，表示生成该Quote的shop中的折扣价格：

public class Discount {
  public enum Code {
    // 源码省略……
  } 
 
  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 format(price * (100 - code.percentage) / 100);
  }
}

2）使用Discount服务

以简单的方式实现使用Discount服务的findPrices方法：

public List<String> findPrices(String product) {
  return shops.stream()
    .map(shop -> shop.getPrice(product)) //取得每个shop对象中商品的原始价格
    .map(Quote::parse)//在Quote对象中对shop返回的字符串进行转换 
    .map(Discount::applyDiscount)//为每个Quote申请折扣
    .collect(toList());
}

通过在shop构成的流上采用流水线方式执行三次map操作，得到了期望的结果。

3）构造同步和异步操作

使用CompletableFuture实现findPrices方法：

public List<String> findPrices(String product) {
  List<CompletableFuture<String>> priceFutures = shops.stream()
    .map(shop -> CompletableFuture.supplyAsync(() -> //以异步方式取得每个shop中指定产品的原始价格 
 shop.getPrice(product), executor))
    .map(future -> future.thenApply(Quote::parse))//Quote对象存在时，对其返回的值进行转换
    .map(future -> future.thenCompose(quote ->//使用另一个异步任务构造期望的Future，申请折扣 CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)))
    .collect(toList()); 
 
    return priceFutures.stream()
      .map(CompletableFuture::join)
      .collect(toList()); 
}

image.png

通常而言，名称中不带Async的方法和它的前一个任务一样，在同一个线程中运行；而名称以Async结尾的方法会将后续的任务提交到一个线程池，所以每个任务是由不同的线程处理的。

4）将两个CompletableFuture对象整合起来

对一个CompletableFuture对象调用了thenCompose方法，并向其传递了第二个CompletableFuture，而第二个CompletableFuture又需要使用第一个CompletableFuture的执行结果作为输入。但是，另一种比较常见的情况是，你需要将两个完 全不相干的CompletableFuture对象的结果整合起来，而且也不希望等到第一个任务完全结 束才开始第二项任务。
这种情况，应该使用thenCombine方法，它接收名为BiFunction的第二参数，这个参数定义了当两个CompletableFuture对象完成计算后，结果如何合并。同thenCompose方法一样，thenCombine方法也提供有一个Async的版本。如果使用thenCombineAsync会导致BiFunction中定义的合并操作被提交到线程池中，由另一个任务以异步的方式执行。
合并两个独立的CompletableFuture对象：

Future<Double> futurePriceInUSD = CompletableFuture
  .supplyAsync(() -> shop.getPrice(product))//创建第一个任务查询 商店取得商品的价格
  .thenCombine(
    CompletableFuture.supplyAsync(() -> exchangeService.getRate(Money.EUR, Money.USD)),//创建第二个独立任务，查询美元和欧元之间的转换汇率 
    (price, rate) -> price * rate //通过乘法整合得到的商品价格和汇率
);

image.png

5）对Future和CompletableFuture的回顾

CompletableFuture利用Lambda表达式以声明式的API提供了一种机制，能够用有效的方式，非常容易地将多个以同步或异步方式执行复杂操作的任务结合到一起。
利用Java 7的方法合并两个Future对象：

ExecutorService executor = Executors.newCachedThreadPool();
final Future<Double> futureRate = executor.submit(new Callable<Double>() {
  public Double call() {
    return exchangeService.getRate(Money.EUR, Money.USD);
  }
});
Future<Double> futurePriceInUSD = executor.submit(new Callable<Double>() {
  public Double call() {
    double priceInEUR = shop.getPrice(product);
    return priceInEUR * futureRate.get();
  }
});  

通过向执行器提交一个Callable对象的方式创建了第一个Future对象，向外部服务查询欧元和美元之间的转换汇率。接着，创建了第二个Future对象，查询指定商店中特定商品的欧元价格。最终，用与上述合并两个独立的CompletableFuture对象一样的方式，在同一个Future中通过查询商店得到的欧元商品价格乘以汇率得到了终的价格。

5、响应CompletableFuture的completion事件

使用randomDelay方法取代原来的固定延迟，模拟生成0.5秒至2.5秒随机延迟的方法：

private static final Random random = new Random();
public static void randomDelay() {
  int delay = 500 + random.nextInt(2000);
  try {
    Thread.sleep(delay);
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
}

对最佳价格查询器应用的优化
要避免的首要问题是，等待创建一个包含了所有价格的List创建完成。应该做的是直接处理CompletableFuture流，这样每个CompletableFuture都在为某个商店执行必要的操作。
重构findPrices方法返回一个由Future构成的流：

public Stream<CompletableFuture<String>> findPricesStream(String product) {
  return 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)));
}

现在，为findPricesStream方法返回的Stream添加了第四个map操作，在此之前，已经在该方法内部调用了三次map。这个新添加的操作只是在每个CompletableFuture上注册一个操作，该操作会在CompletableFuture完成执行后使用它的返回值。Java 8的CompletableFuture通过thenAccept方法提供了这一功能，它接收 CompletableFuture执行完毕后的返回值做参数。
将结果打印输出：

findPricesStream("myPhone").map(f -> f.thenAccept(System.out::println));

和thenCompose和thenCombine方法一样，thenAccept方法也提供 了一个异步版本，名为thenAcceptAsync。异步版本的方法会对处理结果的消费者进行调度，从线程池中选择一个新的线程继续执行，不再由同一个线程完成CompletableFuture的所有任务。
由于thenAccept方法已经定义了如何处理CompletableFuture返回的结果，一旦CompletableFuture计算得到结果，它就返回一个CompletableFuture<Void>。所以，map操作返回的是一个Stream<CompletableFuture<Void>>。
把构成Stream的所有CompletableFuture<Void>对象放到一个数组中，等待所有的任务执行完成：

CompletableFuture[] futures = findPricesStream("myPhone") 
    .map(f -> f.thenAccept(System.out::println))
    .toArray(size -> new CompletableFuture[size]); CompletableFuture.allOf(futures).join(); 

allOf工厂方法接收一个由CompletableFuture构成的数组，数组中的所有CompletableFuture对象执行完成之后，它返回一个CompletableFuture<Void>对象。
如果只要CompletableFuture对象数组中有任何一个执行完毕就不再等待，可以使用一个类似的工厂方法anyOf。该方法接收一个CompletableFuture对象构成的数组，返回由第一个执行完毕的CompletableFuture对象的返回值构成的CompletableFuture<Object>。

--参考文献《Java8实战》