如何在一个对象上 wait() 然后在另一个对象上 notifyAll？

Question

我相信我面临的问题是嵌套监视器锁定的变体。基本上我有两组线程（不是线程组，只是逻辑组）。一组线程（假设是后台组）将在另一组线程（工作组）工作时等待对象。工作线程一个接一个地完成，直到最后一个工作线程处于“完成”方法中。我想要做的是找出一些方法告诉最后一个工作线程等待，然后调用 notifyAll() 来唤醒所有后台线程。正如您可能猜到的那样，两组线程正在来回切换 - 一组正在工作，另一组正在等待，然后组切换。问题是，如果我在当前等待的线程上 notifyAll() 则不能保证最终的工作线程会在通知的线程完成之前进入 wait() 调用并尝试开始下一次交换。

对不起，如果这个问题有点不对劲 - 似乎我在并发上工作的越多，我的代码就越复杂:(

Answer 1

听起来您需要类似 Gate 类的东西，它由两个 CountDownLatch 实例组成。我在很多多线程测试中使用了类似的东西。

您的等待线程全部调用gate.ready()，工作人员在完成后调用gate.go()

请注意，此特定实现假定有 1 个协调器线程。要支持更多，只需使用所需的等待者线程数构造 go 闩锁。

/**
 * Simple starting gate for co-ordinating a bunch of threads.
 */
final class Gate {
  final CountDownLatch ready;
  final CountDownLatch go = new CountDownLatch(1);

  Gate(final int threads) {
    ready = new CountDownLatch(threads);
  }

  /**
   * Called from the racing threads when ready. They will then block until all
   * threads are at this point;
   */
  void ready() {
    ready.countDown();
    await(go);
  }

  /**
   * Called from the starter thread. Blocks until everybody is ready, and then
   * signals go.
   */
  void go() {
    await(ready);
    go.countDown();
  }

  static void await(final CountDownLatch latch) {
    try {
      if (!latch.await(5, TimeUnit.SECONDS)) { // arbitrary, parameterise for production use
        throw new TimedOutException()
      }
    } catch (final InterruptedException e) {
      throw new RuntimeException(e);
    }
  }

  static final class TimedOutException extends IllegalStateException {}
}

如果您需要未知的任意线程数，您可能需要类似于 Doug Lea 在 Java7 中的 Phaser 类。

Answer 2

您可以尝试使用Exchanger 连接线程组。它通常用于在交替工作的两个线程之间来回传输工作。如果您可以使传输正常工作，似乎您也可以使其适用于线程组。

如果每个组都有一个控制器线程怎么办？然后，当控制器在 Exchanger 中收到项目时，您可以让控制器 notifyAll 在他的组中，然后加入他自己的所有组。当所有连接都返回时，他可以将控制权转移回 Exchanger。

或者如果每个组中的线程数是固定的，您可以为具有固定线程数的组创建一个CyclicBarrier，然后指定一个屏障操作，当所有线程完成并达到屏障时运行.该操作可以通过 Exchanger 或 SynchronousQueue（这是一个强制同步协调的 0 长度队列）转移控制。

有关同步器的更多信息，请查看Java Concurrency in Practice 或DZone concurrency refcard。

Answer 3

也许您可以使用一些变量来指示仍在工作的线程数。所以，当线程完成时，它使用这个方法：

synchronized void completed() {
    threads_working--;
    if (threads_working == 0) {
        synchronized (some_lock) {
            some_lock.notifyAll();
        }
    }
}

每个线程在开始工作时都会增加该数字。

Answer 4

是否可以让线程返回并终止而不是让它们等待？如果是这样，您是否考虑过实现一个线程管理器来生成线程并启动对每个组的控制？

线程进程：

public void run()
{
    while (workRemaining())
    {
        doWork();
    }
    this.manager.workCompleted();
}

在线程管理器中：

void workCompleted()
{
    if (--this.runningThreads <= 0)
    {
        spawnNewGroup();
    }
}

void spawnNewGroup()
{
    for (int i=0; i<groupSize; i++)
    {
        startIndividualThread();
        this.runningThreads++;
    }
}