第二十四章 并发编程

爱丽丝：“我可不想到疯子中间去”

猫咪：“啊，那没辙了，我们这都是疯子。我疯了，你也疯了”

爱丽丝：“你怎么知道我疯了”。

猫咪：“你一定是疯了，否则你就不会来这儿” ——爱丽丝梦游仙境 第 6 章。

在本章之前，我们惯用一种简单顺序的叙事方式来编程，有点类似文学中的意识流：第一件事发生了，然后是第二件，第三件……总之，我们完全掌握着事情发生的进展和顺序。如果我们将一个值设置为 5，再看时它已变成 47 的话，这就令人匪夷所思了。

现在，我们来到了陌生的并发世界，在这里这样的结果一点都不奇怪。你原来相信的一切都不再可靠。原有的规则可能生效也可能失效。更可能的是原有的规则只会在某些情况下生效。我们只有完全了解这些情况，才能决定我们处理事情的方式。

比如，我们正常的生活的世界是遵循经典牛顿力学的。物体具有质量：会坠落并且转移动能。电线有电阻，光沿直线传播。假如我们进入到极小、极大、极冷或者极热（那些我们无法生存的世界），这些现象就会发生改变。我们无法判断某物体是粒子还是波，光是否受到重力影响，一些物质还会变为超导体。

假设我们处在多条故事线并行的间谍小说里，非单一意识流地叙事：第一个间谍在岩石底留下了微缩胶片。当第二个间谍来取时，胶片可能已被第三个间谍拿走。小说并没有交代此处的细节。所以直到故事结尾，我们都没搞清楚到底发生了什么。

构建并发程序好比玩搭积木游戏。每拉出一块放在塔顶时都有崩塌的可能。每个积木塔和应用程序都是独一无二的，有着自己的作用。你在某个系统构建中学到的知识并不一定适用于下一个系统。

本章是对并发概念最基本的介绍。虽然我们用到了现代的 Java 8 工具来演示原理，但还远未及全面论述并发。我的目标是为你提供足够的基础知识，使你能够把握问题的复杂性和危险性，从而安全地渡过这片鲨鱼肆虐的困难水域。

更多繁琐和底层的细节，请参阅附录：并发底层原理。要进一步深入这个领域，你还必须阅读 Brian Goetz 等人的 《Java Concurrency in Practice》。在撰写本文时，该书已有十多年的历史了，但它仍包含我们必须要了解和明白的知识要点。理想情况下，本章和上述附录是阅读该书的良好前提。另外，Bill Venner 的 《Inside the Java Virtual Machine》也很值得一看。它详细描述了包括线程在内的 JVM 的内部工作方式。

术语问题

术语“并发”，“并行”，“多任务”，“多处理”，“多线程”，分布式系统（可能还有其他）在整个编程文献中都以多种相互冲突的方式使用，并且经常被混为一谈。 Brian Goetz 在他 2016 年《从并发到并行》的演讲中指出了这一点，之后提出了合理的区分：

• 并发是关于正确有效地控制对共享资源的访问。
• 并行是使用额外的资源来更快地产生结果。

这些定义很好，但是我们已有几十年混乱使用和抗拒解决此问题的历史了。一般来说，当人们使用“并发”这个词时，他们的意思是“所有的一切”。事实上，我自己也经常陷入这样的想法。在大多数书籍中，包括 Brian Goetz 的 《Java Concurrency in Practice》，都在标题中使用这个词。

“并发”通常表示：”不止一个任务正在执行“。而“并行”几乎总是代表：”不止一个任务同时执行“。现在我们能立即看出这些定义中的问题所在：“并行”也有不止一个任务正在执行的语义在里面。区别就在于细节：究竟是怎么“执行”的。此外还有一些重叠：为并行编写的程序依旧可以在单处理器上运行，而并发编写的系统也可以利用多个处理器。

还有另一种方式，围绕”缓慢“出现的情况写下定义：

并发

同时完成多任务。无需等待当前任务完成即可执行其他任务。“并发”解决了程序因外部控制而无法进一步执行的阻塞问题。最常见的例子就是 I/O 操作，任务必须等待数据输入（在一些例子中也称阻塞）。这个问题常见于 I/O 密集型任务。

并行

同时在多个位置完成多任务。这解决了所谓的 CPU 密集型问题：将程序分为多部分，在多个处理器上同时处理不同部分来加快程序执行效率。

上面的定义说明了这两个术语令人困惑的原因：两者的核心都是“同时完成多个任务”，不过并行增加了跨多个处理器的分布。更重要的是，它们可以解决不同类型的问题：并行可能对解决 I / O 密集型问题没有任何好处，因为问题不在于程序的整体执行速度，而在于 I/O 阻塞。而尝试在单个处理器上使用并发来解决计算密集型问题也可能是浪费时间。两种方法都试图在更短的时间内完成更多工作，但是它们实现加速的方式有所不同，这取决于问题施加的约束。

这两个概念混合在一起的一个主要原因是包括 Java 在内的许多编程语言使用相同的机制 - 线程来实现并发和并行。

我们甚至可以尝试以更细的粒度去进行定义（然而这并不是标准化的术语）：

• 纯并发：仍然在单个 CPU 上运行任务。纯并发系统比时序系统更快地产生结果，但是它的运行速度不会因为处理器的增加而变得更快。

• 并发-并行：使用并发技术，结果程序可以利用多处理器更快地产生结果。

• 并行-并发：使用并行编程技术编写，即使只有一个处理器，结果程序仍然可以运行（Java 8 Streams 就是一个很好的例子）。

• 纯并行：只有多个处理器的情况下才能运行。

在某些情况下，这是一个有效的分类法。

支持并发性的语言和库似乎是抽象泄露（Leaky Abstraction一词的完美候选。抽象的目标是“抽象”掉那些对手头的想法不重要的部分，以屏蔽不必要的细节所带来的影响。如果抽象发生泄露，那么即使费很大功夫去隐藏它们，这些细枝末节也总会不断凸显出自己是重要的。

于是我开始怀疑是否真的有高度地抽象。因为当编写这类程序时，底层的系统、工具，甚至是关于 CPU 缓存如何工作的细节，都永远不会被屏蔽。最后，即使你已非常谨慎，你开发的程序也不一定在所有情况下运行正常。有时是因为两台机器的配置不同，有时是程序的预计负载不同。这不是 Java 特有的 - 这是并发和并行编程的本质。

你可能会认为纯函数式语言没有这些限制。实际上，纯函数式语言的确解决了大量并发问题。如果你正在解决一个困难的并发问题，可以考虑用纯函数语言编写这个部分。但是，如果你编写一个使用队列的系统，例如，如果该系统没有被合理地调优，并且输入速率也没有被正确地估计或限制（在不同的情况下，限制意味着具有不同的影响的不同东西），该队列要么被填满并阻塞，要么溢出。最后，你必须了解所有可能会破坏你的系统的细节和问题。这是一种非常不同的编程方式。

并发的新定义

几十年来，我一直在努力解决各种形式的并发问题，其中一个最大的挑战是简洁的定义它。在撰写本章的过程中，我终于有了这样的洞察力，我将其定义为：

并发性是一系列专注于减少等待的性能技术

这实际上是一个相当复杂的表述，所以我将其分解：

• 这是一个集合：包含许多不同的方法来解决这个问题。因为技术差异很大，这是使定义并发性如此具有挑战性的问题之一。
• 这些是性能技术：就是这样。并发的关键点在于让你的程序运行得更快。在 Java 中，并发是非常棘手和困难的，所以绝对不要使用它，除非你有一个重大的性能问题 - 即使这样，使用最简单的方法产生你需要的性能，因为并发很快变得难以管理。
• “减少等待”部分很重要而且微妙。无论（例如）你的程序运行在多少个处理器上，你只能在等待发生时产生效益。如果你发起 I/O 请求并立即获得结果，没有延迟，因此无需改进。如果你在多个处理器上运行多个任务，并且每个处理器都以满容量运行，并且没有任务需要等待其他任务，那么尝试提高吞吐量是没有意义的。并发的唯一机会是程序的某些部分被迫等待。等待会以多种形式出现 - 这解释了为什么存在多种不同的并发方法。

值得强调的是，这个定义的有效性取决于“等待”这个词。如果没有什么可以等待，那就没有机会去加速。如果有什么东西在等待，那么就会有很多方法可以加快速度，这取决于多种因素，包括系统运行的配置，你要解决的问题类型以及其他许多问题。

并发的超能力

想象一下，你置身于一部科幻电影。你必须在一栋大楼中找到一个东西，它被小心而巧妙地隐藏在大楼千万个房间中的一间。你进入大楼，沿着走廊走下去。走廊是分开的。

一个人完成这项任务要花上一百辈子的时间。

现在假设你有一个奇怪的超能力。你可以将自己一分为二，然后在继续前进的同时将另一半送到另一个走廊。每当你在走廊或楼梯上遇到分隔到下一层时，你都会重复这个分裂的技巧。最终，整个建筑中的每个走廊的终点都有一个你。

每个走廊都有一千个房间。此时你的超能力变得弱了一点，你只能克隆 50 个自己来并发搜索走廊里面的房间。

一旦克隆体进入房间，它必须搜索房间的每个角落。这时它切换到了第二种超能力。它分裂成了一百万个纳米机器人，每个机器人都会飞到或爬到房间里一些看不见的地方。你不需要了解这种功能 - 一旦你开启它就会自动工作。在他们自己的控制下，纳米机器人开始行动，搜索房间然后回来重新组装成你，突然间，不知怎么的，你就知道这间房间里有没有那个东西。

我很想说，“并发就是刚才描述的置身于科幻电影中的超能力“。就像你自己可以一分为二然后解决更多的问题一样简单。但是问题在于，我们来描述这种现象的任何模型最终都是抽象泄露的（leaky abstraction)。

以下是其中一个泄露：在理想的世界中，每次克隆自己时，也会复制一个物理处理器来运行克隆搜索者。这当然是不现实的——实际上，你的机器上一般只有 4 个或 8 个处理器核心（编写本文时的典型情况）。或许你拥有更多的处理器，但仍有很多情况下只有一个单核处理器。在关于抽象的讨论中，分配物理处理器核心这本身就是抽象的泄露，甚至也可以支配你的决策。

让我们在科幻电影中改变一些东西。现在当每个克隆搜索者最终到达一扇门时，他们必须敲门并等到有人开门。如果每个搜索者都有一个处理器核心，这没有问题——只是空闲等待直到有人开门。但是如果我们只有 8 个处理器核心却有几千个搜索者，我们不希望处理器仅仅因为某个搜索者恰好在等待回答中被锁住而闲置下来。相反，我们希望将处理器应用于可以真正执行工作的搜索者身上，因此需要将处理器从一个任务切换到另一个任务的机制。

许多模型能够有效地隐藏处理器的数量，允许你假装有很多个处理器。但在某些情况下，当你必须明确知道处理器数量以便于工作的时候，这些模型就会失效。

最大的影响之一取决于是使用单核处理器还是多核处理器。如果你只有单核处理器，那么任务切换的成本也由该核心承担，将并发技术应用于你的系统会使它运行得更慢。

这可能会让你以为，在单核处理器的情况下，编写并发代码是没有意义的。然而，有些情况下，并发模型会产生更简单的代码，光是为了这个目的就值得舍弃一些性能。

在克隆体敲门等待的情况下，即使单核处理器系统也能从并发中受益，因为它可以从等待（阻塞）的任务切换到准备运行的任务。但是如果所有任务都可以一直运行那么切换的成本反而会使任务变慢，在这种情况下，并发只在如果你有多个处理器的情况下有意义。

假设你正在尝试破解某种密码，在同一时间内参与破解的线程越多，你越快得到答案的可能性就越大。每个线程都能持续使用你所分配的处理器时间，在这种情况下（CPU 密集型问题），你代码中的线程数应该和你拥有的处理器的核心数保持一致。

在接听电话的客户服务部门，你只有一定数量的员工，但是你的部门可能会收到很多电话。这些员工（处理器）一次只能接听一个电话直到打完，此时其它打来的电话必须排队等待。

在“鞋匠和精灵”的童话故事中，鞋匠有很多工作要做，当他睡着时，出现了一群精灵来为他制作鞋子。这里的工作是分布式的，但即使使用大量的物理处理器，在制造鞋子的某些部件时也会产生限制——例如，如果鞋底的制作时间最长，这就限制了鞋子的制作速度，这也会改变你设计解决方案的方式。

因此，你要解决的问题驱动了方案的设计。将一个问题分解成“独立运行”的子任务，这是一种美好的抽象，然后就是实际发生的现实：物理现实不断干扰和动摇这个抽象。

这只是问题的一部分：考虑一个制作蛋糕的工厂。我们以某种方式把制作蛋糕的任务分给了工人们，现在是时候让工人把蛋糕放在盒子里了。那里有一个准备存放蛋糕的盒子。但是在一个工人把蛋糕放进盒子之前，另一个工人就冲过去，把蛋糕放进盒子里，砰！这两个蛋糕撞到一起砸坏了。这是常见的“共享内存”问题，会产生所谓的竞态条件（race condition），其结果取决于哪个工人能先把蛋糕放进盒子里（通常使用锁机制来解决问题，因此一个工作人员可以先抓住一个盒子并防止蛋糕被砸烂）。

当“同时”执行的任务相互干扰时，就会出现问题。这可能以一种微妙而偶然的方式发生，因此可以说并发是“可以论证的确定性，但实际上是不确定性的”。也就是说，假设你很小心地编写并发程序，而且通过了代码检查可以正确运行。然而实际上，我们编写的并发程序大部分情况下都能正常运行，但是在一些特定情况下会失败。这些情况可能永远不会发生，或者在你在测试期间几乎很难发现它们。实际上，编写测试代码通常无法为并发程序生成故障条件。由此产生的失败只会偶尔发生，因此它们以客户投诉的形式出现。这是学习并发中最强有力的论点之一：如果你忽略它，你可能会受伤。

因此，并发似乎充满了危险，如果这让你有点害怕，这可能是一件好事。尽管 Java 8 在并发性方面做出了很大改进，但仍然没有像编译时验证 (compile-time verification) 或受检查的异常 (checked exceptions) 那样的安全网来告诉你何时出现错误。关于并发，你只能依靠自己，只有知识渊博、保持怀疑和积极进取的人，才能用 Java 编写可靠的并发代码。

并发为速度而生

在听说并发编程的问题之后，你可能会想知道它是否值得这么麻烦。答案是“不，除非你的程序运行速度不够快。”并且在决定用它之前你会想要仔细思考。不要随便跳进并发编程的悲痛之中。如果有一种方法可以在更快的机器上运行你的程序，或者如果你可以对其进行分析并发现瓶颈并在该位置替换更快的算法，那么请执行此操作。只有在显然没有其他选择时才开始使用并发，然后仅在必要的地方去使用它。

速度问题一开始听起来很简单：如果你想要一个程序运行得更快，将其分解为多个部分，并在单独的处理器上运行每个部分。随着我们提高时钟速度的能力耗尽（至少对传统芯片而言），速度的提高是出现在多核处理器的形式而不是更快的芯片。为了使程序运行得更快，你必须学会利用那些额外的处理器（译者注：处理器一般代表 CPU 的一个逻辑核心），这是并发所带来的好处之一。

对于多处理器机器，可以在这些处理器之间分配多个任务，这可以显著提高吞吐量。强大的多处理器 Web 服务器通常就是这种情况，它可以在程序中为 CPU 分配大量用户请求，每个请求分配一个线程。

但是，并发通常可以提高在单处理器上运行的程序的性能。这听起来有点违反直觉。如果你仔细想想，由于上下文切换的成本增加（从一个任务切换到另一个任务），在单个处理器上运行的并发程序实际上应该比程序的所有部分顺序运行具有更多的开销。从表面上看，将程序的所有部分作为单个任务运行，并且节省上下文切换的成本，这样看似乎更划算。

使这个问题变得有些不同的是阻塞。如果程序中的某个任务由于程序控制之外的某种情况而无法继续（通常是 I/O），我们就称该任务或线程已阻塞（在我们的科幻故事中，就是克隆人已经敲门并等待它打开）。如果没有并发，整个程序就会停下来，直到外部条件发生变化。但是，如果使用并发编写程序，则当一个任务被阻塞时，程序中的其他任务可以继续执行，因此整个程序得以继续运行。事实上，从性能的角度来看，如果没有任务会阻塞，那么在单处理器机器上使用并发是没有意义的。

单处理器系统中性能改进的一个常见例子是事件驱动编程，特别是用户界面编程。考虑一个程序执行一些耗时操作，最终忽略用户输入导致无响应。如果你有一个“退出”按钮，你不想在你编写的每段代码中都检查它的状态（轮询）。这会产生笨拙的代码，也无法保证程序员不会忘了检查。没有并发，生成可响应用户界面的唯一方法是让所有任务都定期检查用户输入。通过创建单独的线程以执行用户输入的响应，能够让程序保证一定程度的响应能力。

实现并发的一种简单方式是使用操作系统级别的进程。与线程不同，进程是在其自己的地址空间中运行的独立程序。进程的优势在于，因为操作系统通常将一个进程与另一个进程隔离，因此它们不会相互干扰，这使得进程编程相对容易。相比之下，线程之间会共享内存和 I/O 等资源，因此编写多线程程序最基本的困难，在于协调不同线程驱动的任务之间对这些资源的使用，以免这些资源同时被多个任务访问。

有些人甚至提倡将进程作为唯一合理的并发实现方式¹，但遗憾的是，通常存在数量和开销方面的限制，从而阻止了进程在并发范围内的适用性（最终你会习惯标准的并发限制，“这种方法适用于一些情况但不适用于其他情况”）

一些编程语言旨在将并发任务彼此隔离。这些通常被称为函数式语言，其中每个函数调用不产生副作用（不会干扰到其它函数），所以可以作为独立的任务来驱动。Erlang 就是这样一种语言，它包括一个任务与另一个任务进行通信的安全机制。如果发现程序的某一部分必须大量使用并发，并且在尝试构建该部分时遇到了过多的问题，那么可以考虑使用这些专用的并发语言创建程序的这个部分。

Java 采用了更传统的方法²，即在顺序语言之上添加对线程的支持而不是在多任务操作系统中分叉外部进程，线程是在表示执行程序的单个进程内创建任务。

并发会带来各种成本，包括复杂性成本，但可以被程序设计、资源平衡和用户便利性方面的改进所抵消。通常，并发性使你能够创建更低耦合的设计；另一方面，你必须特别关注那些使用了并发操作的代码。

Java 并发的四句格言

在经历了多年 Java 并发的实践之后，我总结了以下四个格言：

1.不要用它（避免使用并发）

2.没有什么是真的，一切可能都有问题

3.仅仅是它能运行，并不意味着它没有问题

4.你必须理解它（逃不掉并发）

这些格言专门针对 Java 的并发设计问题，尽管它们也可以适用于其他一些语言。但是，确实存在旨在防止这些问题的语言。

1.不要用它

（而且不要自己去实现它）

避免陷入并发所带来的玄奥问题的最简单方法就是不要用它。尽管尝试一些简单的东西可能很诱人，也似乎足够安全，但是陷阱却是无穷且微妙的。如果你能避免使用它，你的生活将会轻松得多。

使用并发唯一的正当理由是速度。如果你的程序运行速度不够快——这里要小心，因为仅仅想让它运行得更快不是正当理由——应该首先用一个分析器（参见代码校验章中分析和优化）来发现你是否可以执行其他一些优化。

如果你被迫使用并发，请采取最简单，最安全的方法来解决问题。使用知名的库并尽可能少地自己编写代码。对于并发，就没有“太简单了”——自作聪明是你的敌人。

2.没有什么是真的，一切可能都有问题

不使用并发编程，你已经预料到你的世界具有确定的顺序和一致性。对于变量赋值这样简单的操作，很明显它应该总是能够正常工作。

在并发领域，有些事情可能是真的而有些事情却不是，以至于你必须假设没有什么是真的。你必须质疑一切。即使将变量设置为某个值也可能不会按预期的方式工作，事情从这里开始迅速恶化。我已经熟悉了这样一种感觉：我认为应该明显奏效的东西，实际上却行不通。

在非并发编程中你可以忽略的各种事情，在并发下突然变得很重要。例如，你必须了解处理器缓存以及保持本地缓存与主内存一致的问题，你必须理解对象构造的深层复杂性，这样你的构造函数就不会意外地暴露数据，以致于被其它线程更改。这样的例子不胜枚举。

虽然这些主题过于复杂，无法在本章中给你提供专业知识（同样，请参见 Java Concurrency in Practice），但你必须了解它们。

3.仅仅是它能运行，并不意味着它没有问题

我们很容易编写出一个看似正常实则有问题的并发程序，而且问题只有在极少的情况下才会显现出来——在程序部署后不可避免地会成为用户问题（投诉）。

• 你不能验证出并发程序是正确的，你只能（有时）验证出它是不正确的。
• 大多数情况下你甚至没办法验证：如果它出问题了，你可能无法检测到它。
• 你通常无法编写有用的测试，因此你必须依靠代码检查和对并发的深入了解来发现错误。
• 即使是有效的程序也只能在其设计参数下工作。当超出这些设计参数时，大多数并发程序会以某种方式失败。

在其他 Java 主题中，我们养成了决定论的观念。一切都按照语言的承诺的（或暗示的）发生，这是令人欣慰的也是人们所期待的——毕竟，编程语言的意义就是让机器做我们想要它做的事情。从确定性编程的世界进入并发编程领域，我们遇到了一种称为 邓宁-克鲁格效应 的认知偏差，可以概括为“无知者无畏”，意思是：“相对不熟练的人拥有着虚幻的优越感，错误地评估他们的能力远高于实际。

我自己的经验是，无论你是多么确定你的代码是线程安全的，它都可能是有问题的。你可以很容易地了解所有的问题，然后几个月或几年后你会发现一些概念，让你意识到你编写的大多数代码实际上都容易受到并发 bug 的影响。当某些代码不正确时，编译器不会告诉你。为了使它正确，在研究代码时，必须将并发性的所有问题都放在前脑中。

在 Java 的所有非并发领域，“没有明显的 bug 而且没有编译报错“似乎意味着一切都好。但对于并发，它没有任何意义。在这种情况你最糟糕的表现就是“自信”。

4.你必须理解它

在格言 1-3 之后，你可能会对并发性感到害怕，并且认为，“到目前为止，我已经避免了它，也许我可以继续避免它。

这是一种理性的反应。你可能知道其他更好地被设计用于构建并发程序的编程语言——甚至是在 JVM 上运行的语言（从而提供与 Java 的轻松通信），例如 Clojure 或 Scala。为什么不用这些语言来编写并发部分，然后用 Java 来做其他的事情呢?

唉，你不能轻易逃脱：

• 即使你从未显示地创建一个线程，你使用的框架也可能——例如，Swing 图形用户界面（GUI）库，或者像 Timer 类（计时器）那样简单的东西。
• 最糟糕的是：当你创建组件时，必须假设这些组件可能会在多线程环境中重用。即使你的解决方案是放弃并声明你的组件是“非线程安全的”，你仍然必须充分了解这样一个语句的重要性及其含义。

人们有时会认为并发对于介绍语言的书来说太高级了，因此不适合放在其中。他们认为并发是一个独立的主题，并且对于少数出现在日常的程序设计中的情况（例如图形用户界面），可以用特殊的惯用法来处理。如果你可以回避，为什么还要介绍这么复杂的主题呢？

唉，如果是这样就好了。遗憾的是，对于线程何时出现在 Java 程序中，这不是你能决定的。仅仅是你自己没有启动线程，并不代表你就可以回避编写使用线程的代码。例如，Web 系统是最常见的 Java 应用之一，本质上是多线程的 Web 服务器，通常包含多个处理器，而并行是利用这些处理器的理想方式。尽管这样的系统看起来很简单，但你必须理解并发才能正确地编写它。

Java 是一种多线程语言，不管你有没有意识到并发问题，它就在那里。因此，有很多使用并发的 Java 程序，要么只是偶然运行，要么大部分时间都在运行，并且会因为未被发现的并发缺陷而时不时地神秘崩溃。有时这种崩溃是相对温和的，但有时它意味着丢失有价值的数据，如果你没有意识到并发问题，你最终可能会把问题归咎于其他地方而不是你的代码中。如果将程序移动到多处理器系统中，这些类型的问题还会被暴露或放大。基本上，了解并发可以使你意识到明显正确的程序也可能会表现出错误的行为。

残酷的真相

当人类开始烹饪他们的食物时，他们大大减少了他们的身体分解和消化食物所需的能量。烹饪创造了一个“外化的胃”，从而释放出能量去发展其他的能力。火的使用促成了文明。

我们现在通过计算机和网络技术创造了一个“外化大脑”，开始了第二次基本转变。虽然我们只是触及表面，但已经引发了其他转变，例如设计生物机制的能力，并且已经看到文化演变的显著加速（过去，人们通过旅游进行文化交流，但现在他们开始在互联网上做这件事）。这些转变的影响和好处已经超出了科幻作家预测它们的能力（他们在预测文化和个人变化，甚至技术转变的次要影响方面都特别困难）。

有了这种根本性的人类变化，看到许多破坏和失败的实验并不令人惊讶。实际上，进化依赖于无数的实验，其中大多数都失败了。这些实验是向前发展的必要条件。

Java 是在充满自信，热情和睿智的氛围中创建的。在发明一种编程语言时，很容易感觉语言的初始可塑性会持续存在一样，你可以把某些东西拿出来，如果不能解决问题，那么就修复它。编程语言以这种方式是独一无二的 - 它们经历了类似水的改变：气态，液态和最终的固态。在气态阶段，灵活性似乎是无限的，并且很容易认为它总是那样。一旦人们开始使用你的语言，变化就会变得更加严重，环境变得更加粘稠。语言设计的过程本身就是一门艺术。

紧迫感来自互联网的最初兴起。它似乎是一场比赛，第一个通过起跑线的人将“获胜”（事实上，Java，JavaScript 和 PHP 等语言的流行程度可以证明这一点）。唉，通过匆忙设计语言而产生的认知负荷和技术债务最终会赶上我们。

Turing completeness 是不足够的;语言需要更多的东西：它们必须能够创造性地表达，而不是用不必要的东西来衡量我们。解放我们的心理能力只是为了扭转并再次陷入困境，这是毫无意义的。我承认，尽管存在这些问题，我们已经完成了令人惊奇的事情，但我也知道如果没有这些问题我们能做得更多。

热情使原始 Java 设计师加入了一些似乎有必要的特性。信心（以及气态的初始语言）让他们认为任何问题随后都可以解决。在时间轴的某个地方，有人认为任何加入 Java 的东西是固定的和永久性的 -他们非常有信心，并相信第一个决定永远是正确的，因此我们看到 Java 的体系中充斥着糟糕的决策。其中一些决定最终没有什么后果;例如，你可以告诉人们不要使用 Vector，但只能在语言中继续保留它以便对之前版本的支持。

线程包含在 Java 1.0 中。当然，对 java 来说支持并发是一个很基本的设计决定，该特性影响了这个语言的各个角落，我们很难想象以后在以后的版本添加它。公平地说，当时并不清楚基本的并发性是多少。像 C 这样的其他语言能够将线程视为一个附加功能，因此 Java 设计师也纷纷效仿，包括一个 Thread 类和必要的 JVM 支持（这比你想象的要复杂得多）。

C 语言是面向过程语言，这限制了它的野心。这些限制使附加线程库合理。当采用原始模型并将其粘贴到复杂语言中时，Java 的大规模扩展迅速暴露了基本问题。在 Thread 类中的许多方法的弃用以及后续的高级库浪潮中，这种情况变得明显，这些库试图提供更好的并发抽象。

不幸的是，为了在更高级别的语言中获得并发性，所有语言功能都会受到影响，包括最基本的功能，例如标识符代表可变值。在简化并发编程中，所有函数和方法中为了保持事物不变和防止副作用都要做出巨大的改变（这些是纯函数式编程语言的基础），但当时对于主流语言的创建者来说似乎是奇怪的想法。最初的 Java 设计师要么没有意识到这些选择，要么认为它们太不同了，并且会劝退许多潜在的语言使用者。我们可以慷慨地说，语言设计社区当时根本没有足够的经验来理解调整在线程库中的影响。

Java 实验告诉我们，结果是悄然灾难性的。程序员很容易陷入认为 Java 线程并不那么困难的陷阱。表面上看起来正常工作的程序实际上充满了微妙的并发 bug。

为了获得正确的并发性，语言功能必须从头开始设计并考虑并发性。木已成舟；Java 将不再是为并发而设计的语言，而只是一种允许并发的语言。

尽管有这些基本的不可修复的缺陷，但令人印象深刻的是它已经走了这么远。Java 的后续版本添加了库，以便在使用并发时提升抽象级别。事实上，我根本不会想到有可能在 Java 8 中进行改进：并行流和 CompletableFutures - 这是惊人的史诗般的变化，我会惊奇地重复的查看它³。

这些改进非常有用，我们将在本章重点介绍并行流和 CompletableFutures 。虽然它们可以大大简化你对并发和后续代码的思考方式，但基本问题仍然存在：由于 Java 的原始设计，代码的所有部分仍然很脆弱，你仍然必须理解这些复杂和微妙的问题。Java 中的线程绝不是简单或安全的;那种经历必须降级为另一种更新的语言。

本章其余部分

这是我们将在本章的其余部分介绍的内容。请记住，本章的重点是使用最新的高级 Java 并发结构。相比于旧的替代品，使用这些会使你的生活更加轻松。但是，你仍会在遗留代码中遇到一些低级工具。有时，你可能会被迫自己使用其中的一些。附录：并发底层原理 包含一些更原始的 Java 并发元素的介绍。

• Parallel Streams（并行流） 到目前为止，我已经强调了 Java 8 Streams 提供的改进语法。现在该语法（作为一个粉丝，我希望）会使你感到舒适，你可以获得额外的好处：你可以通过简单地将 parallel() 添加到表达式来并行化流。这是一种简单，强大，坦率地说是利用多处理器的惊人方式

添加 parallel() 来提高速度似乎是微不足道的，但是，唉，它就像你刚刚在残酷的真相 中学到的那样简单。我将演示并解释一些盲目添加 parallel() 到 Stream 表达式的缺陷。

• 创建和运行任务 任务是一段可以独立运行的代码。为了解释创建和运行任务的一些基础知识，本节介绍了一种比并行流或 CompletableFutures 更简单的机制：Executor。执行者管理一些低级 Thread 对象（Java 中最原始的并发形式）。你创建一个任务，然后将其交给 Executor 去运行。

有多种类型的 Executor 用于不同的目的。在这里，我们将展示规范形式，代表创建和运行任务的最简单和最佳方法。

• 终止长时间运行的任务 任务独立运行，因此需要一种机制来关闭它们。典型的方法使用了一个标志，这引入了共享内存的问题，我们将使用 Java 的“Atomic”库来回避它。
• Completable Futures 当你将衣服带到干洗店时，他们会给你一张收据。你继续完成其他任务，当你的衣服洗干净时你可以把它取走。收据是你与干洗店在后台执行的任务的连接。这是 Java 5 中引入的 Future 的方法。

Future 比以前的方法更方便，但你仍然必须出现并用收据取出干洗，如果任务没有完成你还需要等待。对于一系列操作，Futures 并没有真正帮助那么多。

Java 8 CompletableFuture 是一个更好的解决方案：它允许你将操作链接在一起，因此你不必将代码写入接口排序操作。有了 CompletableFuture 完美的结合，就可以更容易地做出“采购原料，组合成分，烹饪食物，提供食物，收拾餐具，储存餐具”等一系列链式操作。

• 死锁 某些任务必须去等待 - 阻塞来获得其他任务的结果。被阻止的任务有可能等待另一个被阻止的任务，另一个被阻止的任务也在等待其他任务，等等。如果被阻止的任务链循环到第一个，没有人可以取得任何进展，你就会陷入死锁。

如果在运行程序时没有立即出现死锁，则会出现最大的问题。你的系统可能容易出现死锁，并且只会在某些条件下死锁。程序可能在某个平台上（例如在你的开发机器）运行正常，但是当你将其部署到不同的硬件时会开始死锁。

死锁通常源于细微的编程错误;一系列无辜的决定，最终意外地创建了一个依赖循环。本节包含一个经典示例，演示了死锁的特性。

• 努力，复杂，成本

我们将通过模拟创建披萨的过程完成本章，首先使用并行流实现它，然后是 CompletableFutures。这不仅仅是两种方法的比较，更重要的是探索你应该投入多少工作来使你的程序变得更快。

并行流

Java 8 流的一个显著优点是，在某些情况下，它们可以很容易地并行化。这来自库的仔细设计，特别是流使用内部迭代的方式 - 也就是说，它们控制着自己的迭代器。特别是，他们使用一种特殊的迭代器，称为 Spliterator，它被限制为易于自动分割。我们只需要念 .parallel() 就会产生魔法般的结果，流中的所有内容都作为一组并行任务运行。如果你的代码是使用 Streams 编写的，那么并行化以提高速度似乎是一种琐事

例如，考虑来自 Streams 的 Prime.java。查找质数可能是一个耗时的过程，我们可以看到该程序的计时：

// concurrent/ParallelPrime.java
import java.util.*;
import java.util.stream.*;
import static java.util.stream.LongStream.*;
import java.io.*;
import java.nio.file.*;
import onjava.Timer;

public class ParallelPrime {
    static final int COUNT = 100_000;
    public static boolean isPrime(long n){
        return rangeClosed(2, (long)Math.sqrt(n)).noneMatch(i -> n % i == 0);
        }
    public static void main(String[] args)
        throws IOException {
        Timer timer = new Timer();
        List<String> primes =
            iterate(2, i -> i + 1)
                .parallel()              // [1]
                .filter(ParallelPrime::isPrime)
                .limit(COUNT)
                .mapToObj(Long::toString)
                .collect(Collectors.toList());
        System.out.println(timer.duration());
        Files.write(Paths.get("primes.txt"), primes, StandardOpenOption.CREATE);
        }
    }

输出结果：

    Output:
    1224

请注意，这不是微基准测试，因为我们计时整个程序。我们将数据保存在磁盘上以防止编译器过激的优化;如果我们没有对结果做任何事情，那么一个高级的编译器可能会观察到程序没有意义并且终止了计算（这不太可能，但并非不可能）。请注意使用 nio2 库编写文件的简单性（在文件 一章中有描述）。

当我注释掉[1] parallel() 行时，我的结果用时大约是 parallel() 的三倍。

并行流似乎是一个甜蜜的交易。你所需要做的就是将编程问题转换为流，然后插入 parallel() 以加快速度。实际上，有时候这很容易。但遗憾的是，有许多陷阱。

• parallel() 不是灵丹妙药

作为对流和并行流的不确定性的探索，让我们看一个看似简单的问题：对增长的数字序列进行求和。事实证明有大量的方式去实现它，并且我将冒险用计时器将它们进行比较 - 我会尽量小心，但我承认我可能会在计时代码执行时遇到许多基本陷阱之一。结果可能有一些缺陷（例如 JVM 没有“热身”），但我认为它仍然提供了一些有用的指示。

我将从一个计时方法 timeTest() 开始，它采用 LongSupplier ，测量 getAsLong() 调用的长度，将结果与 checkValue 进行比较并显示结果。

请注意，一切都必须严格使用 long ；我花了一些时间发现隐蔽的溢出，然后才意识到在重要的地方错过了 long 。

所有关于时间和内存的数字和讨论都是指“我的机器”。你的经历可能会有所不同。

// concurrent/Summing.java
import java.util.stream.*;
import java.util.function.*;
import onjava.Timer;
public class Summing {
    static void timeTest(String id, long checkValue,    LongSupplier operation){
        System.out.print(id + ": ");
        Timer timer = new Timer();
        long result = operation.getAsLong();
        if(result == checkValue)
            System.out.println(timer.duration() + "ms");
        else
            System.out.format("result: %d%ncheckValue: %d%n", result, checkValue);
        }
    public static final int SZ = 100_000_000;
    // This even works:
    // public static final int SZ = 1_000_000_000;
    public static final long CHECK = (long)SZ * ((long)SZ + 1)/2; // Gauss's formula
    public static void main(String[] args){
        System.out.println(CHECK);
        timeTest("Sum Stream", CHECK, () ->
        LongStream.rangeClosed(0, SZ).sum());
        timeTest("Sum Stream Parallel", CHECK, () ->
        LongStream.rangeClosed(0, SZ).parallel().sum());
        timeTest("Sum Iterated", CHECK, () ->
        LongStream.iterate(0, i -> i + 1)
        .limit(SZ+1).sum());
        // Slower & runs out of memory above 1_000_000:
        // timeTest("Sum Iterated Parallel", CHECK, () ->
        //   LongStream.iterate(0, i -> i + 1)
        //     .parallel()
        //     .limit(SZ+1).sum());
    }
}

输出结果：

5000000050000000
Sum Stream: 167ms
Sum Stream Parallel: 46ms
Sum Iterated: 284ms

CHECK 值是使用 Carl Friedrich Gauss（高斯）在 1700 年代后期还在上小学的时候创建的公式计算出来的.

main() 的第一个版本使用直接生成 Stream 并调用 sum() 的方法。我们看到流的好处在于即使 SZ 为十亿（1_000_000_000）程序也可以很好地处理而没有溢出（为了让程序运行得快一点，我使用了较小的数字）。使用 parallel() 的基本范围操作明显更快。

如果使用 iterate() 来生成序列，则减速是相当明显的，可能是因为每次生成数字时都必须调用 lambda。但是如果我们尝试并行化，当 SZ 超过一百万时，结果不仅比非并行版本花费的时间更长，而且也会耗尽内存（在某些机器上）。当然，当你可以使用 range() 时，你不会使用 iterate() ，但如果你生成的东西不是简单的序列，你必须使用 iterate() 。应用 parallel() 是一个合理的尝试，但会产生令人惊讶的结果。我们将在后面的部分中探讨内存限制的原因，但我们可以对流并行算法进行初步观察：

• 流并行性将输入数据分成多个部分，因此算法可以应用于那些单独的部分。
• 数组分割成本低，分割均匀且对分割的大小有着完美的掌控。
• 链表没有这些属性;“拆分”一个链表仅仅意味着把它分成“第一元素”和“其余元素”，这相对无用。
• 无状态生成器的行为类似于数组;上面使用的 range() 就是无状态的。
• 迭代生成器的行为类似于链表; iterate() 是一个迭代生成器。

现在让我们尝试通过在数组中填充值并对数组求和来解决问题。因为数组只分配了一次，所以我们不太可能遇到垃圾收集时序问题。

首先我们将尝试一个充满原始 long 的数组：

// concurrent/Summing2.java
// {ExcludeFromTravisCI}import java.util.*;
public class Summing2 {
    static long basicSum(long[] ia) {
        long sum = 0;
        int size = ia.length;
        for(int i = 0; i < size; i++)
            sum += ia[i];return sum;
    }
    // Approximate largest value of SZ before
    // running out of memory on mymachine:
    public static final int SZ = 20_000_000;
    public static final long CHECK = (long)SZ * ((long)SZ + 1)/2;
    public static void main(String[] args) {
        System.out.println(CHECK);
        long[] la = newlong[SZ+1];
        Arrays.parallelSetAll(la, i -> i);
        Summing.timeTest("Array Stream Sum", CHECK, () ->
        Arrays.stream(la).sum());
        Summing.timeTest("Parallel", CHECK, () ->
        Arrays.stream(la).parallel().sum());
        Summing.timeTest("Basic Sum", CHECK, () ->
        basicSum(la));// Destructive summation:
        Summing.timeTest("parallelPrefix", CHECK, () -> {
            Arrays.parallelPrefix(la, Long::sum);
        return la[la.length - 1];
        });
    }
}

输出结果：

200000010000000
Array Stream
Sum: 104ms
Parallel: 81ms
Basic Sum: 106ms
parallelPrefix: 265ms

第一个限制是内存大小；因为数组是预先分配的，所以我们不能创建几乎与以前版本一样大的任何东西。并行化可以加快速度，甚至比使用 basicSum() 循环更快。有趣的是， Arrays.parallelPrefix() 似乎实际上减慢了速度。但是，这些技术中的任何一种在其他条件下都可能更有用 - 这就是为什么你不能做出任何确定性的声明，除了“你必须尝试一下”。

最后，考虑使用包装类 Long 的效果：

// concurrent/Summing3.java
// {ExcludeFromTravisCI}
import java.util.*;
public class Summing3 {
    static long basicSum(Long[] ia) {
        long sum = 0;
        int size = ia.length;
        for(int i = 0; i < size; i++)
            sum += ia[i];
            return sum;
    }
    // Approximate largest value of SZ before
    // running out of memory on my machine:
    public static final int SZ = 10_000_000;
    public static final long CHECK = (long)SZ * ((long)SZ + 1)/2;
    public static void main(String[] args) {
        System.out.println(CHECK);
        Long[] aL = newLong[SZ+1];
        Arrays.parallelSetAll(aL, i -> (long)i);
        Summing.timeTest("Long Array Stream Reduce", CHECK, () ->
        Arrays.stream(aL).reduce(0L, Long::sum));
        Summing.timeTest("Long Basic Sum", CHECK, () ->
        basicSum(aL));
        // Destructive summation:
        Summing.timeTest("Long parallelPrefix",CHECK, ()-> {
            Arrays.parallelPrefix(aL, Long::sum);
            return aL[aL.length - 1];
            });
    }
}

输出结果：

50000005000000
Long Array
Stream Reduce: 1038ms
Long Basic
Sum: 21ms
Long parallelPrefix: 3616ms

现在可用的内存量大约减半，并且所有情况下所需的时间都会很长，除了 basicSum() ，它只是循环遍历数组。令人惊讶的是， Arrays.parallelPrefix() 比任何其他方法都要花费更长的时间。

我将 parallel() 版本分开了，因为在上面的程序中运行它导致了一个冗长的垃圾收集，扭曲了结果：

// concurrent/Summing4.java
// {ExcludeFromTravisCI}
import java.util.*;
public class Summing4 {
    public static void main(String[] args) {
        System.out.println(Summing3.CHECK);
        Long[] aL = newLong[Summing3.SZ+1];
        Arrays.parallelSetAll(aL, i -> (long)i);
        Summing.timeTest("Long Parallel",
        Summing3.CHECK, () ->
        Arrays.stream(aL)
        .parallel()
        .reduce(0L,Long::sum));
    }
}

输出结果：

50000005000000
Long Parallel: 1014ms

它比非 parallel() 版本略快，但并不显着。

导致时间增加的一个重要原因是处理器内存缓存。使用 Summing2.java 中的原始 long ，数组 la 是连续的内存。处理器可以更容易地预测该阵列的使用，并使缓存充满下一个需要的阵列元素。访问缓存比访问主内存快得多。似乎 Long parallelPrefix 计算受到影响，因为它为每个计算读取两个数组元素，并将结果写回到数组中，并且每个都为 Long 生成一个超出缓存的引用。

使用 Summing3.java 和 Summing4.java ，aL 是一个 Long 数组，它不是一个连续的数据数组，而是一个连续的 Long 对象引用数组。尽管该数组可能会在缓存中出现，但指向的对象几乎总是不在缓存中。

这些示例使用不同的 SZ 值来显示内存限制。

为了进行时间比较，以下是 SZ 设置为最小值 1000 万的结果：

Sum Stream: 69msSum Stream Parallel: 18msSum Iterated: 277ms Array Stream Sum: 57ms Parallel: 14ms Basic Sum: 16ms parallelPrefix: 28ms Long Array Stream Reduce: 1046ms Long Basic Sum: 21ms Long parallelPrefix: 3287ms Long Parallel: 1008ms

虽然 Java 8 的各种内置“并行”工具非常棒，但我认为它们被视为神奇的灵丹妙药：“只需添加 parallel() 并且它会更快！” 我希望我已经开始表明情况并非所有都是如此，并且盲目地应用内置的“并行”操作有时甚至会使运行速度明显变慢。

• parallel()/limit() 交点

使用 parallel() 时会有更复杂的问题。从其他语言中吸取的流机制被设计为大约是一个无限的流模型。如果你拥有有限数量的元素，则可以使用集合以及为有限大小的集合设计的关联算法。如果你使用无限流，则使用针对流优化的算法。

Java 8 将两者合并起来。例如，Collections 没有内置的 map() 操作。在 Collection 和 Map 中唯一类似流的批处理操作是 forEach() 。如果要执行 map() 和 reduce() 等操作，必须首先将 Collection 转换为存在这些操作的 Stream :

// concurrent/CollectionIntoStream.java
import onjava.*;
import java.util.*;
import java.util.stream.*;
public class CollectionIntoStream {
    public static void main(String[] args) {
    List<String> strings = Stream.generate(new Rand.String(5))
    .limit(10)
    .collect(Collectors.toList());
    strings.forEach(System.out::println);
    // Convert to a Stream for many more options:
    String result = strings.stream()
    .map(String::toUpperCase)
    .map(s -> s.substring(2))
    .reduce(":", (s1, s2) -> s1 + s2);
    System.out.println(result);
    }
}

输出结果：

btpen
pccux
szgvg
meinn
eeloz
tdvew
cippc
ygpoa
lkljl
bynxt
:PENCUXGVGINNLOZVEWPPCPOALJLNXT

Collection 确实有一些批处理操作，如 removeAll() ，removeIf() 和 retainAll() ，但这些都是破坏性的操作。ConcurrentHashMap 对 forEach 和 reduce 操作有特别广泛的支持。

在许多情况下，只在集合上调用 stream() 或者 parallelStream() 没有问题。但是，有时将 Stream 与 Collection 混合会产生意想不到的结果。这是一个有趣的难题：

// concurrent/ParallelStreamPuzzle.java
import java.util.*;
import java.util.function.*;
import java.util.stream.*;
public class ParallelStreamPuzzle {
    static class IntGenerator
    implements Supplier<Integer> {
        private int current = 0;
        @Override
        public Integer get() {
            return current++;
        }
    }
    public static void main(String[] args) {
        List<Integer> x = Stream.generate(new IntGenerator())
        .limit(10)
        .parallel()  // [1]
        .collect(Collectors.toList());
        System.out.println(x);
    }
}
/* Output:
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
*/

如果[1] 注释运行它，它会产生预期的： [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 每次。但是包含了 parallel()，它看起来像一个随机数生成器，带有输出（从一次运行到下一次运行不同），如： [0, 3, 6, 8, 11, 14, 17, 20, 23, 26] 这样一个简单的程序怎么会如此糟糕呢？让我们考虑一下我们在这里要实现的目标：“并行生成。”那意味着什么？一堆线程都在从一个生成器取值，然后以某种方式选择有限的结果集？代码看起来很简单，但它变成了一个特别棘手的问题。

为了看到它，我们将添加一些仪器。由于我们正在处理线程，因此我们必须将任何跟踪信息捕获到并发数据结构中。在这里我使用 ConcurrentLinkedDeque ：

// concurrent/ParallelStreamPuzzle2.java
import java.util.*;
import java.util.function.*;
import java.util.stream.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.nio.file.*;
public class ParallelStreamPuzzle2 {
    public static final Deque<String> TRACE =
    new ConcurrentLinkedDeque<>();
    static class
    IntGenerator implements Supplier<Integer> {
        private AtomicInteger current =
        new AtomicInteger();
        @Override
        public Integer get() {
            TRACE.add(current.get() + ": " +Thread.currentThread().getName());
            return current.getAndIncrement();
        }
    }
    public static void main(String[] args) throws Exception {
    List<Integer> x = Stream.generate(newIntGenerator())
    .limit(10)
    .parallel()
    .collect(Collectors.toList());
    System.out.println(x);
    Files.write(Paths.get("PSP2.txt"), TRACE);
    }
}

输出结果：

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

current 是使用线程安全的 AtomicInteger 类定义的，可以防止竞争条件；parallel() 允许多个线程调用 get() 。

在查看 PSP2.txt .IntGenerator.get() 被调用 1024 次时，你可能会感到惊讶。

0: main
1: ForkJoinPool.commonPool-worker-1
2: ForkJoinPool.commonPool-worker-2
3: ForkJoinPool.commonPool-worker-2
4: ForkJoinPool.commonPool-worker-1
5: ForkJoinPool.commonPool-worker-1
6: ForkJoinPool.commonPool-worker-1
7: ForkJoinPool.commonPool-worker-1
8: ForkJoinPool.commonPool-worker-4
9: ForkJoinPool.commonPool-worker-4
10: ForkJoinPool.commonPool-worker-4
11: main
12: main
13: main
14: main
15: main...10
17: ForkJoinPool.commonPool-worker-110
18: ForkJoinPool.commonPool-worker-610
19: ForkJoinPool.commonPool-worker-610
20: ForkJoinPool.commonPool-worker-110
21: ForkJoinPool.commonPool-worker-110
22: ForkJoinPool.commonPool-worker-110
23: ForkJoinPool.commonPool-worker-1

这个块大小似乎是内部实现的一部分（尝试使用limit() 的不同参数来查看不同的块大小）。将parallel()与limit()结合使用可以预取一串值，作为流输出。

试着想象一下这里发生了什么：一个流抽象出无限序列，按需生成。当你要求它并行产生流时，你要求所有这些线程尽可能地调用get()。添加limit()，你说“只需要这些。”基本上，当你为了随机输出而选择将parallel()与limit()结合使用时，这种方法可能对你正在解决的问题有效。但是当你这样做时，你必须明白。这是一个仅限专家的功能，而不是要争辩说“Java 弄错了”。

什么是更合理的方法来解决问题？好吧，如果你想生成一个 int 流，你可以使用 IntStream.range()，如下所示：

// concurrent/ParallelStreamPuzzle3.java
// {VisuallyInspectOutput}
import java.util.*;
import java.util.stream.*;
public class ParallelStreamPuzzle3 {
    public static void main(String[] args) {
    List<Integer> x = IntStream.range(0, 30)
        .peek(e -> System.out.println(e + ": " +Thread.currentThread()
        .getName()))
        .limit(10)
        .parallel()
        .boxed()
        .collect(Collectors.toList());
        System.out.println(x);
    }
}

输出结果：

8: main
6: ForkJoinPool.commonPool-worker-5
3: ForkJoinPool.commonPool-worker-7
5: ForkJoinPool.commonPool-worker-5
1: ForkJoinPool.commonPool-worker-3
2: ForkJoinPool.commonPool-worker-6
4: ForkJoinPool.commonPool-worker-1
0: ForkJoinPool.commonPool-worker-4
7: ForkJoinPool.commonPool-worker-1
9: ForkJoinPool.commonPool-worker-2
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

为了表明 parallel() 确实有效，我添加了一个对 peek() 的调用，这是一个主要用于调试的流函数：它从流中提取一个值并执行某些操作但不影响从流向下传递的元素。注意这会干扰线程行为，但我只是尝试在这里做一些事情，而不是实际调试任何东西。

你还可以看到 boxed() 的添加，它接受 int 流并将其转换为 Integer 流。

现在我们得到多个线程产生不同的值，但它只产生 10 个请求的值，而不是 1024 个产生 10 个值。

它更快吗？一个更好的问题是：什么时候开始有意义？当然不是这么小的一套；上下文切换的代价远远超过并行性的任何加速。很难想象什么时候用并行生成一个简单的数字序列会有意义。如果你要生成的东西需要很高的成本，它可能有意义 - 但这都是猜测。只有通过测试我们才能知道用并行是否有效。记住这句格言：“首先使它工作，然后使它更快地工作 - 只有当你必须这样做时。”将 parallel() 和 limit() 结合使用仅供专家操作（把话说在前面，我不认为自己是这里的专家）。

• 并行流只看起来很容易

实际上，在许多情况下，并行流确实可以毫不费力地更快地产生结果。但正如你所见，仅仅将 parallel() 加到你的 Stream 操作上并不一定是安全的事情。在使用 parallel() 之前，你必须了解并行性如何帮助或损害你的操作。一个基本认知错误就是认为使用并行性总是一个好主意。事实上并不是。Stream 意味着你不需要重写所有代码便可以并行运行它。但是流的出现并不意味着你可以不用理解并行的原理以及不用考虑并行是否真的有助于实现你的目标。

创建和运行任务

如果无法通过并行流实现并发，则必须创建并运行自己的任务。稍后你将看到运行任务的理想 Java 8 方法是 CompletableFuture，但我们将使用更基本的工具介绍概念。

Java 并发的历史始于非常原始和有问题的机制，并且充满了各种尝试的改进。这些主要归入附录：低级并发 (Appendix: Low-Level Concurrency)。在这里，我们将展示一个规范形式，表示创建和运行任务的最简单，最好的方法。与并发中的所有内容一样，存在各种变体，但这些变体要么降级到该附录，要么超出本书的范围。

• Tasks and Executors

在 Java 的早期版本中，你通过直接创建自己的 Thread 对象来使用线程，甚至将它们子类化以创建你自己的特定“任务线程”对象。你手动调用了构造函数并自己启动了线程。

创建所有这些线程的开销变得非常重要，现在不鼓励采用手动操作方法。在 Java 5 中，添加了类来为你处理线程池。你可以将任务创建为单独的类型，然后将其交给 ExecutorService 以运行该任务，而不是为每种不同类型的任务创建新的 Thread 子类型。ExecutorService 为你管理线程，并且在运行任务后重新循环线程而不是丢弃线程。

首先，我们将创建一个几乎不执行任务的任务。它“sleep”（暂停执行）100 毫秒，显示其标识符和正在执行任务的线程的名称，然后完成：

// concurrent/NapTask.java
import onjava.Nap;
public class NapTask implements Runnable {
    final int id;
    public NapTask(int id) {
        this.id = id;
        }
    @Override
    public void run() {
        new Nap(0.1);// Seconds
        System.out.println(this + " "+
            Thread.currentThread().getName());
        }
    @Override
    public String toString() {
        return"NapTask[" + id + "]";
    }
}

这只是一个 Runnable ：一个包含 run() 方法的类。它没有包含实际运行任务的机制。我们使用 Nap 类中的“sleep”：

// onjava/Nap.java
package onjava;
import java.util.concurrent.*;
public class Nap {
    public Nap(double t) { // Seconds
        try {
            TimeUnit.MILLISECONDS.sleep((int)(1000 * t));
        } catch(InterruptedException e){
            throw new RuntimeException(e);
        }
    }
    public Nap(double t, String msg) {
        this(t);
        System.out.println(msg);
    }
}

为了消除异常处理的视觉干扰，这被定义为实用程序。第二个构造函数在超时时显示一条消息

对 TimeUnit.MILLISECONDS.sleep() 的调用获取“当前线程”并在参数中将其置于休眠状态，这意味着该线程被挂起。这并不意味着底层处理器停止。操作系统将其切换到其他任务，例如在你的计算机上运行另一个窗口。OS 任务管理器定期检查 sleep() 是否超时。当它执行时，线程被“唤醒”并给予更多处理时间。

你可以看到 sleep() 抛出一个受检的 InterruptedException ;这是原始 Java 设计中的一个工件，它通过突然断开它们来终止任务。因为它往往会产生不稳定的状态，所以后来不鼓励终止。但是，我们必须在需要或仍然发生终止的情况下捕获异常。

要执行任务，我们将从最简单的方法–SingleThreadExecutor 开始:

//concurrent/SingleThreadExecutor.java
import java.util.concurrent.*;
import java.util.stream.*;
import onjava.*;
public class SingleThreadExecutor {
    public static void main(String[] args) {
        ExecutorService exec =
            Executors.newSingleThreadExecutor();
        IntStream.range(0, 10)
            .mapToObj(NapTask::new)
            .forEach(exec::execute);
        System.out.println("All tasks submitted");
        exec.shutdown();
        while(!exec.isTerminated()) {
            System.out.println(
            Thread.currentThread().getName()+
            " awaiting termination");
            new Nap(0.1);
        }
    }
}

输出结果：

All tasks submitted
main awaiting termination
main awaiting termination
NapTask[0] pool-1-thread-1
main awaiting termination
NapTask[1] pool-1-thread-1
main awaiting termination
NapTask[2] pool-1-thread-1
main awaiting termination
NapTask[3] pool-1-thread-1
main awaiting termination
NapTask[4] pool-1-thread-1
main awaiting termination
NapTask[5] pool-1-thread-1
main awaiting termination
NapTask[6] pool-1-thread-1
main awaiting termination
NapTask[7] pool-1-thread-1
main awaiting termination
NapTask[8] pool-1-thread-1
main awaiting termination
NapTask[9] pool-1-thread-1

首先请注意，没有 SingleThreadExecutor 类。newSingleThreadExecutor() 是 Executors 中的一个工厂方法，它创建特定类型的 ExecutorService ⁴

我创建了十个 NapTasks 并将它们提交给 ExecutorService，这意味着它们开始自己运行。然而，在此期间，main() 继续做事。当我运行 callexec.shutdown() 时，它告诉 ExecutorService 完成已经提交的任务，但不接受任何新任务。此时，这些任务仍然在运行，因此我们必须等到它们在退出 main() 之前完成。这是通过检查 exec.isTerminated() 来实现的，这在所有任务完成后变为 true。

请注意，main() 中线程的名称是 main，并且只有一个其他线程 pool-1-thread-1。此外，交错输出显示两个线程确实同时运行。

如果你只是调用 exec.shutdown()，程序将完成所有任务。也就是说，不需要 while(!exec.isTerminated()) 。

// concurrent/SingleThreadExecutor2.java
import java.util.concurrent.*;
import java.util.stream.*;
public class SingleThreadExecutor2 {
    public static void main(String[] args)throws InterruptedException {
        ExecutorService exec
        =Executors.newSingleThreadExecutor();
        IntStream.range(0, 10)
            .mapToObj(NapTask::new)
            .forEach(exec::execute);
        exec.shutdown();
    }
}

输出结果：

NapTask[0] pool-1-thread-1
NapTask[1] pool-1-thread-1
NapTask[2] pool-1-thread-1
NapTask[3] pool-1-thread-1
NapTask[4] pool-1-thread-1
NapTask[5] pool-1-thread-1
NapTask[6] pool-1-thread-1
NapTask[7] pool-1-thread-1
NapTask[8] pool-1-thread-1
NapTask[9] pool-1-thread-1

一旦你调用了 exec.shutdown()，尝试提交新任务将抛出 RejectedExecutionException。

// concurrent/MoreTasksAfterShutdown.java
import java.util.concurrent.*;
public class MoreTasksAfterShutdown {
    public static void main(String[] args) {
        ExecutorService exec
        =Executors.newSingleThreadExecutor();
        exec.execute(newNapTask(1));
        exec.shutdown();
        try {
            exec.execute(newNapTask(99));
        } catch(RejectedExecutionException e) {
            System.out.println(e);
        }
    }
}

输出结果：

java.util.concurrent.RejectedExecutionException: TaskNapTask[99] rejected from java.util.concurrent.ThreadPoolExecutor@4e25154f[Shutting down, pool size = 1,active threads = 1, queued tasks = 0, completed tasks =0]NapTask[1] pool-1-thread-1

exec.shutdown() 的替代方法是 exec.shutdownNow() ，它除了不接受新任务外，还会尝试通过中断任务来停止任何当前正在运行的任务。同样，中断是错误的，容易出错并且不鼓励。

• 使用更多线程

使用线程的重点是（几乎总是）更快地完成任务，那么我们为什么要限制自己使用 SingleThreadExecutor 呢？查看执行 Executors 的 Javadoc，你将看到更多选项。例如 CachedThreadPool：

// concurrent/CachedThreadPool.java
import java.util.concurrent.*;
import java.util.stream.*;
public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService exec
        =Executors.newCachedThreadPool();
        IntStream.range(0, 10)
        .mapToObj(NapTask::new)
        .forEach(exec::execute);
        exec.shutdown();
    }
}

输出结果：

NapTask[7] pool-1-thread-8
NapTask[4] pool-1-thread-5
NapTask[1] pool-1-thread-2
NapTask[3] pool-1-thread-4
NapTask[0] pool-1-thread-1
NapTask[8] pool-1-thread-9
NapTask[2] pool-1-thread-3
NapTask[9] pool-1-thread-10
NapTask[6] pool-1-thread-7
NapTask[5] pool-1-thread-6

当你运行这个程序时，你会发现它完成得更快。这是有道理的，每个任务都有自己的线程，所以它们都并行运行，而不是使用相同的线程来顺序运行每个任务。这似乎没毛病，很难理解为什么有人会使用 SingleThreadExecutor。

要理解这个问题，我们需要一个更复杂的任务：

// concurrent/InterferingTask.java
public class InterferingTask implements Runnable {
    final int id;
    private static Integer val = 0;
    public InterferingTask(int id) {
        this.id = id;
    }
    @Override
    public void run() {
        for(int i = 0; i < 100; i++)
        val++;
    System.out.println(id + " "+
        Thread.currentThread().getName() + " " + val);
    }
}

每个任务增加 val 一百次。这似乎很简单。让我们用 CachedThreadPool 尝试一下：

// concurrent/CachedThreadPool2.java
import java.util.concurrent.*;
import java.util.stream.*;
public class CachedThreadPool2 {
    public static void main(String[] args) {
    ExecutorService exec
    =Executors.newCachedThreadPool();
    IntStream.range(0, 10)
    .mapToObj(InterferingTask::new)
    .forEach(exec::execute);
    exec.shutdown();
    }
}

输出结果：

0 pool-1-thread-1 200
1 pool-1-thread-2 200
4 pool-1-thread-5 300
5 pool-1-thread-6 400
8 pool-1-thread-9 500
9 pool-1-thread-10 600
2 pool-1-thread-3 700
7 pool-1-thread-8 800
3 pool-1-thread-4 900
6 pool-1-thread-7 1000

输出不是我们所期望的，并且从一次运行到下一次运行会有所不同。问题是所有的任务都试图写入 val 的单个实例，并且他们正在踩着彼此的脚趾。我们称这样的类是线程不安全的。让我们看看 SingleThreadExecutor 会发生什么：

// concurrent/SingleThreadExecutor3.java
import java.util.concurrent.*;
import java.util.stream.*;
public class SingleThreadExecutor3 {
    public static void main(String[] args)throws InterruptedException {
        ExecutorService exec
        =Executors.newSingleThreadExecutor();
        IntStream.range(0, 10)
        .mapToObj(InterferingTask::new)
        .forEach(exec::execute);
        exec.shutdown();
    }
}

输出结果：

0 pool-1-thread-1 100
1 pool-1-thread-1 200
2 pool-1-thread-1 300
3 pool-1-thread-1 400
4 pool-1-thread-1 500
5 pool-1-thread-1 600
6 pool-1-thread-1 700
7 pool-1-thread-1 800
8 pool-1-thread-1 900
9 pool-1-thread-1 1000

现在我们每次都得到一致的结果，尽管 InterferingTask 缺乏线程安全性。这是 SingleThreadExecutor 的主要好处 - 因为它一次运行一个任务，这些任务不会相互干扰，因此强加了线程安全性。这种现象称为线程封闭，因为在单线程上运行任务限制了它们的影响。线程封闭限制了加速，但可以节省很多困难的调试和重写。

• 产生结果

因为 InterferingTask 是一个 Runnable ，它没有返回值，因此只能使用副作用产生结果 - 操纵缓冲值而不是返回结果。副作用是并发编程中的主要问题之一，因为我们看到了 CachedThreadPool2.java 。InterferingTask 中的 val 被称为可变共享状态，这就是问题所在：多个任务同时修改同一个变量会产生竞争。结果取决于首先在终点线上执行哪个任务，并修改变量（以及其他可能性的各种变化）。

避免竞争条件的最好方法是避免可变的共享状态。我们可以称之为自私的孩子原则：什么都不分享。

使用 InterferingTask ，最好删除副作用并返回任务结果。为此，我们创建 Callable 而不是 Runnable ：

// concurrent/CountingTask.java
import java.util.concurrent.*;
public class CountingTask implements Callable<Integer> {
    final int id;
    public CountingTask(int id) { this.id = id; }
    @Override
    public Integer call() {
    Integer val = 0;
    for(int i = 0; i < 100; i++)
        val++;
    System.out.println(id + " " +
        Thread.currentThread().getName() + " " + val);
    return val;
    }
}

call() 完全独立于所有其他 CountingTasks 生成其结果，这意味着没有可变的共享状态

ExecutorService 允许你使用 invokeAll() 启动集合中的每个 Callable：

// concurrent/CachedThreadPool3.java
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;
public class CachedThreadPool3 {
    public static Integer extractResult(Future<Integer> f) {
        try {
            return f.get();
        } catch(Exception e) {
            throw new RuntimeException(e);
        }
    }
    public static void main(String[] args)throws InterruptedException {
    ExecutorService exec =
    Executors.newCachedThreadPool();
    List<CountingTask> tasks =
        IntStream.range(0, 10)
            .mapToObj(CountingTask::new)
            .collect(Collectors.toList());
        List<Future<Integer>> futures =
            exec.invokeAll(tasks);
        Integer sum = futures.stream()
            .map(CachedThreadPool3::extractResult)
            .reduce(0, Integer::sum);
        System.out.println("sum = " + sum);
        exec.shutdown();
    }
}

输出结果：

1 pool-1-thread-2 100
0 pool-1-thread-1 100
4 pool-1-thread-5 100
5 pool-1-thread-6 100
8 pool-1-thread-9 100
9 pool-1-thread-10 100
2 pool-1-thread-3 100
3 pool-1-thread-4 100
6 pool-1-thread-7 100
7 pool-1-thread-8 100
sum = 1000

只有在所有任务完成后，invokeAll() 才会返回一个 Future 列表，每个任务一个 Future 。Future 是 Java 5 中引入的机制，允许你提交任务而无需等待它完成。在这里，我们使用 ExecutorService.submit() :

// concurrent/Futures.java
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;
public class Futures {
    public static void main(String[] args)throws InterruptedException, ExecutionException {
    ExecutorService exec
        =Executors.newSingleThreadExecutor();
    Future<Integer> f =
        exec.submit(newCountingTask(99));
    System.out.println(f.get()); // [1]
    exec.shutdown();
    }
}

输出结果：

99 pool-1-thread-1 100
100

• [1] 当你的任务在尚未完成的 Future 上调用 get() 时，调用会阻塞（等待）直到结果可用。

但这意味着，在 CachedThreadPool3.java 中，Future 似乎是多余的，因为 invokeAll() 甚至在所有任务完成之前都不会返回。但是，这里的 Future 并不用于延迟结果，而是用于捕获任何可能发生的异常。

还要注意在 CachedThreadPool3.java.get() 中抛出异常，因此 extractResult() 在 Stream 中执行此提取。

因为当你调用 get() 时，Future 会阻塞，所以它只能解决等待任务完成才暴露问题。最终，Futures 被认为是一种无效的解决方案，现在不鼓励，我们推荐 Java 8 的 CompletableFuture ，这将在本章后面探讨。当然，你仍会在遗留库中遇到 Futures。

我们可以使用并行 Stream 以更简单，更优雅的方式解决这个问题:

// concurrent/CountingStream.java
// {VisuallyInspectOutput}
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;
public class CountingStream {
    public static void main(String[] args) {
        System.out.println(
            IntStream.range(0, 10)
                .parallel()
                .mapToObj(CountingTask::new)
                .map(ct -> ct.call())
                .reduce(0, Integer::sum));
    }
}

输出结果：

1 ForkJoinPool.commonPool-worker-3 100
8 ForkJoinPool.commonPool-worker-2 100
0 ForkJoinPool.commonPool-worker-6 100
2 ForkJoinPool.commonPool-worker-1 100
4 ForkJoinPool.commonPool-worker-5 100
9 ForkJoinPool.commonPool-worker-7 100
6 main 100
7 ForkJoinPool.commonPool-worker-4 100
5 ForkJoinPool.commonPool-worker-2 100
3 ForkJoinPool.commonPool-worker-3 100
1000

这不仅更容易理解，而且我们需要做的就是将 parallel() 插入到其他顺序操作中，然后一切都在同时运行。

• Lambda 和方法引用作为任务

在 java8 有了 lambdas 和方法引用，你不需要受限于只能使用 Runnable 和 Callable 。因为 java8 的 lambdas 和方法引用可以通过匹配方法签名来使用（即，它支持结构一致性），所以我们可以将非 Runnable 或 Callable 的参数传递给 ExecutorService :

// concurrent/LambdasAndMethodReferences.java
import java.util.concurrent.*;
class NotRunnable {
    public void go() {
        System.out.println("NotRunnable");
    }
}
class NotCallable {
    public Integer get() {
        System.out.println("NotCallable");
        return 1;
    }
}
public class LambdasAndMethodReferences {
    public static void main(String[] args)throws InterruptedException {
    ExecutorService exec =
        Executors.newCachedThreadPool();
    exec.submit(() -> System.out.println("Lambda1"));
    exec.submit(new NotRunnable()::go);
    exec.submit(() -> {
        System.out.println("Lambda2");
        return 1;
    });
    exec.submit(new NotCallable()::get);
    exec.shutdown();
    }
}

输出结果：

Lambda1
NotCallable
NotRunnable
Lambda2

这里，前两个 submit() 调用可以改为调用 execute() 。所有 submit() 调用都返回 Futures ，你可以在后两次调用的情况下提取结果。

终止耗时任务

并发程序通常使用长时间运行的任务。可调用任务在完成时返回值;虽然这给它一个有限的寿命，但仍然可能很长。可运行的任务有时被设置为永远运行的后台进程。你经常需要一种方法在正常完成之前停止 Runnable 和 Callable 任务，例如当你关闭程序时。

最初的 Java 设计提供了中断运行任务的机制（为了向后兼容，仍然存在）;中断机制包括阻塞问题。中断任务既乱又复杂，因为你必须了解可能发生中断的所有可能状态，以及可能导致的数据丢失。使用中断被视为反对模式，但我们仍然被迫接受。

InterruptedException，因为设计的向后兼容性残留。

任务终止的最佳方法是设置任务周期性检查的标志。然后任务可以通过自己的 shutdown 进程并正常终止。不是在任务中随机关闭线程，而是要求任务在到达了一个较好时自行终止。这总是产生比中断更好的结果，以及更容易理解的更合理的代码。

以这种方式终止任务听起来很简单：设置任务可以看到的 boolean flag。编写任务，以便定期检查标志并执行正常终止。这实际上就是你所做的，但是有一个复杂的问题：我们的旧克星，共同的可变状态。如果该标志可以被另一个任务操纵，则存在碰撞可能性。

在研究 Java 文献时，你会发现很多解决这个问题的方法，经常使用 volatile 关键字。我们将使用更简单的技术并避免所有易变的参数，这些都在附录：低级并发 中有所涉及。

Java 5 引入了 Atomic 类，它提供了一组可以使用的类型，而不必担心并发问题。我们将添加 AtomicBoolean 标志，告诉任务清理自己并退出。

// concurrent/QuittableTask.java
import java.util.concurrent.atomic.AtomicBoolean;import onjava.Nap;
public class QuittableTask implements Runnable {
    final int id;
    public QuittableTask(int id) {
        this.id = id;
    }
    private AtomicBoolean running =
        new AtomicBoolean(true);
    public void quit() {
        running.set(false);
    }
    @Override
    public void run() {
        while(running.get())         // [1]
            new Nap(0.1);
        System.out.print(id + " ");  // [2]
    }
}

虽然多个任务可以在同一个实例上成功调用 quit() ，但是 AtomicBoolean 可以防止多个任务同时实际修改 running ，从而使 quit() 方法成为线程安全的。

• [1]:只要运行标志为 true，此任务的 run() 方法将继续。

需要 running AtomicBoolean 证明编写 Java program 并发时最基本的困难之一是，如果 running 是一个普通的布尔值，你可能无法在执行程序中看到问题。实际上，在这个例子中，你可能永远不会有任何问题 - 但是代码仍然是不安全的。编写表明该问题的测试可能很困难或不可能。因此，你没有任何反馈来告诉你已经做错了。通常，你编写线程安全代码的唯一方法就是通过了解事情可能出错的所有细微之处。

作为测试，我们将启动很多 QuittableTasks 然后关闭它们。尝试使用较大的 COUNT 值

// concurrent/QuittingTasks.java
import java.util.*;
import java.util.stream.*;
import java.util.concurrent.*;
import onjava.Nap;
public class QuittingTasks {
    public static final int COUNT = 150;
    public static void main(String[] args) {
    ExecutorService es =
    Executors.newCachedThreadPool();
    List<QuittableTask> tasks =
    IntStream.range(1, COUNT)
        .mapToObj(QuittableTask::new)
        .peek(qt -> es.execute(qt))
        .collect(Collectors.toList());
    new Nap(1);
    tasks.forEach(QuittableTask::quit);    es.shutdown();
    }
}

输出结果：

24 27 31 8 11 7 19 12 16 4 23 3 28 32 15 20 63 60 68 6764 39 47 52 51 55 40 43 48 59 44 56 36 35 71 72 83 10396 92 88 99 100 87 91 79 75 84 76 115 108 112 104 107111 95 80 147 120 127 119 123 144 143 116 132 124 128
136 131 135 139 148 140 2 126 6 5 1 18 129 17 14 13 2122 9 10 30 33 58 37 125 26 34 133 145 78 137 141 138 6274 142 86 65 73 146 70 42 149 121 110 134 105 82 117106 113 122 45 114 118 38 50 29 90 101 89 57 53 94 4161 66 130 69 77 81 85 93 25 102 54 109 98 49 46 97

我使用 peek() 将 QuittableTasks 传递给 ExecutorService ，然后将这些任务收集到 List.main() 中，只要任何任务仍在运行，就会阻止程序退出。即使为每个任务按顺序调用 quit() 方法，任务也不会按照它们创建的顺序关闭。独立运行的任务不会确定性地响应信号。

CompletableFuture 类

作为介绍，这里是使用 CompletableFutures 在 QuittingTasks.java 中：

// concurrent/QuittingCompletable.java
import java.util.*;
import java.util.stream.*;
import java.util.concurrent.*;
import onjava.Nap;
public class QuittingCompletable {
    public static void main(String[] args) {
    List<QuittableTask> tasks =
        IntStream.range(1, QuittingTasks.COUNT)
            .mapToObj(QuittableTask::new)
            .collect(Collectors.toList());
        List<CompletableFuture<Void>> cfutures =
        tasks.stream()
            .map(CompletableFuture::runAsync)
            .collect(Collectors.toList());
        new Nap(1);
        tasks.forEach(QuittableTask::quit);
        cfutures.forEach(CompletableFuture::join);
    }
}

输出结果：

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 2526 27 28 29 30 31 32 33 34 6 35 4 38 39 40 41 42 43 4445 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 6263 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 8081 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 9899 100 101 102 103 104 105 106 107 108 109 110 111 1121 113 114 116 117 118 119 120 121 122 123 124 125 126127 128 129 130 131 132 133 134 135 136 137 138 139 140141 142 143 144 145 146 147 148 149 5 115 37 36 2 3

任务是一个 List<QuittableTask>，就像在 QuittingTasks.java 中一样，但是在这个例子中，没有 peek() 将每个 QuittableTask 提交给 ExecutorService。相反，在创建 cfutures 期间，每个任务都交给 CompletableFuture::runAsync。这执行 VerifyTask.run() 并返回 CompletableFuture<Void> 。因为 run() 不返回任何内容，所以在这种情况下我只使用 CompletableFuture 调用 join() 来等待它完成。

在本例中需要注意的重要一点是，运行任务不需要使用 ExecutorService。而是直接交给 CompletableFuture 管理 (不过你可以向它提供自己定义的 ExectorService)。您也不需要调用 shutdown();事实上，除非你像我在这里所做的那样显式地调用 join()，否则程序将尽快退出，而不必等待任务完成。

这个例子只是一个起点。你很快就会看到 ComplempleFuture 能够做得更多。

基本用法

这是一个带有静态方法 work() 的类，它对该类的对象执行某些工作：

// concurrent/Machina.java
import onjava.Nap;
public class Machina {
    public enum State {
        START, ONE, TWO, THREE, END;
        State step() {
            if(equals(END))
            return END;
          return values()[ordinal() + 1];
        }
    }
    private State state = State.START;
    private final int id;
    public Machina(int id) {
        this.id = id;
    }
    public static Machina work(Machina m) {
        if(!m.state.equals(State.END)){
            new Nap(0.1);
            m.state = m.state.step();
        }
        System.out.println(m);
        return m;
    }
    @Override
    public String toString() {
        return"Machina" + id + ": " + (state.equals(State.END)? "complete" : state);
    }
}

这是一个有限状态机，一个微不足道的机器，因为它没有分支……它只是从头到尾遍历一条路径。work() 方法将机器从一个状态移动到下一个状态，并且需要 100 毫秒才能完成“工作”。

CompletableFuture 可以被用来做的一件事是, 使用 completedFuture() 将它感兴趣的对象进行包装。

// concurrent/CompletedMachina.java
import java.util.concurrent.*;
public class CompletedMachina {
    public static void main(String[] args) {
        CompletableFuture<Machina> cf =
        CompletableFuture.completedFuture(
            new Machina(0));
        try {
            Machina m = cf.get();  // Doesn't block
        } catch(InterruptedException |
            ExecutionException e) {
        throw new RuntimeException(e);
        }
    }
}

completedFuture() 创建一个“已经完成”的 CompletableFuture 。对这样一个未来做的唯一有用的事情是 get() 里面的对象，所以这看起来似乎没有用。注意 CompletableFuture 被输入到它包含的对象。这个很重要。

通常，get() 在等待结果时阻塞调用线程。此块可以通过 InterruptedException 或 ExecutionException 中断。在这种情况下，阻止永远不会发生，因为 CompletableFuture 已经完成，所以结果立即可用。

当我们将 handle() 包装在 CompletableFuture 中时，发现我们可以在 CompletableFuture 上添加操作来处理所包含的对象，使得事情变得更加有趣：

// concurrent/CompletableApply.java
import java.util.concurrent.*;
public class CompletableApply {
    public static void main(String[] args) {
        CompletableFuture<Machina> cf =
        CompletableFuture.completedFuture(
            new Machina(0));
        CompletableFuture<Machina> cf2 =
            cf.thenApply(Machina::work);
        CompletableFuture<Machina> cf3 =
            cf2.thenApply(Machina::work);
        CompletableFuture<Machina> cf4 =
            cf3.thenApply(Machina::work);
        CompletableFuture<Machina> cf5 =
            cf4.thenApply(Machina::work);
    }
}

输出结果：

Machina0: ONE
Machina0: TWO
Machina0: THREE
Machina0: complete

thenApply() 应用一个接收输入并产生输出的函数。在本例中，work() 函数产生的类型与它所接收的类型相同 （Machina），因此每个 CompletableFuture添加的操作的返回类型都为 Machina，但是 (类似于流中的 map() ) 函数也可以返回不同的类型，这将体现在返回类型上。

你可以在此处看到有关 CompletableFutures 的重要信息：它们会在你执行操作时自动解包并重新包装它们所携带的对象。这使得编写和理解代码变得更加简单， 而不会在陷入在麻烦的细节中。

我们可以消除中间变量并将操作链接在一起，就像我们使用 Streams 一样：

// concurrent/CompletableApplyChained.javaimport java.util.concurrent.*;
import onjava.Timer;
public class CompletableApplyChained {
    public static void main(String[] args) {
        Timer timer = new Timer();
        CompletableFuture<Machina> cf =
        CompletableFuture.completedFuture(
            new Machina(0))
                  .thenApply(Machina::work)
                  .thenApply(Machina::work)
                  .thenApply(Machina::work)
                  .thenApply(Machina::work);
        System.out.println(timer.duration());
    }
}

输出结果：

Machina0: ONE
Machina0: TWO
Machina0: THREE
Machina0: complete
514

这里我们还添加了一个 Timer，它的功能在每一步都显性地增加 100ms 等待时间之外，还将 CompletableFuture 内部 thenApply 带来的额外开销给体现出来了。 CompletableFutures 的一个重要好处是它们鼓励使用私有子类原则（不共享任何东西）。默认情况下，使用 thenApply() 来应用一个不对外通信的函数 - 它只需要一个参数并返回一个结果。这是函数式编程的基础，并且它在并发特性方面非常有效⁵。并行流和 ComplempleFutures 旨在支持这些原则。只要你不决定共享数据（共享非常容易导致意外发生）你就可以编写出相对安全的并发程序。

回调 thenApply() 一旦开始一个操作，在完成所有任务之前，不会完成 CompletableFuture 的构建。虽然这有时很有用，但是开始所有任务通常更有价值，这样就可以运行继续前进并执行其他操作。我们可通过thenApplyAsync() 来实现此目的：

// concurrent/CompletableApplyAsync.java
import java.util.concurrent.*;
import onjava.*;
public class CompletableApplyAsync {
    public static void main(String[] args) {
        Timer timer = new Timer();
        CompletableFuture<Machina> cf =
            CompletableFuture.completedFuture(
                new Machina(0))
                .thenApplyAsync(Machina::work)
                .thenApplyAsync(Machina::work)
                .thenApplyAsync(Machina::work)
                .thenApplyAsync(Machina::work);
            System.out.println(timer.duration());
            System.out.println(cf.join());
            System.out.println(timer.duration());
    }
}

输出结果：

116
Machina0: ONE
Machina0: TWO
Machina0:THREE
Machina0: complete
Machina0: complete
552

同步调用 (我们通常使用的那种) 意味着：“当你完成工作时，才返回”，而异步调用以意味着： “立刻返回并继续后续工作”。 正如你所看到的，cf 的创建现在发生的更快。每次调用 thenApplyAsync() 都会立刻返回，因此可以进行下一次调用，整个调用链路完成速度比以前快得多。

事实上，如果没有回调 cf.join() 方法，程序会在完成其工作之前退出。而 cf.join() 直到 cf 操作完成之前，阻止 main() 进程结束。我们还可以看出本示例大部分时间消耗在 cf.join() 这。

这种“立即返回”的异步能力需要 CompletableFuture 库进行一些秘密（client 无感）工作。特别是，它将你需要的操作链存储为一组回调。当操作的第一个链路（后台操作）完成并返回时，第二个链路（后台操作）必须获取生成的 Machina 并开始工作，以此类推！ 但这种异步机制没有我们可以通过程序调用栈控制的普通函数调用序列，它的调用链路顺序会丢失，因此它使用一个函数地址来存储的回调来解决这个问题。

幸运的是，这就是你需要了解的有关回调的全部信息。程序员将这种人为制造的混乱称为 callback hell（回调地狱）。通过异步调用，CompletableFuture 帮你管理所有回调。 除非你知道你系统中的一些特定逻辑会导致某些改变，或许你更想使用异步调用来实现程序。

• 其他操作

当你查看CompletableFuture的 Javadoc 时，你会看到它有很多方法，但这个方法的大部分来自不同操作的变体。例如，有 thenApply()，thenApplyAsync() 和第二种形式的 thenApplyAsync()，它们使用 Executor 来运行任务 (在本书中，我们忽略了 Executor 选项)。

下面的示例展示了所有"基本"操作，这些操作既不涉及组合两个 CompletableFuture，也不涉及异常 (我们将在后面介绍)。首先，为了提供简洁性和方便性，我们应该重用以下两个实用程序:

package onjava;
import java.util.concurrent.*;

public class CompletableUtilities {
  // Get and show value stored in a CF:
  public static void showr(CompletableFuture<?> c) {
    try {
      System.out.println(c.get());
    } catch(InterruptedException
            | ExecutionException e) {
      throw new RuntimeException(e);
    }
  }
  // For CF operations that have no value:
  public static void voidr(CompletableFuture<Void> c) {
    try {
      c.get(); // Returns void
    } catch(InterruptedException
            | ExecutionException e) {
      throw new RuntimeException(e);
    }
  }
}

showr() 在 CompletableFuture<Integer> 上调用 get()，并显示结果，try/catch 两个可能会出现的异常。

voidr() 是 CompletableFuture<Void> 的 showr() 版本，也就是说，CompletableFutures 只为任务完成或失败时显示信息。

为简单起见，下面的 CompletableFutures 只包装整数。cfi() 是一个便利的方法，它把一个整数包装在一个完整的 CompletableFuture<Integer> :

// concurrent/CompletableOperations.java
import java.util.concurrent.*;
import static onjava.CompletableUtilities.*;

public class CompletableOperations {
    static CompletableFuture<Integer> cfi(int i) {
        return
                CompletableFuture.completedFuture(
                        Integer.valueOf(i));
    }

    public static void main(String[] args) {
        showr(cfi(1)); // Basic test
        voidr(cfi(2).runAsync(() ->
                System.out.println("runAsync")));
        voidr(cfi(3).thenRunAsync(() ->
                System.out.println("thenRunAsync")));
        voidr(CompletableFuture.runAsync(() ->
                System.out.println("runAsync is static")));
        showr(CompletableFuture.supplyAsync(() -> 99));
        voidr(cfi(4).thenAcceptAsync(i ->
                System.out.println("thenAcceptAsync: " + i)));
        showr(cfi(5).thenApplyAsync(i -> i + 42));
        showr(cfi(6).thenComposeAsync(i -> cfi(i + 99)));
        CompletableFuture<Integer> c = cfi(7);
        c.obtrudeValue(111);
        showr(c);
        showr(cfi(8).toCompletableFuture());
        c = new CompletableFuture<>();
        c.complete(9);
        showr(c);
        c = new CompletableFuture<>();
        c.cancel(true);
        System.out.println("cancelled: " +
                c.isCancelled());
        System.out.println("completed exceptionally: " +
                c.isCompletedExceptionally());
        System.out.println("done: " + c.isDone());
        System.out.println(c);
        c = new CompletableFuture<>();
        System.out.println(c.getNow(777));
        c = new CompletableFuture<>();
        c.thenApplyAsync(i -> i + 42)
                .thenApplyAsync(i -> i * 12);
        System.out.println("dependents: " +
                c.getNumberOfDependents());
        c.thenApplyAsync(i -> i / 2);
        System.out.println("dependents: " +
                c.getNumberOfDependents());
    }
}

输出结果 ：

1
runAsync
thenRunAsync
runAsync is static
99
thenAcceptAsync: 4
47
105
111
8
9
cancelled: true
completed exceptionally: true
done: true
java.util.concurrent.CompletableFuture@6d311334[Complet ed exceptionally]
777
dependents: 1
dependents: 2

• main() 包含一系列可由其 int 值引用的测试。
  • cfi(1) 演示了 showr() 正常工作。
  • cfi(2) 是调用 runAsync() 的示例。由于 Runnable 不产生返回值，因此使用了返回 CompletableFuture <Void> 的voidr() 方法。
  • 注意使用 cfi(3),thenRunAsync() 效果似乎与 上例 cfi(2) 使用的 runAsync()相同，差异在后续的测试中体现：
    • runAsync() 是一个 static 方法，所以你通常不会像cfi(2)一样调用它。相反你可以在 QuittingCompletable.java 中使用它。
    • 后续测试中表明 supplyAsync() 也是静态方法，区别在于它需要一个 Supplier 而不是Runnable, 并产生一个CompletableFuture<Integer> 而不是 CompletableFuture<Void>。
  • then 系列方法将对现有的 CompletableFuture<Integer> 进一步操作。
    • 与 thenRunAsync() 不同，cfi(4)，cfi(5) 和cfi(6) “then” 方法的参数是未包装的 Integer。
    • 通过使用 voidr()方法可以看到:
      • AcceptAsync()接收了一个 Consumer，因此不会产生结果。
      • thenApplyAsync() 接收一个Function, 并生成一个结果（该结果的类型可以不同于其输入类型）。
      • thenComposeAsync() 与 thenApplyAsync()非常相似，唯一区别在于其 Function 必须产生已经包装在CompletableFuture中的结果。
  • cfi(7) 示例演示了 obtrudeValue()，它强制将值作为结果。
  • cfi(8) 使用 toCompletableFuture() 从 CompletionStage 生成一个CompletableFuture。
  • c.complete(9) 显示了如何通过给它一个结果来完成一个task（future）（与 obtrudeValue() 相对，后者可能会迫使其结果替换该结果）。
  • 如果你调用 CompletableFuture中的 cancel()方法，如果已经完成此任务，则正常结束。 如果尚未完成，则使用 CancellationException 完成此 CompletableFuture。
  • 如果任务（future）完成，则 getNow() 方法返回CompletableFuture的完成值，否则返回getNow()的替换参数。
  • 最后，我们看一下依赖 (dependents) 的概念。如果我们将两个thenApplyAsync()调用链路到CompletableFuture上，则依赖项的数量不会增加，保持为 1。但是，如果我们另外将另一个thenApplyAsync()直接附加到c，则现在有两个依赖项：两个一起的链路和另一个单独附加的链路。
    • 这表明你可以使用一个CompletionStage，当它完成时，可以根据其结果派生多个新任务。

结合 CompletableFuture

第二种类型的 CompletableFuture 方法采用两种 CompletableFuture 并以各异方式将它们组合在一起。就像两个人在比赛一样, 一个CompletableFuture通常比另一个更早地到达终点。这些方法允许你以不同的方式处理结果。 为了测试这一点，我们将创建一个任务，它有一个我们可以控制的定义了完成任务所需要的时间量的参数。 CompletableFuture 先完成:

// concurrent/Workable.java
import java.util.concurrent.*;
import onjava.Nap;

public class Workable {
    String id;
    final double duration;

    public Workable(String id, double duration) {
        this.id = id;
        this.duration = duration;
    }

    @Override
    public String toString() {
        return "Workable[" + id + "]";
    }

    public static Workable work(Workable tt) {
        new Nap(tt.duration); // Seconds
        tt.id = tt.id + "W";
        System.out.println(tt);
        return tt;
    }

    public static CompletableFuture<Workable> make(String id, double duration) {
        return CompletableFuture
                .completedFuture(
                        new Workable(id, duration)
                )
                .thenApplyAsync(Workable::work);
    }
}

在 make()中，work()方法应用于CompletableFuture。work()需要一定的时间才能完成，然后它将字母 W 附加到 id 上，表示工作已经完成。

现在我们可以创建多个竞争的 CompletableFuture，并使用 CompletableFuture 库中的各种方法来进行操作:

// concurrent/DualCompletableOperations.java
import java.util.concurrent.*;
import static onjava.CompletableUtilities.*;

public class DualCompletableOperations {
    static CompletableFuture<Workable> cfA, cfB;

    static void init() {
        cfA = Workable.make("A", 0.15);
        cfB = Workable.make("B", 0.10); // Always wins
    }

    static void join() {
        cfA.join();
        cfB.join();
        System.out.println("*****************");
    }

    public static void main(String[] args) {
        init();
        voidr(cfA.runAfterEitherAsync(cfB, () ->
                System.out.println("runAfterEither")));
        join();

        init();
        voidr(cfA.runAfterBothAsync(cfB, () ->
                System.out.println("runAfterBoth")));
        join();

        init();
        showr(cfA.applyToEitherAsync(cfB, w -> {
            System.out.println("applyToEither: " + w);
            return w;
        }));
        join();

        init();
        voidr(cfA.acceptEitherAsync(cfB, w -> {
            System.out.println("acceptEither: " + w);
        }));
        join();

        init();
        voidr(cfA.thenAcceptBothAsync(cfB, (w1, w2) -> {
            System.out.println("thenAcceptBoth: "
                    + w1 + ", " + w2);
        }));
        join();

        init();
        showr(cfA.thenCombineAsync(cfB, (w1, w2) -> {
            System.out.println("thenCombine: "
                    + w1 + ", " + w2);
            return w1;
        }));
        join();

        init();
        CompletableFuture<Workable>
                cfC = Workable.make("C", 0.08),
                cfD = Workable.make("D", 0.09);
        CompletableFuture.anyOf(cfA, cfB, cfC, cfD)
                .thenRunAsync(() ->
                        System.out.println("anyOf"));
        join();

        init();
        cfC = Workable.make("C", 0.08);
        cfD = Workable.make("D", 0.09);
        CompletableFuture.allOf(cfA, cfB, cfC, cfD)
                .thenRunAsync(() ->
                        System.out.println("allOf"));
        join();
    }
}

输出结果：

Workable[BW]
runAfterEither
Workable[AW]
*****************
Workable[BW]
Workable[AW]
runAfterBoth
*****************
Workable[BW]
applyToEither: Workable[BW]
Workable[BW]
Workable[AW]
*****************
Workable[BW]
acceptEither: Workable[BW]
Workable[AW]
*****************
Workable[BW]
Workable[AW]
thenAcceptBoth: Workable[AW], Workable[BW]
****************
 Workable[BW]
 Workable[AW]
 thenCombine: Workable[AW], Workable[BW]
 Workable[AW]
 *****************
 Workable[CW]
 anyOf
 Workable[DW]
 Workable[BW]
 Workable[AW]
 *****************
 Workable[CW]
 Workable[DW]
 Workable[BW]
 Workable[AW]
 *****************
 allOf

• 为了方便访问， 将 cfA 和 cfB 定义为 static的。
  • init()方法用于 A, B 初始化这两个变量，因为 B 总是给出比A较短的延迟，所以总是 win 的一方。
  • join() 是在两个方法上调用 join() 并显示边框的另一个便利方法。
• 所有这些 “dual” 方法都以一个 CompletableFuture 作为调用该方法的对象，第二个 CompletableFuture 作为第一个参数，然后是要执行的操作。
• 通过使用 showr() 和 voidr() 可以看到，“run”和“accept”是终端操作，而“apply”和“combine”则生成新的 payload-bearing (承载负载) 的 CompletableFuture。
• 方法的名称不言自明，你可以通过查看输出来验证这一点。一个特别有趣的方法是 combineAsync()，它等待两个 CompletableFuture 完成，然后将它们都交给一个 BiFunction，这个 BiFunction 可以将结果加入到最终的 CompletableFuture 的有效负载中。

模拟

作为使用 CompletableFuture 将一系列操作组合的示例，让我们模拟一下制作蛋糕的过程。在第一阶段，我们准备并将原料混合成面糊:

// concurrent/Batter.java
import java.util.concurrent.*;
import onjava.Nap;

public class Batter {
    static class Eggs {
    }

    static class Milk {
    }

    static class Sugar {
    }

    static class Flour {
    }

    static <T> T prepare(T ingredient) {
        new Nap(0.1);
        return ingredient;
    }

    static <T> CompletableFuture<T> prep(T ingredient) {
        return CompletableFuture
                .completedFuture(ingredient)
                .thenApplyAsync(Batter::prepare);
    }

    public static CompletableFuture<Batter> mix() {
        CompletableFuture<Eggs> eggs = prep(new Eggs());
        CompletableFuture<Milk> milk = prep(new Milk());
        CompletableFuture<Sugar> sugar = prep(new Sugar());
        CompletableFuture<Flour> flour = prep(new Flour());
        CompletableFuture
                .allOf(eggs, milk, sugar, flour)
                .join();
        new Nap(0.1); // Mixing time
        return CompletableFuture.completedFuture(new Batter());
    }
}

每种原料都需要一些时间来准备。allOf() 等待所有的配料都准备好，然后使用更多些的时间将其混合成面糊。接下来，我们把单批面糊放入四个平底锅中烘烤。产品作为 CompletableFutures 流返回：

// concurrent/Baked.java

import java.util.concurrent.*;
import java.util.stream.*;
import onjava.Nap;

public class Baked {
    static class Pan {
    }

    static Pan pan(Batter b) {
        new Nap(0.1);
        return new Pan();
    }

    static Baked heat(Pan p) {
        new Nap(0.1);
        return new Baked();
    }

    static CompletableFuture<Baked> bake(CompletableFuture<Batter> cfb) {
        return cfb
                .thenApplyAsync(Baked::pan)
                .thenApplyAsync(Baked::heat);
    }

    public static Stream<CompletableFuture<Baked>> batch() {
        CompletableFuture<Batter> batter = Batter.mix();
        return Stream.of(
                bake(batter),
                bake(batter),
                bake(batter),
                bake(batter)
        );
    }
}

最后，我们制作了一批糖，并用它对蛋糕进行糖化：

// concurrent/FrostedCake.java

import java.util.concurrent.*;
import java.util.stream.*;
import onjava.Nap;

final class Frosting {
    private Frosting() {
    }

    static CompletableFuture<Frosting> make() {
        new Nap(0.1);
        return CompletableFuture
                .completedFuture(new Frosting());
    }
}

public class FrostedCake {
    public FrostedCake(Baked baked, Frosting frosting) {
        new Nap(0.1);
    }

    @Override
    public String toString() {
        return "FrostedCake";
    }

    public static void main(String[] args) {
        Baked.batch().forEach(
                baked -> baked
                        .thenCombineAsync(Frosting.make(),
                                (cake, frosting) ->
                                        new FrostedCake(cake, frosting))
                        .thenAcceptAsync(System.out::println)
                        .join());
    }
}

一旦你习惯了这种背后的想法, CompletableFuture 它们相对易于使用。

异常

与 CompletableFuture 在处理链中包装对象的方式相同，它也会缓冲异常。这些在处理时调用者是无感的，但仅当你尝试提取结果时才会被告知。为了说明它们是如何工作的，我们首先创建一个类，它在特定的条件下抛出一个异常:

// concurrent/Breakable.java
import java.util.concurrent.*;
public class Breakable {
    String id;
    private int failcount;

    public Breakable(String id, int failcount) {
        this.id = id;
        this.failcount = failcount;
    }

    @Override
    public String toString() {
        return "Breakable_" + id + " [" + failcount + "]";
    }

    public static Breakable work(Breakable b) {
        if (--b.failcount == 0) {
            System.out.println(
                    "Throwing Exception for " + b.id + ""
            );
            throw new RuntimeException(
                    "Breakable_" + b.id + " failed"
            );
        }
        System.out.println(b);
        return b;
    }
}

当failcount > 0，且每次将对象传递给 work() 方法时， failcount - 1 。当failcount - 1 = 0 时，work() 将抛出一个异常。如果传给 work() 的 failcount = 0 ，work() 永远不会抛出异常。

注意，异常信息此示例中被抛出（ RuntimeException )

在下面示例 test() 方法中，work() 多次应用于 Breakable，因此如果 failcount 在范围内，就会抛出异常。然而，在测试A到E中，你可以从输出中看到抛出了异常，但它们从未出现:

// concurrent/CompletableExceptions.java
import java.util.concurrent.*;
public class CompletableExceptions {
    static CompletableFuture<Breakable> test(String id, int failcount) {
        return CompletableFuture.completedFuture(
                new Breakable(id, failcount))
                .thenApply(Breakable::work)
                .thenApply(Breakable::work)
                .thenApply(Breakable::work)
                .thenApply(Breakable::work);
    }

    public static void main(String[] args) {
        // Exceptions don't appear ...
        test("A", 1);
        test("B", 2);
        test("C", 3);
        test("D", 4);
        test("E", 5);
        // ... until you try to fetch the value:
        try {
            test("F", 2).get(); // or join()
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
        // Test for exceptions:
        System.out.println(
                test("G", 2).isCompletedExceptionally()
        );
        // Counts as "done":
        System.out.println(test("H", 2).isDone());
        // Force an exception:
        CompletableFuture<Integer> cfi =
                new CompletableFuture<>();
        System.out.println("done? " + cfi.isDone());
        cfi.completeExceptionally(
                new RuntimeException("forced"));
        try {
            cfi.get();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}

输出结果：

Throwing Exception for A
Breakable_B [1]
Throwing Exception for B
Breakable_C [2]
Breakable_C [1]
Throwing Exception for C
Breakable_D [3]
Breakable_D [2]
Breakable_D [1]
Throwing Exception for D
Breakable_E [4]
Breakable_E [3]
Breakable_E [2]
Breakable_E [1]
Breakable_F [1]
Throwing Exception for F
java.lang.RuntimeException: Breakable_F failed
Breakable_G [1]
Throwing Exception for G
true
Breakable_H [1]
Throwing Exception for H
true
done? false
java.lang.RuntimeException: forced

测试 A 到 E 运行到抛出异常，然后…并没有将抛出的异常暴露给调用方。只有在测试 F 中调用 get() 时，我们才会看到抛出的异常。 测试 G 表明，你可以首先检查在处理期间是否抛出异常，而不抛出该异常。然而，test H 告诉我们，不管异常是否成功，它仍然被视为已“完成”。 代码的最后一部分展示了如何将异常插入到 CompletableFuture 中，而不管是否存在任何失败。 在连接或获取结果时，我们使用 CompletableFuture 提供的更复杂的机制来自动响应异常，而不是使用粗糙的 try-catch。 你可以使用与我们看到的所有 CompletableFuture 相同的表单来完成此操作:在链中插入一个 CompletableFuture 调用。有三个选项 exceptionally()，handle()， whenComplete():

// concurrent/CatchCompletableExceptions.java
import java.util.concurrent.*;
public class CatchCompletableExceptions {
    static void handleException(int failcount) {
        // Call the Function only if there's an
        // exception, must produce same type as came in:
        CompletableExceptions
                .test("exceptionally", failcount)
                .exceptionally((ex) -> { // Function
                    if (ex == null)
                        System.out.println("I don't get it yet");
                    return new Breakable(ex.getMessage(), 0);
                })
                .thenAccept(str ->
                        System.out.println("result: " + str));

        // Create a new result (recover):
        CompletableExceptions
                .test("handle", failcount)
                .handle((result, fail) -> { // BiFunction
                    if (fail != null)
                        return "Failure recovery object";
                    else
                        return result + " is good";
                })
                .thenAccept(str ->
                        System.out.println("result: " + str));

        // Do something but pass the same result through:
        CompletableExceptions
                .test("whenComplete", failcount)
                .whenComplete((result, fail) -> { // BiConsumer
                    if (fail != null)
                        System.out.println("It failed");
                    else
                        System.out.println(result + " OK");
                })
                .thenAccept(r ->
                        System.out.println("result: " + r));
    }

    public static void main(String[] args) {
        System.out.println("**** Failure Mode ****");
        handleException(2);
        System.out.println("**** Success Mode ****");
        handleException(0);
    }
}