如何在一个对象上等待(),然后在另一个对象上通知所有对象?

您可以尝试使用 Exchanger . 它通常用于在两个交替工作的线程之间来回传输工作。如果你能让传输正常工作的话,似乎你也可以让它为线程组工作。

如果每个组都有一个控制器线程呢?然后,当控制器在交换器中收到项目时,可以让他通知组中的所有人,然后加入他自己的组中的所有人。当连接全部返回时,他可以将控制权转移回交换机。

或者如果每个组中的线程数是固定的,则可以创建 CyclicBarrier 对于具有固定线程数的组,然后指定在所有线程完成并达到屏障时要运行的屏障操作。这个动作可以通过交换器或 SynchronousQueue (这是一个0长度的队列,强制执行同步协调)。

有关同步器的详细信息,请查看 Java Concurrency in Practice 或者 DZone concurrency refcard .

听起来你需要一个由两个倒计时闩锁实例组成的门类。我在很多多线程测试中使用类似的方法。

你的等待线程都调用 gate.ready() 工人们打电话来 gate.go() 完成后

注意这个特殊的实现假设有一个协调线程。为了支持更多,只需构造 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的 Phaser Java7中的类。

也许您可以使用一些变量来指示仍在工作的线程数。因此,当线程完成时,它使用以下方法:

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

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

是否可以允许线程返回并终止,而不是让它们等待?如果是,您是否考虑过实现一个线程管理器来生成线程并初始化对每个组的控制?

螺纹加工:

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++;
    }
}