问题标签 [fork-join]

我开始学习 ExecutorService 类。文档（和在线教程）说总是调用 ExecutorService.shutDown() 来回收资源。但是，文档还说，在您调用 shutdown() 后，将不会接受任何新任务。所以，我的问题是，每当我需要并行化数据处理时，我是否总是必须实例化一个新的 ExecutorService ？

现在我有一个可调用对象列表，我执行以下操作。

但是，我的 OuterCallable 也使用 InnerCallable 拆分数据或并行执行数据处理。

我不记得它是用于 ExecutorService 还是 Fork/Join 方法，但我记得文档和教程说操作数据的实际并行过程不应该涉及 I/O 操作，并且一切都应该在内存中完成。但是，在我的 InnerCallable 中，我实际上是在进行 JDBC 调用（此处未显示）。

最终，我使用 ExecutorService 的方式有效，但我仍然有挥之不去的担忧。

1. 我的方法是否高于使用 ExecutorService 的良好编程实践？
2. 我应该使用 ExecutorService 的单例实例吗？
3. 我不仅应该避免在我的并行方法中进行 I/O 操作，而且还应该避免 JDBC 调用吗？

作为最后一个问题，我试图对 Fork/Join 与 ExecutorService 进行一些研究。我遇到了一篇彻底抨击 Fork/Join API/类的文章。学习 Fork/Join 是否值得？我看到了一些关于 stackoverflow 和其他地方的文章，其中测试用于比较 Fork/Join 与 ExecutorService，并且有图表显示 Fork/Join 与 ExecutorService 更好的 CPU 使用率（通过 Windows 任务管理器）。但是，当我使用 ExecutorService (JDK 1.7.x) 时，我的 CPU 使用率最高。ExecutorService 是否使用最新的 JDK 进行了改进？

任何帮助/指导表示赞赏。

我想从持有写锁的线程向 ForkJoinPool 或 ParallelArray 提交任务。通过检查当前线程是否持有相关锁来保护对我们域模型的访问。为了允许 FJ 工作人员在其上执行任务（只读，例如查询），他们需要将访问检查委托给产生它们的线程。

我使用对生成线程的引用对 ForkJoinWorkerThread 进行了子类化。然后，我将 ReentrantReadWriteLock 子类化并覆盖 isWriteLockedByCurrentThread 以执行通常的检查，并回退到检查，如果线程是委托 FJWorker 的实例，则委托线程（父线程）是锁的所有者，使用ReentrantReadWriteLock#getOwner () :

但是，getOwner() 的文档说明了以下内容：

当不是所有者的线程调用此方法时，返回值反映了当前锁定状态的最大努力近似值。例如，即使有线程试图获取锁但尚未这样做，所有者也可能暂时为空。

我想理解这意味着如果我在已被授予访问权限的线程中提交任务，此方法将正确返回对它的引用。不幸的是，这甚至没有暗示。

如果我不能使用这种方法，这种委托还有哪些其他选择？

谢谢你。

我正在比较一个测试程序的两个变体。两者都ForkJoinPool在具有四个内核的机器上使用 4 线程运行。

在“模式 1”中，我非常像使用执行器服务一样使用池。我把一堆任务扔进ExecutorService.invokeAll. 与普通的固定线程执行器服务相比，我获得了更好的性能（即使有对 Lucene 的调用，在那里执行一些 I/O）。

这里没有分而治之。从字面上看，我愿意

在“模式 2”中，我将单个任务提交到池中，并在该任务中调用 ForkJoinTask.invokeAll 来提交子任务。所以，我有一个继承自 的对象，RecursiveAction它被提交到池中。在该类的计算方法中，我调用了来自另一个invokeAll类的对象集合，该类也继承自. 出于测试目的，我一次只提交一个第一个对象。我天真地期望看到所有四个线程都忙的事情，因为线程调用会为自己抓取其中一个子任务，而不是仅仅坐着阻塞。我能想到一些为什么它可能无法以这种方式工作的原因。RecursiveActioninvokeAll

在 VisualVM 中观察，在模式 2 中，一个线程几乎总是在等待。我期望看到的是调用invokeAll 的线程会立即处理其中一个被调用的任务，而不是静止不动。这肯定比使用普通线程池尝试这种方案所导致的死锁要好，但是，怎么办？如果提交其他内容，它是否会阻止一个线程？如果是这样，为什么模式 1 中没有同样的问题？

到目前为止，我一直在使用添加到 java 1.6 的引导类路径中的 jsr166 jar 来运行它。

这是两者的javadoc：

1. get()：如有必要，等待计算完成，然后检索其结果。
2. invoke()：开始执行此任务，如有必要，等待其完成，并返回其结果，或者如果底层计算这样做，则抛出（未经检查的）RuntimeException 或 Error。

在 Doug Lea 的论文“A Java Fork/Join Framework”中：

http://gee.cs.oswego.edu/dl/papers/fj.pdf

在 2.1 Work-Stealing 中，他说：

当工作线程遇到连接操作时，它会处理其他任务（如果可用），直到发现目标任务已完成（通过 isDone）。否则，所有任务都会在没有阻塞的情况下运行完成。

那么谁能具​​体告诉我这些“其他任务”来自哪里？它们来自其他工作线程的任务队列吗？这是否意味着每当工作线程遇到连接调用时，它会继续“从其他线程窃取任务”而不是“跳到自己队列中的其他任务”？

据我了解ForkJoinPool，该池创建固定数量的线程（默认值：核心数）并且永远不会创建更多线程（除非应用程序通过 using 指示需要这些线程managedBlock）。

然而，使用ForkJoinPool.getPoolSize()我发现在一个创建 30,000 个任务 ( RecursiveAction) 的程序中，ForkJoinPool执行这些任务平均使用 700 个线程（每次创建任务时都会计算线程数）。这些任务不做 I/O，而是纯粹的计算；唯一的任务间同步是调用ForkJoinTask.join()和访问AtomicBooleans，即没有线程阻塞操作。

由于join()按照我的理解不会阻塞调用线程，因此池中的任何线程都没有理由阻塞，因此（我假设）没有理由创建任何进一步的线程（这显然仍在发生） .

那么，为什么要ForkJoinPool创建这么多线程呢？哪些因素决定了创建的线程数？

我曾希望可以在不发布代码的情况下回答这个问题，但这里是应要求提供的。这段代码是从一个四倍大小的程序中摘录出来的，精简到了基本部分；它不会按原样编译。如果需要，我当然也可以发布完整的程序。

该程序使用深度优先搜索在迷宫中搜索从给定起点到给定终点的路径。保证存在解决方案。主要逻辑在: A的compute()方法中，它从某个给定点开始，并继续从当前点可到达的所有相邻点。而不是创造一个新的SolverTaskRecursiveActionSolverTask在每个分支点（这将创建太多任务），它将除一个之外的所有邻居推送到回溯堆栈以供稍后处理，并且仅继续仅一个未推送到堆栈的邻居。一旦它以这种方式到达死胡同，最近推送到回溯堆栈的点就会被弹出，并且从那里继续搜索（相应地削减从 taks 的起点构建的路径）。一旦任务发现其回溯堆栈大于某个阈值，就会创建一个新任务；从那时起，该任务在继续从其回溯堆栈中弹出直到用完为止，但在到达分支点时不会将任何其他点推入其堆栈，而是为每个这样的点创建一个新任务。因此，可以使用堆栈限制阈值来调整任务的大小。

我上面引用的数字（“30,000 个任务，平均 700 个线程”）来自于搜索 5000x5000 个单元的迷宫。所以，这里是基本代码：

这是否意味着它将迁移到 Scala 2.10 或 jsr166y 将单独发布？...或者是其他东西？

我读到了 Java 7 中引入的 Fork/Join 框架的实现，我只是想检查一下我是否理解这个魔法是如何工作的。

据我了解，当一个线程分叉时，它会在其队列中创建子任务（其他线程可能会或可能不会窃取）。当线程尝试“加入”时，它实际上会检查其队列中的现有任务，然后递归地执行它们，这意味着对于任何“加入”操作 - 2 帧将被添加到线程调用堆栈中（一个用于加入，一个用于对于新的任务调用）。

据我所知，JVM 不支持尾调用优化（在这种情况下可能用于删除 join 方法堆栈帧），我相信在执行具有大量分叉和连接的复杂操作时，线程可能会抛出StackOverflowError.

我是对的还是他们找到了一些很酷的方法来防止它？

编辑

这是一个有助于澄清问题的场景：假设（为简单起见）我们在 forkjoin 池中只有一个线程。在某个时间点 - 线程分叉，然后调用 join。在 join 方法中，线程发现它可以执行分叉的任务（正如它在其队列中发现的那样），因此它调用下一个任务。该任务依次分叉，然后调用 join - 因此在执行 join 方法时，线程将在其队列中找到分叉的任务（如前所述）并调用它。在那个阶段，调用堆栈将至少包含两个连接和两个任务的帧。

如您所见，fork join 框架已转换为普通递归。因为 java 不支持尾调用优化 - java 中的每个递归都可能导致StackOverflowError如果它足够深。

我的问题是——fork/join 框架的实现者是否找到了一些很酷的方法来防止这种情况。

我目前正在评估各种并发解决方案来解决业务问题。该用例类似于“令人尴尬的并行”算法。

基本上对于单个用户请求，我们需要在计算响应之前从多个不同的数据源中检索数据。目前所有 3 个 DAO 调用都是串行进行的，但没有相互依赖关系，因此可以并行进行。

目前实施的解决方案：

• java.util.concurrent.ExecutorService 使用 Callables 和 Futures
• org.springframework.scheduling.annotation.Async 启用 spring 管理线程池，但仍然允许我进行异步调用
• Akka（被认为是矫枉过正）用于我们相对简单的用例

我想评估的最后一个框架是 Java ForkJoin 框架，我可以看到多个使用 RecursiveTasks 的示例，但我的用例本质上不是递归的，因此不适合模型：如果任务足够小，则拆分它并递归调用相同的方法（即分而治之）

我的用例是将任务拆分为 3 个任务。分叉所有 3 并再次加入。这甚至是 ForkJoin 实现的有效用例吗？或者我应该坚持使用通用的 ExecutorService 实现。

例如，在 Java 平台上的 Fork / Join 框架中可以使用工作窃取。（请参阅Fork /join framework 如何比线程池更好？） - OmniThreadLibrary 是否有类似的可能？

工作窃取：无事可做的工作线程可以从其他仍在忙碌的线程中窃取任务。