问题标签 [gpars]

正在尝试使用 Gpars 执行并发操作。

最初它似乎工作

但是在迭代一段时间后，一些线程停止执行。

直到最终我们只剩下一个。

知道为什么会这样吗？有什么解决方案可以保持所有线程处于活动状态？

就我而言，值得一提的是，我们的迭代速度约为 50k-200k。

我想使用并行处理但在特定超时内检索多个“代价高昂”的结果。

我正在使用GPars Dataflow.task 但看起来我遗漏了一些东西，因为只有当所有数据流变量都被绑定时，进程才会返回。

我确实在GPars Timeouts 文档中看到了一种使用 Select 在超时内获得最快结果的方法。但我正在寻找一种在给定时间范围内检索尽可能多的结果的方法。

有没有更好的“GPars”方法来实现这一目标？还是使用 Java 8 Future/Callable ？

我想让我的主演员在第一个演员完成工作后立即终止所有奴隶演员。
但是，我不知道如何从我的主人向所有奴隶发送广播。有这方面的功能或编程模式吗？

解决这个问题的另一种方法是给主节点一个所有从节点的列表并循环遍历它们，向每个节点发送一条消息以终止，但是从节点也需要主节点作为属性，我认为这是一个问题：

此代码无法编译。
错误信息：

Actor 线程 Actor 线程 2 java.lang.NullPointerException 发生异常：无法在 org.codehaus.groovy 的 org.codehaus.groovy.runtime.NullObject.invokeMethod(NullObject.java:91) 的空对象上调用方法 send() .runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:48) 在 org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:48) 在 org.codehaus.groovy.runtime.callsite.NullCallSite.call (NullCallSite.java:35) 在 org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:48) 在 org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:113) 在org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:125) 在 Slave.act(ConsoleScript14:11) 在 groovyx.gpars.actor.DefaultActor。handleStart(DefaultActor.java:342) 在 groovyx.gpars.actor.AbstractLoopingActor$1.handleMessage(AbstractLoopingActor.java:70) 在 groovyx.gpars.util.AsyncMessagingCore.run(AsyncMessagingCore.java:132) 在 java.util.concurrent。 ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745)爪哇：745）爪哇：745）

看起来我创建奴隶后无法设置 sl.master 。

有解决方法吗？

任何帮助是极大的赞赏。
先感谢您！

我有一种方法，我尝试使用 GPARS 并行化计算并计算调用中的聚合布尔“和”结果。此方法被包装为 @ActiveObject ，它将结果作为数据流传递 - 下面的代码具有我尝试使用 AtomicBoolean 存储聚合以保护它的原始方法。

这在计算出的“最终真相”上不起作用（有时我的测试会通过其他他们会失败的测试）。为了解决这个问题，我从 AtomicBoolean 更改为 Agent(boolean) 方法，我认为它已经“修复”了它——至少我的 spock 测试一直在成功。

尝试使用 AtomicBoolean 构建最终结果时，我的逻辑在哪里存在缺陷？感觉它应该工作 - 但没有，我不明白为什么。

下面的方法 - 我把原始版本和更正的版本放在下面

通过使用像这样的代理表单修复了问题

我把 debug println 放在这里，这样我就可以看到代码在做什么。

具有保护布尔聚合值的代理的版本似乎正在运行。

为什么原子布尔值不起作用？

我编写了一个小方法来执行git命令行工具并捕获其输出：

我将它与 GPars 一起使用以同时克隆多个存储库，例如

但是，我相信我的git方法不是线程安全的：例如，在第一个线程到达方法的第五行command之前，第二个线程可以在方法的第一行进行修改。command.execute()

我可以通过制作整个git方法来解决这个问题synchronized，但这会破坏在不同线程中运行它的目的，因为我希望克隆并行发生。

所以我想像做部分同步

但我想这也不安全，因为线程二waitForProcessOutput()可能比线程一更早返回，从而搞砸了outputStream/errorStream变量。

获得此线程安全的正确方法是什么？

我一直在尝试做一个小型 Groovy 项目，并且想要一个ConcurrentLinkedHashSet，但 Java 没有提供。因此，我开始使用 Gpars 代理创建自己的代理来“保护”普通的 LinkedHashSet。然后我创建了我的包装类来保存代理，并将我的类上的方法委托给内部代理，就像这样（这个版本使用 methodMissing 作为委托方法）。我尝试了 Groovy interceptable/invokeMethod 并且可以让它工作

但是，当我尝试使用它时 - 它第一次工作，我的字符串被添加，但在第二次调用相同的缺失方法时，agent.send 调用永远不会进行，并且会被跳过。

所以我的简单脚本使用者看起来像这样

我的简单跟踪日志在控制台输出上看起来像这样

正如您在第一次尝试委派时看到的那样，您可以看到“concHashSet methodMissing:”跟踪，并且代理在代理中对其调用 invokeMethod 以影响添加元素。

在第二次调用 conhs.add ('col2') 时，agent.sand 调用永远不会发生，因此我的额外项目永远不会被添加。

这很烦人，因为我认为我有一个简单的方法来创建我的 ConcurrentLinkedHashSet，但是代码不起作用。我可以使用什么机制来获得正确的结果？

正如你所看到的，当我直接调用方法（添加）时Agent<LinkedHashSet>它工作得很好。在我真正的消费类中，如果我用普通的 LinkedHashSet 替换我的 ConcurrentLinkedHashSet 它可以实现梦想，但不是线程安全的。我想创建一个线程安全版本，这取决于试图让它工作。

我想我可以尝试替换代理并仅在我的 LinkedHashSet 周围使用同步块 - 但它有点难看 - 我认为 Gpars 代理会为我将所有这些作为通用解决方案模式作为具有委托的包装器进行排序。

PS我尝试了另一种方法，我认为这种工作方式，但感觉不对 - 它在实现 GroovyInterceptable 的类上使用 @Synchronise on invokeMethod 以在委派时实现线程安全调用。我不确定这是否真的线程安全。

需要一些帮助。查看 Gpars 数据流/管道，但有些我不明白

如果您查看下面的示例（我已经使用运算符、管道线、chainWith 完成了此操作并遇到了同样的问题）。

在此示例中，我使用了任务，但也很容易没有任务，并且出现相同的问题。在此示例中，我设置了两个 DataflowQueue，一个用于初始条件，一个用于根据谓词评估的结果。然后我布局一个管道，根据谓词（偶数测试）评估输入并将结果存储在输出结果队列中

设置了管道并将一些条目发布到第一个队列中，我相信这些条目将在数据可用时被处理（这也不适用于操作员版本），如您所见，我测试了 resultQ 的大小为零（如果我在我将条目写入 sessionQ 之后，删除仍然正确的任务）。因此，写入数据不会“触发”处理。

第一个任务将一些条目保存到队列中。

它仅在第二个任务/chainWith 等中 - 您在引擎开始运行的第二个队列上调用 .val （或 get() ），所有条目都从第一个队列处理，结果绑定到 resultQ。

您可以从断言中看到这一点，因为一旦进行了第一个触发器 (.val) 同步调用，引擎就会运行并处理起始 sessionQ 中的所有绑定条目。

这是一个问题，因为直到您运行第一个 .val 调用 - 如果您执行 poll() 或 resultQ.interator.size() 例如它是空的且未绑定，size()=0。所以你不能写

因为它总是空的，直到您使用 sessionQ 中的第一个项目。我不明白为什么？在条目绑定到第一个 dataflowQueue 后，我认为这些项目将在它们可用时被消耗（已绑定） - 但它们不是。

现在这很棘手，因为您无法通过，检查结果的大小，在 resultQ 上执行 poll()，因为它会失败，直到读取 sessionQ 的第一个 DF。

我最终不得不使用初始值数组的大小（告诉我保存到队列中的条目），因为唯一意味着从 resultQ 中读取相同的数字以清空它（在上面我只消耗了 3 resultsQ 中的记录和断言表明 resultQ 中还剩下 2 条记录（但只有在第一次 .val 调用之后，如果您注释掉所有断言开始失败）

我用 Dataflow.operator、Pipeline 等尝试了这个并得到了同样的问题。为什么每个输入都绑定到 SessionQ 时，工作没有得到处理？

最后，在 Pipeline 的情况下，有一个 .complete() 方法，如果你在管道中处理闭包 {}，它会保持打开状态 (!complete())，但是当你运行像 .binaryChoice() 这样的方法时，它会标记管道已完成，无法添加进一步的操作。为什么这样做？

当然，我不明白那个状态在说什么（不再进行处理），如果您尝试在这种方法之后再执行另一个步骤，则会引发异常。

无论哪种方式 - 我尝试过这样的管道线

但是，当您将值绑定到 Q 时，什么都不会发生 - 直到您使用类似的输出

当管道突然“运行”并处理存储在 Q 中的所有 DF 项目时。

除了第一个 .val 消耗之外，我没有尝试过启动工作的安排

任何人都可以解释为什么会这样，我必须在这里忽略这一点，但是在读取第一个条目之前“什么都不做”不是我所期望的，并且会使任何大小评估（.iterator.size（），poll（）等无效) 类型调用 DataflowWriteChannel 目标。

我很感激对此的任何帮助-我已经为此苦苦挣扎了两天，但一无所获。我也查看了所有 Gpars 测试，它们只是调用 .val 的次数与输入绑定的次数相同 - 所以不要显示我描述的问题。

Vaclav Pech 或任何其他观看这些问题的 Gpars 大师，我将不胜感激任何有关这方面的帮助见解，以帮助我度过这个难关

提前问候

我正在尝试在后端调用中实现异步，我一直在阅读，似乎 GParse 是一个很好的库来实现这一点，但我不清楚如何以正确的方式实现这一点。谁能帮我？这是我的例子：

所以，我需要调用 2 个 api 调用，然后在第三个调用中，我需要使用之前称为 async 的响应之一，最后将其添加到我的模型中。我怎样才能做到这一点？

在 i7-2960xm（4 核超线程）上试验多线程时，我遇到的 Groovy Gpars 的性能比预期的要差。在我的测试中，我一直在使用递归 fib 计算器来模拟工作负载：

为了测试 Gpars，我目前正在使用以下代码：

并行计算fib(36)大约需要 1.9 秒withPool(4)。withPool(1)大约需要 1.4 秒，我认为这有点类似于在 Gpars 之外调用函数，但只需要 0.4 秒，例如：

有人可以解释为什么我可能会遇到这种性能下降吗？谢谢！

这是我的 SplitTimer 以防万一：

我一直在关注 Grails 提供的关于 Promise 的文档： https ://docs.grails.org/2.3.x/guide/async.html#promises

我有一个巨大的 id 列表，我将它们分成不同的列表tasks，然后调用waitAll()promise 列表。基于此，当其中一个并发任务抛出异常时，是否可以回滚所有更改？

查看代码可能更容易：

当一切正常时，这很有效，但当然，当事情进展不顺利时，我必须做好准备。理想情况下，即使最后一个任务抛出异常，我也想回滚所有任务中所做的所有更改。

这可能吗？