如果其他 2 个线程调用 signalAll，如何唤醒线程？

Question

我有 2 个工作线程和 1 个处理线程。

当处理线程试图处理某事并且工作线程正在执行它们的工作时，处理线程应该等待，并在工作线程中完成的所有工作完成后唤醒。

我怎样才能唤醒这个线程？我将尝试在此伪代码中演示我的意思

class Worker{
      /... variables ../
     private final Condition condition;
     public Worker(Condition condition){
        // constructor
     } 

     public void start(){
        Thread thread = new Thread(this::run);
        thread.start();
     }

     private void run(){
        try{
            ...
            condition.signalAll();    
        }
     }
}

和处理线程类似

      class Processing{
          private final Lock lock = new ReentrantLock();
          private final Condition condition = lock.newCondition();
          public void start(){
    
            Worker worker1 = new Worker(condition);
            Worker worket2 = new Worker(condition);
            worker1.start();
            worker2.start();
    
            // some code //
            while( // some flag indicating that workers are doing something){
                condition.await();
            }
            // this should be processed only after both worker1 and worker2 called signalAll
         }
   }

这样的事情可能吗？让线程等待多个源调用 notifyAll() 而不仅仅是一次。我希望我清楚地构建了这个问题。

感谢您的帮助！

Answer 1

这样的事情可能吗？让线程等到多个 名为 notifyAll() 的来源而且不止一次。

是的，例如，您可以使用CyclicBarrier：

允许一组线程全部等待的同步辅助 彼此到达共同的障碍点。 CyclicBarriers 很有用 在涉及固定大小的线程方的程序中，必须 偶尔互相等待。屏障被称为循环的，因为 它可以在等待线程释放后重新使用。一种 CyclicBarrier 支持运行一次的可选 Runnable 命令 每个障碍点，在派对中的最后一个线程到达之后，但是 在任何线程被释放之前。这种屏障作用对于 在任何一方继续之前更新共享状态。

CyclicBarrier 与 CountDownLatch 不同，可以多次重复使用，这一点毫无价值

一种同步辅助，允许一个或多个线程等待直到 在其他线程中执行的一组操作完成。一种 CountDownLatch 使用给定的计数进行初始化。等待方法 阻塞直到当前计数由于调用 countDown() 方法，之后释放所有等待的线程并 任何后续调用 await 都会立即返回。 这是一个 单发现象——计数无法重置。如果你需要一个 重置计数的版本，请考虑使用 CyclicBarrier。

class Worker{
      /... variables ../
     private final CyclicBarrier barrier;
     public Worker(CyclicBarrier condition){
        // constructor
     } 

     public void start(){
        Thread thread = new Thread(this::run);
        thread.start();
     }

     private void run(){
         try {
             ...
             barrier.await();
         } catch (InterruptedException | BrokenBarrierException e) {
            // Do Something
       }
     }
 }

和

      class Processing{
          final int total_threads = 3;
          CyclicBarrier barrier = new CyclicBarrier(total_threads);
          public void start(){
    
            Worker worker1 = new Worker(barrier);
            Worker worker2 = new Worker(barrier);
            worker1.start();
            worker2.start();
    
            // some code //
            try {
               barrier.await();
           } catch (InterruptedException | BrokenBarrierException e) {
              // Do Something
           }
   }

在本例中，master 线程等待worker1 和worker2 调用屏障，之后三个线程全部恢复工作。

Answer 2

您可以使用CountDownLatch 来实现该行为。

Answer 3

对于这种简单的情况，您不需要使用 Condition（这不是那么简单，顺便说一句 https://www.baeldung.com/java-concurrent-locks#working-with-conditions）。还有其他几个微不足道的选择：

1. 将您的工作线程定义为线程：

    class Worker extends Thread {
        public void run() {
            ...
        } 
    }

并在处理器中使用旧的好 .join() 来等待两个线程完成：

   class Processing {
       public void start() throws Exception {
           Worker worker1 = new Worker(condition);
           Worker worket2 = new Worker(condition);
           worker1.start();
           worker2.start();
           
           worker1.join();
           worker2.join();
           // this will be processed only after both worker1 and worker2 have their run() finished
        }
   }

1. 使用 CountDownLatch：

    class Worker {
        private final CountDownLatch latch;
    
        public Worker(CountDownLatch latch) {
            this.latch = latch;
        }
    
        private void run() {
            try {
             ...
            } finally {
               latch.countDown();
            }
        }
    }
    
    class Processing {
        public void start() throws Exception {
            CountDownLatch latch = new CountDownLatch(2);
            Worker worker1 = new Worker(latch);
            Worker worket2 = new Worker(latch);
            worker1.start();
            worker2.start();
            
            latch.await();
    
            // this will be processed only after both worker1 and worker2 called countDown()
        }
    }

当然，你可以使用 CyclicBarrier，但是为什么你需要这种可重复使用的东西呢？您甚至可以使用 Phaser https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Phaser.html，但这也是 imo，过度工程 :)

Answer 4

使用两个条件并让处理线程同时等待它们。