Java - 在时间基础上同步多个线程

Question

假设我有一个保存SlaveThread对象列表的主设备。在每个时间步中，我希望主设备并行运行SlaveThreads，但是，在时间步结束时，我希望SlaveThreads在向前推进之前等待对方完成当前时间步。另外，我不想在每个时间步骤重新实例化SlaveThreads。我有两种可能的解决方案，我不知道如何使它们中的任何一个都工作：

1）SlaveThread中的run（）方法在while（true）循环中。在SlaveThread一个循环的执行后，我会SlaveThread通知主（我不知道该怎么做），主做类似

try{ 
    for (int i = 0; i < numSlaveThreads; i++) { 
     while (!slaveThreads[i].getCompletedThisIter()) { 
     wait() 
     } 
     } 
    System.out.println("Joined"); 

}

前推进到下一个时间步骤。我将如何做到这一点？我怎么能有一个单独的SlaveThread通知主人？

2）Slave中的run（）不在while（true）循环中，那么我必须在每次迭代时调用start（）。但是此时Slave的线程状态将被终止。如何在不重新实例化的情况下再次调用start（）？

Answer 1

这正是障碍所在，你可以通过CyclicBarrier或CountDownLatch来实现。这些是同步器，用于延迟线程的进度，直到达到所需状态，在您的情况下线程已完成其计算。

这取决于细节要如何实现：

锁存器等待事件;障碍是等待其他 线程。

对于将在以下方式进行的CyclicBarrier：

// whereby count is the number of your slave threads 
this.barrier = new CyclicBarrier(count); 

然后在你的奴隶Runnable定义要插入在计算结束：barrier.await()

public class Slaves implements Runnable { 

    // ... 

    @Override 
    public void run() { 

     while(condition) { 

     // computation 
     // ... 

     try { 
      // do not proceed, until all [count] threads 
      // have reached this position 
      barrier.await(); 
     } catch (InterruptedException ex) { 
      return; 
     } catch (BrokenBarrierException ex) { 
      return; 
     } 
     } 
    } 
} 

直到所有线程完成计算后，您的从线程才会继续。这样你就不需要在另一个主线程之间实现信号传输。

但是，如果您在所有线程到达该位置（主信令）后都想要执行某些代码，则可以将其他Runnable传递给CyclicBarrier构造函数，该构造函数将在所有线程到达屏障后执行。

this.barrier = new CyclicBarrier(count, 
    new Runnable() { 
     @Override 
     public void run() { 
     // signal your master thread, update values, etc. 
     } 
    } 
); 

Answer 2

你可以使用一个ExecutorService的组合来管理你的线程（即您回收线程，而无需创建在每个周期新的）和一个CyclicBarrier将所有从机同步。

查看下面的一个简单的例子，其中主人启动一个循环的奴隶，确保他们都完成之前再次开始。奴隶，有点懒，只是睡一些（不是真正的随机）时间：

public class Test { 

    private static final ExecutorService executor = Executors.newFixedThreadPool(5); 
    private static final CyclicBarrier barrier = new CyclicBarrier(5); //4 slaves + 1 master 

    public static void main(String[] args) throws InterruptedException { 
     Runnable master = new Runnable() { 
      @Override 
      public void run() { 
       try { 
        while (true) { 
         System.out.println("Starting slaves"); 
         for (int i = 100; i < 500; i += 100) { 
          executor.submit(getRunnable(i)); 
         } 
         barrier.await(); 
         System.out.println("All slaves done"); 
        } 
       } catch (InterruptedException | BrokenBarrierException ex) { 
        System.out.println("Bye Bye"); 
       } 
      } 
     }; 

     executor.submit(master); 
     Thread.sleep(2000); 
     executor.shutdownNow(); 

    } 

    public static Runnable getRunnable(final int sleepTime) { 
     return new Runnable() { 
      @Override 
      public void run() { 
       try { 
        System.out.println("Entering thread " + Thread.currentThread() + " for " + sleepTime + " ms."); 
        Thread.sleep(sleepTime); 
        System.out.println("Exiting thread " + Thread.currentThread()); 
        barrier.await(); 
       } catch (BrokenBarrierException | InterruptedException ex) { 
       } 
      } 
     }; 

    } 
}