第二十四章 并发编程

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

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

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

猫咪：“你一定是疯了，否则你就不会来这儿” ——爱丽丝梦游仙境 第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 8Streams就是一个很好的例子）。

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

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

支持并发性的语言和库似乎是抽象泄露（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进程并正常终止。不是在任务中随机关闭线程，而是要求任务在到达了一个较好时自行终止。这总是产生比中断更好的结果，以及更容易理解的更合理的代码。

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

在研究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())函数也可以返回不同的类型，这将体现在返回类型上。