जावा - समय-चरण आधार पर एकाधिक धागे सिंक्रनाइज़ करना

Question

मान लीजिए मेरे पास एक मास्टर है जो स्लेव थ्रेड ऑब्जेक्ट्स की एक सूची रखता है। प्रत्येक बार कदम पर, मैं चाहता हूं कि मास्टर समानांतर में स्लेव थ्रेड चलाए, हालांकि, समय के अंत में, मैं चाहता हूं कि स्लेव थ्रेड्स आगे बढ़ने से पहले वर्तमान समय चरण को पूरा करने के लिए एक-दूसरे की प्रतीक्षा करें। साथ ही, मैं प्रत्येक बार चरण पर स्लेव थ्रेड को पुनर्स्थापित नहीं करना चाहता हूं। मेरे पास 2 संभावित समाधान हैं, और मुझे नहीं पता कि उनमें से कोई भी काम कैसे करें:

1) स्लेव थ्रेड में रन() विधि थोड़ी देर (सत्य) लूप में है। SlaveThread SlaveThread में एक भी पाश के निष्पादन के बाद, मैं होगा मास्टर (जो मैं कैसे करना है पता नहीं है) को सूचित, और मास्टर से पहले

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

कुछ}

करता है अगली बार कदम आगे बढ़ना। यह मैं कैसे करूंगा? मैं एक स्लेव थ्रेड कैसे मास्टर को सूचित कर सकता हूं?

2) स्लेव में रन() जबकि (सत्य) लूप में नहीं है, तो मुझे प्रत्येक पुनरावृत्ति पर प्रारंभ() पर कॉल करना होगा। लेकिन इस बिंदु पर दास की धागा स्थिति समाप्त हो जाएगी। मैं इसे फिर से शुरू किए बिना फिर से शुरू() कैसे कॉल कर सकता हूं?

Answer 1

यह वास्तव में बाधाओं के लिए है, आप इसे CyclicBarrier या CountDownLatch के साथ महसूस कर सकते हैं। ये पर सिंक्रनाइज़र्स हैं जो वांछित स्थिति तक पहुंचने तक की प्रगति में देरी करते हैं, आपके मामले में धागे ने उनकी गणना पूरी की है।

यहाँ यह विवरणों पर निर्भर करता है कि कैसे आप महसूस करना चाहते हैं:

सिटकनी की घटनाओं के लिए इंतजार कर के लिए कर रहे हैं; बाधाएं अन्य धागे की प्रतीक्षा के लिए हैं।

CyclicBarrier कि निम्नलिखित फैशन में किया जाएगा के लिए: 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; 
     } 
     } 
    } 
} 

:

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

फिर अपने दासों की Runnable परिभाषा आप गणना के अंत में दर्ज हो जाएगा आपके दास धागे आगे नहीं बढ़ेंगे, जब तक आपके सभी धागे गणना नहीं कर लेते। इस तरह आपको किसी अन्य मास्टर थ्रेड के बीच सिग्नलिंग का एहसास करने की आवश्यकता नहीं है।

अगर, हालांकि, आप कुछ कोड आप के बाद सभी धागे उस स्थिति (मास्टर संकेतन) पर पहुँच गए हैं निष्पादित करने के लिए आप CyclicBarrier निर्माता, जिसके बाद सभी धागे बाधा पर आ चुके हैं निष्पादित किया जाएगा के लिए एक अतिरिक्त Runnable पारित कर सकते हैं करना चाहते हैं ।

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

    } 
}