问题标签 [phaser]

我现在遇到了两次问题，生产者线程产生 N 个工作项，将它们提交给一个ExecutorService，然后需要等到所有 N 个项目都被处理完。

注意事项

• N 事先不知道。如果是，我会简单地创建一个CountDownLatch然后拥有生产者线程await()，直到所有工作完成。
• 使用 aCompletionService是不合适的，因为尽管我的生产者线程需要阻塞（即通过调用take()），但无法发出所有工作已完成的信号，从而导致生产者线程停止等待。

我目前最喜欢的解决方案是使用整数计数器，并在提交工作项时增加它，并在处理工作项时减少它。在提交所有 N 个任务之后，我的生产者线程将需要等待一个锁，检查是否counter == 0收到通知。消费者线程将需要通知生产者，如果它已经减少了计数器并且新值是 0。

有没有更好的方法来解决这个问题，或者java.util.concurrent我应该使用合适的构造而不是“滚动我自己的”？

提前致谢。

这个问题是两年前提出的，但它提到的资源要么不是很有帮助（恕我直言），要么链接不再有效。

必须有一些很好的教程才能理解Phaser。我已经阅读了 javadoc，但是我的眼睛呆滞了，因为为了真正理解 javadoc，您必须知道应该如何使用这些类。

有人有什么建议吗？

鉴于：

我想知道我是否是最后一个要触发doSomething()的任务，这取决于任务的内部状态。我找到了 Phaser.onAdvance(int, int)但不清楚在这种情况下如何使用它。

所以，我有这个非常灵活的 Phaser，但似乎我错过了一些东西。
我已经成功使用了 CyclicBarrier，但现在我也想要更灵活的东西，就像我说的那样。所以这里是代码：

声明：

代码：

sp.getSomething();是 I/O 调用。
考虑到此代码示例由多个线程运行。

这是我的问题：我已确保不会超过 MAX_CLIENTS 到达移相器，所以如果 MAX_CLIENTS 到达，一切都很好。但是，我遇到了 TimeoutException 的问题。第一个是客户端（比如线程 A）将能够到达阶段的时间窗口（又名竞争条件），然后线程 B 中发生 TimeoutException，我正在线程 B 中动态实例化另一个带有到达方数量的移相器（比如 5），但随后线程 A 已经到达阶段（又名 phase1a 未发现 < 0）。我该如何纠正？我正在考虑使用信号量，但我认为这不值得努力，因为那样我可能需要重新考虑我这样做的方式。我还考虑过使用 Timer 并增加一个AtomicInteger变量和当定时器到期时动态实例化Phaser。关于如何解决这个问题的任何想法？

编辑：
文档有一种bulkRegister(int parties)方法，但措辞有点奇怪：

将给定数量的新未到达方添加到此移相器。如果正在进行的 onAdvance(int, int) 调用正在进行中，则此方法可能会在返回之前等待其完成。如果此移相器有父移相器，并且给定的参与方数量大于零，并且此移相器之前没有注册参与方，则此子移相器也向其父移相器注册。如果此移相器终止，则注册尝试无效，并返回负值。

问题： “可能”这个词让我感到困惑！“可能”如可能或“可能”

编辑：
解决。在下面检查我的答案。

我正在编写使用 Phaser 来知道何时完成工作的多线程应用程序。问题是在 ExecutorCompletionService 中甚至可以有 100k 的线程在一个队列中，但在 Phaser 中的最大到达方数是 65535。当有 65536 方到达时我该怎么办？

我的示例代码：

问题出在 addParties() 方法中。单线程（SimpleParser）可以返回即100个新文件名，将有100个新线程提交给ExecutorCompletionService，100个新方在Phaser中注册。我试图使用这样的东西：

并创建一个相位器链，但它没有帮助，因为 p.getUnarrivedParties() 返回 0，但我无法注册下一个参与方...

印刷：

65535

0

并抛出 IllegalStateException

那么如何创建与这个旧的Phaser 连接的新Phaser？

//编辑

谢谢@bowmore。我还有两个问题。

让我们看一下这个例子：

它打印：

为什么在 child1.arriveAndDeregister(); 之后 child1 没有终止，如何检查它是否真的终止？

第二个问题。我在达到 65535 方后询问是否要创建新的 Phaser，因为我认为创建数千个新对象是没有用的 - 你认为这样做不会有内存问题，或者它甚至可以提高性能吗？

据javadoc，在Phaser课堂上，

移相器可以分层（即，以树结构构建）以减少争用。可以改为设置具有大量参与方的移相器，否则这些移相器将经历沉重的同步争用成本，以便子移相器组共享一个共同的父级。这可能会大大增加吞吐量，即使它会产生更大的每次操作开销。

谁能澄清一下这个说法，这让我有点困惑。

我已经实现MyListener了从两个不同的线程访问。

我在主线程中这样使用它

我有几个担忧：

1. Java 文档指出调用arrive()和arriveAndAwaitAdvance()不register()先调用是不正确的。
2. changed()可能会为同一事件多次调用。我希望其中一个await()可能会因为以前的事件而立即返回。

关于这个问题的任何想法？

我试图产生一些线程，并在它们执行时将它们放在一个列表中。当他们完成处理时，我想收集他们的结果以进行演示。这样我就可以拥有一个包含许多线程的列表，然后一旦它们可用，我就可以调用future.get并使用它们的回调信息。

出于某种原因，我错过了许多结果。当我单步执行代码f.get()时，它不应该被忽略，我不知道为什么。

我的代码如下：

关于为什么会发生这种情况的任何想法？

编辑：

好吧，我想我已经启动并运行了：

我有一个具有请求映射的 spring @Controller。当向它发送请求时，它会创建一个延迟结果并在委托类上运行一个方法。

在控制器中：

委托器中的 doStuff 方法是 @Async 注释的 - 所以它在另一个线程中运行。在这个线程中，创建了一个新的 Phaser 来跟踪它将创建的子线程。委托线程本身向移相器注册，并触发对“AnotherClass”中的方法的调用，该方法也带有@Async 注释。即委托人现在正在产生孩子。

委托人：

每个子派生线程由委托中的 createThreadDoWork 调用创建，这里向移相器注册自己，做一些工作，然后到达并取消注册。

另一个类：

当所有子线程完成时，坐在这段代码中的委托人：

将继续，设置其延迟结果，委托线程将结束。

所以这里的问题（S）：

我希望每个请求都有一个 Phaser - 即，每次请求创建一个新的委托线程时，我都会创建一个新的 Phaser()。

如果我收到 5 个对我的控制器的请求，它们每个都会产生一个委托线程，每个委托人都会创建一个新的 Phaser 并将其传递给他们的孩子。然而，一旦其中一个线程完成了它的处理并且它的 Phaser 的特定实例被终止，其他委托线程继续并返回而没有完成它们的任务。

相位器可以这样使用吗？难道我做错了什么？我认为这将是跟踪子线程完成并等待它们完成的有用方法？

谢谢。

我刚刚阅读了关于Phaser 那里的javadoc，并且对该类的用法有疑问。javadoc 提供了一个示例，但现实生活中的示例呢？这样的屏障实现在实践中在哪里有用？