代码之家 › 专栏 › 技术社区 › Hamy

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

deadlock monitoring locking concurrency java

Hamy · 技术社区 · 14 年前

我相信我面临的问题是嵌套监视器锁定的一个变体。基本上我有两组线程(不是线程组,只是逻辑组)。一组线程(比如说后台组)将等待一个对象,而另一组线程正在工作(工作组)。工作线程一个接一个地完成,直到最后一个工作线程处于“complete”方法中。我想做的是想办法告诉这个最后的工作线程等待,然后然后调用notifyAll()唤醒所有后台线程。正如您可能猜到的,这两组线程正在来回切换—一组正在工作,而另一组正在等待,然后组切换。问题是,如果在当前等待的线程上使用notifyAll(),则无法保证最终工作线程将在通知的线程完成并尝试启动下一次交换之前进入wait()调用。

很抱歉,如果这个问题有点离题-似乎我处理并发的次数越多,代码就越复杂:(

4 回复 | 直到 14 年前

Alex Miller 14 年前

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

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

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

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

Jed Wesley-Smith 14 年前

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

你的等待线程都调用 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中的类。

Rogach 14 年前

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

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

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

Jason 14 年前

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

螺纹加工:

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