Java8实战-总结47

CompletableFuture：组合式异步编程

让代码免受阻塞之苦

使用定制的执行器

就这个主题而言，明智的选择似乎是创建一个配有线程池的执行器，线程池中线程的数目取决于你预计你的应用需要处理的负荷，但是该如何选择合适的线程数目呢？

调整线程池的大小
《Java并发编程实战》一书中，Brian Goetz和合著者们为线程池大小
的优化提供了不少中肯的建议。这非常重要，如果线程池中线程的数量过多，最终它们会竞争
稀缺的处理器和内存资源，浪费大量的时间在上下文切换上。反之，如果线程的数目过少，正
如你的应用所面临的情况，处理器的一些核可能就无法充分利用。Brian Goetz建议，线程池大
小与处理器的利用率之比可以使用下面的公式进行估算：
Nthreads = NCPU * UCPU * (1 + W/C) 
其中：
❑NCPU是处理器的核的数目，可以通过Runtime.getRuntime().availableProcessors()得到
❑UCPU是期望的CPU利用率（该值应该介于0和1之间）
❑W/C是等待时间与计算时间的比率
1
2
3
4
5
6
7
8
9
10
11

你的应用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; 
			}
}); 
1
2
3
4
5
6
7
8
9

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

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

改进之后，使用CompletableFuture方案的程序处理5个商店仅耗时1021秒，处理9个商店时耗时1022秒。一般而言，这种状态会一直持续，直到商店的数目达到我们之前计算的阈值400。这个例子证明了要创建更适合你的应用特性的执行器，利用CompletableFutures向其提交任务执行是个不错的主意。处理需大量使用异步操作的情况时，这几乎是最有效的策略。

并行——使用流还是CompletableFutures？

目前为止，你已经知道对集合进行并行计算有两种方式：要么将其转化为并行流，利用map这样的操作开展工作，要么枚举出集合中的每一个元素，创建新的线程，在CompletableFuture内对其进行操作。后者提供了更多的灵活性，你可以调整线程池的大小，而这能帮助你确保整体的计算不会因为线程都在等待I/O而发生阻塞。

我们对使用这些API的建议如下。
❑如果你进行的是计算密集型的操作，并且没有I/O，那么推荐使用Stream接口，因为实
现简单，同时效率也可能是最高的（如果所有的线程都是计算密集型的，那就没有必要
创建比处理器核数更多的线程）。
❑反之，如果你并行的工作单元还涉及等待I/O的操作（包括网络连接等待），那么使用
CompletableFuture灵活性更好，你可以像前文讨论的那样，依据等待/计算，或者
W/C的比率设定需要使用的线程数。这种情况不使用并行流的另一个原因是，处理流的
流水线中如果发生I/O等待，流的延迟特性会让我们很难判断到底什么时候触发了等待。
1
2
3
4
5
6
7
8
9
10
11
12

现在已经了解了如何利用CompletableFuture为你的用户提供异步API，以及如何将一个同步又缓慢的服务转换为异步的服务。不过到目前为止，每个Future中进行的都是单次的操作。

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

让我们假设所有的商店都同意使用一个集中式的折扣服务。该折扣服务提供了五个不同的折扣代码，每个折扣代码对应不同的折扣率。你使用一个枚举型变量Discount.Code来实现这一想法，具体代码如下所示。

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类的具体实现这里暂且不表示
} 
1
2
3
4
5
6
7
8
9
10
11
12

还假设所有的商店都同意修改getPrice方法的返回格式。getPrice现在以ShopName:price:DiscountCode的格式返回一个String类型的值。示例实现中会返回一个随机生成的Discount.Code，以及已经计算得出的随机价格：

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
2
3
4
5
6
7
8
9
10

调用getPrice方法可能会返回像下面这样一个String值：

BestPrice:123.26:GOLD
1