From c1f70fe70b96360a53f2da0e53a4422f6937c903 Mon Sep 17 00:00:00 2001 From: AlfredAlan <854723360@mail.com> Date: Fri, 19 Jul 2019 01:47:12 +0800 Subject: [PATCH] =?UTF-8?q?=E9=99=84=E5=BD=95=EF=BC=9A=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E5=BA=95=E5=B1=82=E5=8E=9F=E7=90=86=20=E7=BF=BB=E8=AF=91?= =?UTF-8?q?=E8=87=B3=E8=B5=84=E6=BA=90=E7=AB=9E=E4=BA=89=E5=B0=8F=E8=8A=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/book/Appendix-Low-Level-Concurrency.md | 216 +++++++++++++++++++- 1 file changed, 208 insertions(+), 8 deletions(-) diff --git a/docs/book/Appendix-Low-Level-Concurrency.md b/docs/book/Appendix-Low-Level-Concurrency.md index e09e7be..c06c5dd 100644 --- a/docs/book/Appendix-Low-Level-Concurrency.md +++ b/docs/book/Appendix-Low-Level-Concurrency.md @@ -14,7 +14,7 @@ ## 什么是线程？ -并发将程序划分成分离的，独立运行的任务。每个任务都由一个 *执行线程* 来驱动，我们通常将其简称为 *线程* 。而一个 *线程* 就是操作系统进程中单一顺序的控制流。因此，单个进程可以有多个并发执行的任务，但是你的程序使得每个任务都好像有自己的处理器一样。这线程模型为编程带来了便利，它简化了在单一程序中处理变戏法般的多任务过程。操作系统则从处理器上分配时间到您程序的所有线程中。 +并发将程序划分成独立分离运行的任务。每个任务都由一个 *执行线程* 来驱动，我们通常将其简称为 *线程* 。而一个 *线程* 就是操作系统进程中单一顺序的控制流。因此，单个进程可以有多个并发执行的任务，但是你的程序使得每个任务都好像有自己的处理器一样。这线程模型为编程带来了便利，它简化了在单一程序中处理变戏法般的多任务过程。操作系统则从处理器上分配时间片到你程序的所有线程中。 Java 并发的核心机制是 **Thread** 类，在该语言最初版本中， **Thread （线程）** 是由程序员直接创建和管理的。随着语言的发展以及人们发现了更好的一些方法，中间层机制 - 特别是 **Executor** 框架 - 被添加进来，以消除自己管理线程时候的心理负担（及错误）。 最终，甚至发展出比 **Executor** 更好的机制，如 [并发编程](./24-Concurrent-Programming.md) 一章所示。 @@ -347,9 +347,9 @@ caught java.lang.RuntimeException */ ``` -额外会跟踪验证工厂对象创建的线程是否获得新 **UncaughtExceptionHandler** 。现在未捕获的异常由 **uncaughtException** 方法捕获。 +额外会在代码中添加跟踪机制，用来验证工厂对象创建的线程是否获得新 **UncaughtExceptionHandler** 。现在未捕获的异常由 **uncaughtException** 方法捕获。 -上面的示例根据具体情况来设置处理对象。如果你明白你想要在任何地方使用相同的异常处理对象，一个更简单的方法是设置默认的未捕获异常处理对象，它定义在 **Thread** 类中作为一个 **static**（静态） 字段： +上面的示例根据具体情况来设置处理器。如果你知道你将要在代码中处处使用相同的异常处理器，那么更简单的方式是在 **Thread** 类中设置一个 **static**（静态） 字段，并将这个处理器设置为默认的未捕获异常处理器： ```java // lowlevel/SettingDefaultHandler.java @@ -370,7 +370,7 @@ caught java.lang.RuntimeException */ ``` -只有在每个线程没有设置异常处理对象时候，默认处理对象才会被调用。系统会检查每个线程的版本，如果没有找到，则检查是否线程组中有专门的 **uncaughtException()** 方法；如果都没有，就会调用 **defaultUncaughtExceptionHandler** 方法。 +只有在每个线程没有设置异常处理器时候，默认处理器才会被调用。系统会检查线程专有的版本，如果没有，则检查是否线程组中有专有的 **uncaughtException()** 方法；如果都没有，就会调用 **defaultUncaughtExceptionHandler** 方法。 可以将此方法与 **CompletableFuture**s 的改进方法进行比较。 @@ -379,17 +379,212 @@ caught java.lang.RuntimeException 你可以将单线程程序看作一个孤独的实体，在你的问题空间中移动并一次只做一件事。因为只有一个实体，你永远不会想到两个实体试图同时使用相同资源的问题：问题犹如两个人试图同时停放在同一个空间，同时走过一扇门，甚至同时说话。 -通过并发，事情不再孤单，但现在两个或更多任务可能会相互干扰。如果您不阻止这种冲突，您将有两个任务同时尝试访问同一个银行帐户，打印到同一个打印机，调整相同的阀门，等等。 +通过并发，事情不再孤单，但现在两个或更多任务可能会相互干扰。如果您不阻止这种冲突，您将有两个任务同时尝试访问同一个银行帐户，打印到同一个打印机，调整同一个阀门，等等。 ### 资源竞争 +当你启动一个任务来执行某些工作时，可以通过两种不同的方式捕获该工作的结果:通过副作用或通过返回值。 + +从编程方式上看，副作用似乎更容易:你只需使用结果来操作环境中的某些东西。例如，你的任务可能会执行一些计算，然后直接将其结果写入集合。 + +这种方法的问题是集合通常是共享资源。当运行多个任务时，任何任务都可能同时读写 *共享资源* 。这揭示了 *资源竞争* 问题，这是处理任务时的主要陷阱之一。 + +在单线程系统中，您不会考虑资源竞争，因为你一次只做一件事。当你有多个任务时，必须始终防止资源竞争。 + +解决此问题的的一种方法是使用能够应对资源竞争的集合，如果多个任务同时尝试对此类集合进行写入，那么此类集合可以应付该问题。在 Java 并发库中，你将发现许多尝试解决资源争用问题的类；在本附录中，您将看到其中的一些，但覆盖范围并不全面。 + +请思考以下的示例，其中一个任务负责生成偶数，其他任务则负责消费这些数字。在这里，消费者任务的唯一工作就是检查偶数的有效性。 + +我们将定义消费者任务 **EvenChecker** 类，以便在后续示例中可复用。为了将 **EvenChecker** 与我们的各种实验生成器类解耦，我们首先创建一个名为 **IntGenerator** 的抽象类，它包含 **EvenChecker** 必须知道的最少必要方法：它包含一个 **next()** 方法，以及可以取消生成的方法。 + +```java +// lowlevel/IntGenerator.java +import java.util.concurrent.atomic.AtomicBoolean; + +public abstract class IntGenerator { + private AtomicBoolean canceled = + new AtomicBoolean(); + public abstract int next(); + public void cancel() { canceled.set(true); } + public boolean isCanceled() { + return canceled.get(); + } +} +``` + +**cancel()** 方法改变 **AtomicBoolean canceled** 标志位的状态， 而 **isCanceled()** 方法则告诉标志位是否设置。因为 **canceled** 标志位是 **AtomicBoolean** 类型，所以它是原子性的，这意味着分配和值返回等简单操作发生时没有中断的可能性，因此你无法在这些简单操作中看到该字段处于中间状态。您将在本附录的后面部分了解有关原子性和 **Atomic** 类的更多信息 + +任何 **IntGenerator** 都可以使用下面的 **EvenChecker** 类进行测试: + +```java +// lowlevel/EvenChecker.java +import java.util.*; +import java.util.stream.*; +import java.util.concurrent.*; +import onjava.TimedAbort; + +public class EvenChecker implements Runnable { + private IntGenerator generator; + private final int id; + public EvenChecker(IntGenerator generator, int id) { + this.generator = generator; + this.id = id; + } + @Override + public void run() { + while(!generator.isCanceled()) { + int val = generator.next(); + if(val % 2 != 0) { + System.out.println(val + " not even!"); + generator.cancel(); // Cancels all EvenCheckers + } + } + } + // Test any IntGenerator: + public static void test(IntGenerator gp, int count) { + List> checkers = + IntStream.range(0, count) + .mapToObj(i -> new EvenChecker(gp, i)) + .map(CompletableFuture::runAsync) + .collect(Collectors.toList()); + checkers.forEach(CompletableFuture::join); + } + // Default value for count: + public static void test(IntGenerator gp) { + new TimedAbort(4, "No odd numbers discovered"); + test(gp, 10); + } +} +``` + +**test()** 方法开启了许多访问同一个 **IntGenerator** 的 **EvenChecker**。**EvenChecker** 任务们会不断读取和测试与其关联的 **IntGenerator** 对象中的生成值。如果 **IntGenerator** 导致失败，**test()** 方法会报告并返回。 + +依赖于 **IntGenerator** 对象的所有 **EvenChecker** 任务都会检查它是否已被取消。如果 **generator.isCanceled()** 返回值为 true ，则 **run()** 方法返回。 任何 **EvenChecker** 任务都可以在 **IntGenerator** 上调用**cancel()** ，这会导致使用该 **IntGenerator** 的其他所有 **EvenChecker** 正常关闭。 + +在本设计中，共享公共资源（ **IntGenerator** ）的任务会监视该资源的终止信号。这消除所谓的竞争条件，其中两个或更多的任务竞争响应某个条件并因此冲突或不一致结果的情况。 + +你必须仔细考虑并防止并发系统失败的所有可能途径。例如，一个任务不能依赖于另一个任务，因为任务关闭的顺序无法得到保证。这里，通过使任务依赖于非任务对象，我们可以消除潜在的竞争条件。 + +一般来说，我们假设 **test()** 方法最终失败，因为各个 **EvenChecker** 的任务在 **IntGenerator** 处于 “不恰当的” 状态时，仍能够访问其中的信息。但是，直到 **IntGenerator** 完成许多循环之前，它可能无法检测到问题，具体取决于操作系统的详细信息和其他实现细节。为确保本书的自动构建不会卡住，我们使用 **TimedAbort** 类，在此处定义： + +```java +// onjava/TimedAbort.java +// Terminate a program after t seconds +package onjava; +import java.util.concurrent.*; + +public class TimedAbort { + private volatile boolean restart = true; + public TimedAbort(double t, String msg) { + CompletableFuture.runAsync(() -> { + try { + while(restart) { + restart = false; + TimeUnit.MILLISECONDS + .sleep((int)(1000 * t)); + } + } catch(InterruptedException e) { + throw new RuntimeException(e); + } + System.out.println(msg); + System.exit(0); + }); + } + public TimedAbort(double t) { + this(t, "TimedAbort " + t); + } + public void restart() { restart = true; } +} +``` + +我们使用 lambda 表达式创建一个 **Runnable** ，该表达式使用 **CompletableFuture** 的 **runAsync()** 静态方法执行。 **runAsync()** 方法的值会立即返回。 因此，**TimedAbort** 不会保持任何打开的任务，否则已完成任务，但如果它需要太长时间，它仍将终止该任务（ **TimedAbort** 有时被称为守护进程）。 + +**TimedAbort** 还允许你 **restart()** 方法重启任务，在有某些有用的活动进行时保持程序打开。 + +我们可以看到正在运行的 **TimedAbort** 示例: + +```java +// lowlevel/TestAbort.java +import onjava.*; + +public class TestAbort { + public static void main(String[] args) { + new TimedAbort(1); + System.out.println("Napping for 4"); + new Nap(4); + } +} +/* Output: +Napping for 4 +TimedAbort 1.0 +*/ +``` + +如果你注释掉 **Nap** 创建实列那行，程序执行会立即退出，表明 **TimedAbort** 没有维持程序打开。 + +我们将看到第一个 **IntGenerator** 示例有一个生成一系列偶数值的 **next()** 方法： + +```java +// lowlevel/EvenProducer.java +// When threads collide +// {VisuallyInspectOutput} + +public class EvenProducer extends IntGenerator { + private int currentEvenValue = 0; + @Override + public int next() { + ++currentEvenValue; // [1] + ++currentEvenValue; + return currentEvenValue; + } + public static void main(String[] args) { + EvenChecker.test(new EvenProducer()); + } +} +/* Output: +419 not even! +425 not even! +423 not even! +421 not even! +417 not even! +*/ +``` +* [1] 一个任务有可能在另外一个任务执行第一个对 **currentEvenValue** 的递增操作之后，但是没有执行第二个操作之前，调用 **next()** 方法。这将使这个值处于 “不恰当” 的状态。 + +为了证明这是可能发生的， **EvenChecker.test()** 创建了一组 **EventChecker** 对象，以连续读取 **EvenProducer** 的输出并测试检查每个数值是否都是偶数。如果不是，就会报告错误，而程序也将关闭。 + +多线程程序的部分问题是，即使存在 bug ，如果失败的可能性很低，程序仍然可以正确显示。 + +重要的是要注意到递增操作自身需要多个步骤，并且在递增过程中任务可能会被线程机制挂起 - 也就是说，在 Java 中，递增不是原子性的操作。因此，如果不保护任务，即使单一的递增也不是线程安全的。 + +该示例程序并不总是在第一次非偶数产生时终止。所有任务都不会立即关闭，这是并发程序的典型特征。 + ### 解决资源竞争 +前面的示例揭示了当你使用线程时的基本问题：你永远不知道线程何时运行。想象一下坐在一张桌子上，用叉子，将最后一块食物放在盘子上，当叉子到达时，食物突然消失...因为你的线程被挂起而另一个用餐者进来吃了食物了。这就是在编写并发程序时要处理的问题。为了使并发工作，您需要某种方式来阻止两个任务访问同一个资源，至少在关键时期是这样。 + +防止这种冲突的方法就是当资源被一个任务使用时，在其上加锁。第一个访问某项资源的任务必须锁定这项资源，使其他任务在其被解锁之前，就无法访问它了，而在其被解锁之时，另一个任务就可以锁定并使用它，以此类推。如果汽车前排座位是受限资源，那么大喊着 “冲呀” 的孩子就会（在这次旅途过程中）获得该资源的锁。 + +为了解决线程冲突的问题，基本的并发方案将序列化访问共享资源。这意味着一次只允许一个任务访问共享资源。这通常是通过在访问资源的代码片段周围加上一个子句来实现的，该子句一次只允许一个任务访问这段代码。因为这个子句产生 *互斥* 效果，所以这种机制的通常称为是 *mutex* （互斥量）。 + +考虑一下屋子里的浴室：多个人（即多个由线程驱动的任务）都希望能独立使用浴室（即共享资源）。为了使用浴室，一个人先敲门来看看是否可用。如果没人的话，他就能进入浴室并锁上门。任何其他想使用浴室的任务就会被 “阻挡”，因此这些任务就在门口等待，直到浴室是可用的。 + +当浴室使用完毕，就是时候给其他任务进入，这时比喻就有点不准确了。事实上没有人排队，我们也不知道下一个使用浴室是谁，因为线程调度机制并不是确定性的。相反，就好像在浴室前面有一组被阻止的任务一样，当锁定浴室的任务解锁并出现时，线程调度机制将会决定下一个要进入的任务。 + +Java 以提供关键字 **synchronized** 的形式，为防止资源冲突提供了内置支持。当任务希望执行被 **synchronized** 关键字保护的代码片段的时候，Java 编译器会生成代码以查看锁是否可用。如果可用，该任务获取锁，执行代码，然后释放锁。 + ### 同步多个生产者 -## volatile关键字 +## volatile 关键字 +### 字分裂 + +### 可见性 + +### 重排与 *Happen-Before*原则 + +### 什么时候使用 volatile ## 原子性 @@ -402,6 +597,11 @@ caught java.lang.RuntimeException ## 库组件 +### DelayQueue + +### PriorityBlockingQueue + +### Lock-Free Collections ## 本章小结 @@ -418,13 +618,13 @@ caught java.lang.RuntimeException 通常可以只使用 java.util.concurrent 库组件来编写并发程序，完全避免来自应用 volatile 和 synchronized 的挑战。注意，我可以通过 [并发编程](./24-Concurrent-Programming.md) 中的示例来做到这一点。 -[^1]: 在某些平台上，特别是 Windows，默认值可能非常难以查明。您可以使用 -Xss 标志调整堆栈大小。 +[^1]: 在某些平台上，特别是 Windows ，默认值可能非常难以查明。您可以使用 -Xss 标志调整堆栈大小。 [^2]: 出自 Brian Goetz, Java Concurrency in Practice 一书的作者 , 该书由 Brian Goetz, Tim Peierls, Joshua Bloch, Joseph Bowbeer, David Holmes, and Doug Lea 联合著作 (Addison-Wesley 出版社, 2006)。↩ [^3]: 请注意，在64位处理器上可能不会发生这种情况，从而消除了这个问题。 -[^4]: 这个测试的一个推论是，“如果有人暗示线程是直接的，请确保这个人没有对您的项目做出重要的决策。如果那个人已经做出，那么你就有麻烦了。” +[^4]: 这个测试的推论是，“如果某人表示线程是容易并且简单的，请确保这个人没有对你的项目做出重要的决策。如果那个人已经做出，那么你就已经陷入麻烦之中了。” [^5]: 这版本是我参与的；这可能在以后的标准中得到了修正