问题标签 [forkjoinpool]

我正在尝试在浮点值的输入上完成我的一维中值滤波器顺序代码的并行版本的编码。我一直在使用一些较小长度的测试值来对并行版本进行编码，最终似乎使代码正常工作，但是现在在 +-36000 个元素的大型数据集上，代码似乎只是增加了 CPU 资源而不是完成，而我的顺序版本在相同的给定输入上完成。有谁知道我哪里出错了？

测试有效的输入：2、6、80、3、1 并正确产生：2、6、6、3、1

我编写的另一个串行代码运行良好，但是这个并行版本没有给出任何输出，它只是打印你好，我正在用“嘿”测试它，它甚至从未到达那一行，即它卡在调用方法上，我不能在网上找不到任何有用的资源。

请帮助我了解我哪里出错了我是并行编程的新手。

该代码是一个中值滤波程序，当给定一个数组 x=[2,80,6,3] 时，过滤后的数组是 y=[2,6,6,3]，计算如下：

y[1] = 中位数[2 2 80] = 2

y[2] = 中值[2 80 6] = 中值[2 6 80] = 6

y[3] = 中值[80 6 3] = 中值[3 6 80] = 6

y[4] = 中值[6 3 3] = 中值[3 3 6] = 3

我对并行性和并发性很陌生，我正在尝试使用 Java 中的 Fork-Join 实现中值过滤算法。基本上，我将输入文件读入 ArrayList 并使用该列表生成过滤中位数的新 ArrayList（包括原始 ArrayList 的第一个和最后一个元素）。

现在我设法制作了算法的串行/顺序版本，它工作正常。但是，当我尝试制作 Fork-Join 版本时，它似乎不适用于大型 ArrayLists（100000+）。我用一个大小为 5 的非常小的 ArrayList 进行了尝试，它工作正常。我似乎无法找到我的错误（我确定这是一个逻辑错误和/或实现错误）。任何帮助，将不胜感激。

这是顺序算法片段：

这是我制作的并行课程。这是不起作用的部分：

非常感谢任何帮助。在花了几个小时并围绕 SO 和 Google 进行了大量研究后，我无法做到这一点。

编辑：musical_coder 建议发布我面临的错误，它们就在这里。这是很多错误：

通常，当使用Java 8 的parallelStream() 时，结果是通过默认的通用fork-join 池（即ForkJoinPool.commonPool()）执行。

然而，这显然是不可取的，如果一个人的工作远离 CPU 限制，例如，可能大部分时间都在等待 IO。在这种情况下，人们会希望使用一个单独的池，根据其他标准（例如任务可能实际使用 CPU 的时间）来确定大小。

没有明显的方法可以让 parallelStream() 使用不同的池，但是这里有一种详细的方法。

不幸的是，这种方法需要从 fork-join 池线程调用并行流上的终端操作。这样做的缺点是，如果目标分叉连接池完全忙于现有工作，则整个执行将等待它，而完全什么也不做。因此，池可能成为比单线程执行更严重的瓶颈。相比之下，当以“正常”方式使用 parallelStream() 时，ForkJoinPool.common.externalHelpComplete() 或 ForkJoinPool.common.tryExternalUnpush() 用于让来自池外的调用线程帮助处理。

有谁知道一种方法可以让parallelStream() 使用非默认的fork-join 池，并让来自fork-join 池外部的调用线程帮助处理这项工作（但不是fork-的其余部分-加入游泳池的工作）？

从java文档，

ForkJoinPool 与其他类型的 ExecutorService 的不同之处主要在于采用了工作窃取：池中的所有线程都尝试查找并执行由其他活动任务创建的子任务（如果不存在，则最终阻塞等待工作）。

当大多数任务产生其他子任务（大多数 ForkJoinTasks 也是如此）时，这可以实现高效处理。在构造函数中将 asyncMode 设置为 true 时，ForkJoinPools 也可能适用于从未加入的事件样式任务。

经过下面的ForkJoinPool 示例后，与 ThreadPoolExecutor 不同，我没有看到设置队列大小的参数。我不知道 ForkJoinPool 如何窃取机制。

假设我已经创建了 10 个线程的 ThreadPoolExecutor，并且已经提交了 3000 个 Callable 任务。这些线程如何分担子任务的执行负载？

对于相同的用例，ForkJoin 池的行为如何不同？

这是一个例子：

问题 1：
如何确保 RecursiveTaskA 在调用doSomethingElse()之前完成？
目前我的解决方案是：

这是正确的方法吗？或者，还有更好的方法？
问题2：
如果我使用上面的解决方案，那么pool.submit(new RecursiveTaskB()); 结果错误。我不想创建一个新的 ForkJoinPool，怎么办？

任何链接，参考都会对我有所帮助，谢谢。

伙计们

我有以下代码可以在我的爬虫上添加访问过的链接。提取链接后，我有一个for 循环，循环遍历每个单独的href 标签。

在我访问了一个链接并打开它之后，我会将 URL 添加到上面定义的访问过的链接集合变量中。

爬虫实现是多线程的，假设我访问了 100,000 个 url，如果我没有终止爬虫，它会一天比一天增长。它会产生内存问题吗？请问，我有什么选项可以刷新变量而不造成线程间的不一致？

提前致谢！

我正在开发一个使用 forkjoinpool 进行多线程和 log4j 进行日志记录的项目。在我们的项目中，有多个“作业”并发运行，每个作业仅由一个线程处理。

日志记录的问题在于工作的主要流程中，我们可以将作业 ID 前缀到记录器，以便相应的日志条目将具有作业 ID 信息。但是，在每个作业的处理过程中，都会调用许多实用程序 API，并且来自此类 API 的日志条目没有作业信息。log4j 确实打印线程 ID，因此这是区分不同作业日志的一种方法，但我想知道将作业 ID 添加到实用程序 API 日志的最简单/最佳方法是什么？

我大概有两个想法，我不确定我是否走对了方向，

1. 如果有办法让我用我的工作 ID 替换线程名称，那么我认为我对 log4j 的默认线程前缀很好。
2. 或者我应该使用实用程序方法将记录器作为输入，以便记录器可以打印作业信息？

请指教，谢谢。

我试图弄清楚如何在 Java 8 并行流中复制 ThreadLocal 值。

所以如果我们考虑这个：

哪个输出

有没有办法让我将 ThreadLocal 从 main() 方法克隆到为每个并行执行产生的线程中？

这样我的结果是：

而不是第一个？

在 Java 8 中有两种启动异步计算的方法 -CompletableFuture和ForkJoinTask. 它们看起来都非常相似 - CompletableFutureeven extend的内部类ForkJoinTask。

有理由使用其中一个吗？

我可以看到的一个关键区别是，该CompletableFuture.join方法只是阻塞，直到未来完成（waitingGet只是使用 a 旋转ManagedBlocker），而 aForkJoinTask.join可以从队列中窃取工作以帮助您完成加入的任务。

是否比其中一个有好处？