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) { 
       } 
      } 
     }; 

    } 
}