解决 Java 中由于 parallelStream 导致的死锁

并发性是软件开发的福音，也是祸根。通过并行处理提高性能的承诺与错综复杂的挑战相伴而生，例如臭名昭著的死锁。死锁是多线程编程世界中的隐患，它甚至可以使最强大的应用程序陷入瘫痪。它描述了两个或多个线程永远被阻塞，相互等待的情况。

在这篇博文中，我们深入探讨了现实世界中因使用 Java 的“ parallelStream ”而引发的死锁事件。我们将剖析根本原因，仔细检查线程堆栈跟踪。

场景

想象一下一个平静的代码库，它利用 Java 的“parallelStream”对 Collection 进行处理以提高处理速度。然而，随着我们的应用程序变得越来越复杂，出现了一个意想不到的隐蔽问题——死锁。线程曾经是盟友，现在却陷入了矛盾的境地。在这篇文章的后面，我们将仔细研究一些线程的堆栈跟踪，其中的主角是“ForkJoinPool.commonPool-worker-0”和“ForkJoinPool.commonPool-worker-1”。

以下两个线程堆栈跟踪：

线程 1：ForkJoinPool.commonPool-worker-0

stackTrace:

java.lang.Thread.State: BLOCKED (on object monitor)
at com.example.app.DataParser.read(DataParser.java:58)
- waiting to lock <0x00000001d6ff09f0> (a com.example.app.DataParser)
at com.example.app.ObjectLoader.read(ObjectLoader.java:196)
at com.example.app.MemorySnapshot$ObjectCacheManager.load(MemorySnapshot.java:2152)
at com.example.app.MemorySnapshot$ObjectCacheManager.load(MemorySnapshot.java:1)
at com.example.app.ObjectCache.get(ObjectCache.java:52)
- locked <0x00000001d6fafb00> (a com.example.app.MemorySnapshot$ObjectCacheManager)
at com.example.app.MemorySnapshot.getObject(MemorySnapshot.java:1453)
:
:
:
at com.example.app.AnalyzerImpl.lambda$37(AnalyzerImpl.java:3248)
at com.example.app.AnalyzerImpl$$Lambda$150/179364589.apply(Unknown Source)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:747)
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:721)
at java.util.stream.AbstractTask.compute(AbstractTask.java:327)
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Locked ownable synchronizers:
- None

线程2：ForkJoinPool.commonPool-worker-1

stackTrace:

java.lang.Thread.State: BLOCKED (on object monitor)
at com.example.app.ObjectCache.get(ObjectCache.java:44)
- waiting to lock <0x00000001d6fafb00> (a com.example.app.MemorySnapshot$ObjectCacheManager)
at com.example.app.MemorySnapshot.getObject(MemorySnapshot.java:1453)
at com.example.app.DataParser.readObjectArrayDump(DataParser.java:135)
at com.example.app.DataParser.read(DataParser.java:65)
- locked <0x00000001d6ff09f0> (a com.example.app.DataParser)
at com.example.app.ObjectLoader.read(ObjectLoader.java:196)
at com.example.app.ObjectInstance.read(ObjectInstance.java:135)
- locked <0x00000001e5822bf8> (a com.example.app.ObjectInstance)
at com.example.app.ObjectInstance.getAllFields(ObjectInstance.java:101)
:
:
:
at com.example.app.AnalyzerImpl.lambda$37(AnalyzerImpl.java:3248)
at com.example.app.AnalyzerImpl$$Lambda$150/179364589.apply(Unknown Source)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:747)
at java.util.stream.ReduceOps$ReduceTask.doLeaf(ReduceOps.java:721)
at java.util.stream.AbstractTask.compute(AbstractTask.java:327)
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Locked ownable synchronizers:
- None

线程堆栈跟踪让我们得以一窥死锁的核心。有趣的是，两个线程都纠缠在同一个方法中 — — 这清楚地表明它们正在争夺一个共享资源。有问题的资源是一个对象监视器，它通常是死锁场景中的罪魁祸首。

进一步分析表明，这些线程正在尝试执行需要独占访问“ com.example.app ”包内的对象缓存的操作。这种对资源访问的竞争是导致死锁的根本原因。这种资源争用构成了死锁的核心，每个线程都在等待对方放弃控制权。令人惊讶的是，这场灾难的罪魁祸首是人们所熟悉的“ parallelStream ”……

罪魁祸首：parallelStream

更深入地研究这种情况，我们可以在 Collection 上使用 Java 的parallelStream* 。对于不熟悉的人， “parallelStream”是一种旨在通过利用并行性来提高处理速度的机制。虽然“parallelStream”机制有望通过利用并行性来加速处理，但它可能会带来复杂情况，包括死锁。*

陷入死锁的线程都使用了默认的“fork-join 池”——一种促进并行操作的共享资源。我们的线程不知不觉地但不可避免地成为了该池中同一资源的竞争对手。这个共享池无意中造成了一种情况，即线程最终争夺资源，从而导致死锁。

更顺畅的道路：选择 Stream 而不是 ParallelStream

在解开死锁难题的过程中，我们发现了改变游戏规则的因素。在解开错综复杂的死锁网络时，我们意识到集合上使用 parallelStream 无意中加剧了资源争用。

我们在集合中使用 parallelStream 就像邀请多个朋友同时使用同一个玩具一样——这会导致很多推搡。这种骚动导致我们的线程发生冲突并导致死锁。但我们没有放弃；我们决定换一种玩法。

我们没有让所有人一起玩玩具，而是让他们轮流玩。我们从“parallelStream”切换到一种更简单的方式——只需“流式传输”。这一变化意味着一次只有一个朋友玩玩具。这种友好的轮流减少了打架的机会，并使我们的线程一起工作而不会发生冲突。

这种切换不仅仅是改变代码中的一个单词；它就像在游戏中选择一种新策略。猜猜怎么着？它成功了！线程现在一个接一个地运行，不再相互碰撞。这意味着不再有死锁，我们的应用程序可以松一口气了。

下面，您将看到突出显示这一关键转变的代码片段：

原始代码：

List<?> elements = …

elements.parallelStream().map(e -> {

   // Some code here that is causing deadlock…

}).collect(Collectors.toList());

修订后的代码：

List<?> elements = …

elements.stream().map(e -> {

   // Some code here that was causing deadlock…

}).collect(Collectors.toList());

带来重大改变

我们的死锁挑战表明，小的改变可以产生巨大的效果。从“parallelStream”切换到常规流虽然简单，但却带来了显著的变化。