Java ExecutorService: awaitTermination aller rekursiv erstellte Aufgaben

Dies ist wirklich ein Idealer Kandidat für einen Phaser-Effekt. Java 7 kommt mit dieser neuen Klasse. Seine flexible CountdonwLatch/CyclicBarrier. Sie erhalten eine stabile version auf JSR 166-Interest-Site.

So, wie es ist ein flexibler CountdownLatch/CyclicBarrier ist, weil es in der Lage, nicht nur die Unterstützung einer unbekannten Anzahl von Parteien (threads), aber es ist auch wiederverwendbar (das ist, wo die phase Teil kommt in)

Für jede Aufgabe, die Sie Einreichen, Sie würden registrieren, wenn die Aufgabe abgeschlossen ist, wenn Sie ankommen. Diese kann getan werden, rekursiv.

Phaser phaser = new Phaser();
ExecutorService e = //

Runnable recursiveRunnable = new Runnable(){
   public void run(){
      //do work recursively if you have to

      if(shouldBeRecursive){
           phaser.register();
           e.submit(recursiveRunnable);
      }

      phaser.arrive();
   }
}

public void doWork(){
   int phase = phaser.getPhase();

   phaser.register();
   e.submit(recursiveRunnable);

   phaser.awaitAdvance(phase);
}

Edit: Danke @depthofreality für den Hinweis auf die race-condition in meinem vorherigen Beispiel. Ich bin eine Aktualisierung es so, dass die Ausführung thread wartet nur noch auf vorher von der aktuellen phase, wie Sie Blöcke für die rekursive Funktion zu vervollständigen.

Phase Anzahl nicht Reise, bis die Anzahl der arrives == registers. Da vor jedem rekursiven Aufruf, ruft register eine phase Inkrement wird geschehen, wenn alle Aufrufe abgeschlossen sind.

InformationsquelleAutor John Vint

Könnte man erstellen Sie Ihre eigenen thread-pool erstreckt sich ThreadPoolExecutor. Sie möchten wissen, wenn eine Aufgabe abgegeben wurde und wenn es vollendet.

public class MyThreadPoolExecutor extends ThreadPoolExecutor {
    private int counter = 0;

    public MyThreadPoolExecutor() {
        super(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
    }

    @Override
    public synchronized void execute(Runnable command) {
        counter++;
        super.execute(command);
    }

    @Override
    protected synchronized void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        counter--;
        notifyAll();
    }

    public synchronized void waitForExecuted() throws InterruptedException {
        while (counter == 0)
            wait();
    }
}

InformationsquelleAutor Peter Lawrey

(mea culpa: es ist ein 'bisschen' hinter meinem Schlafenszeit 😉 aber hier ist ein Erster Versuch, ein dynamisches latch):

package oss.alphazero.sto4958330;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class DynamicCountDownLatch {
    @SuppressWarnings("serial")
    private static final class Sync extends AbstractQueuedSynchronizer {
        private final CountDownLatch toplatch;
        public Sync() {
            setState(0);
            this.toplatch = new CountDownLatch(1);
        }

        @Override
        protected int tryAcquireShared(int acquires){
            try {
                toplatch.await();
            } 
            catch (InterruptedException e) {
                throw new RuntimeException("Interrupted", e);
            }
            return getState() == 0 ? 1 : -1;
        }
        public boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc)) 
                    return nextc == 0;
            }
        }
        public boolean tryExtendState(int acquires) {
            for (;;) {
                int s = getState();
                int exts = s+1;
                if (compareAndSetState(s, exts)) {
                    toplatch.countDown();
                    return exts > 0;
                }
            }
        }
    }
    private final Sync sync;
    public DynamicCountDownLatch(){
        this.sync = new Sync();
    }
    public void await() 
        throws InterruptedException   
    {
        sync.acquireSharedInterruptibly(1);
    }

    public boolean await(long timeout, TimeUnit   unit) 
        throws InterruptedException   
    {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
    public void countDown() {
        sync.releaseShared(1);
    }
    public void join() {
        sync.tryExtendState(1);
    }
}

Dieser Riegel stellt eine neue Methode join (), um die vorhandenen (geklont) CountDownLatch-API, die verwendet wird, indem Sie Aufgaben, um zu signalisieren Ihrem Eintritt in die größere Aufgabe der Gruppe.

Den Riegel pass um von übergeordneten Aufgaben an untergeordnete Aufgabe. Jede Aufgabe, pro Suraj Muster, ersten 'join()' den Riegel, der seine Aufgabe(), und dann countDown().

Zur Behandlung von Situationen, wo der main-thread startet den task-Gruppe und dann sofort erwartet() -- vor der Aufgabe, die Fäden haben eine chance hatte, sich selbst () -- topLatch verwendet wird int inneren Sync Klasse. Dies ist ein latch, das wird nach unten gezählt, auf jedem join(); nur der erste countdown ist natürlich groß, da alle nachfolgenden NOPs für.

Die erste Implementierung oben führt zu einem semantischen Falten des möglichen, da die tryAcquiredShared(int) ist nicht zugelassen werden, wirft eine InterruptedException, aber dann müssen wir befassen sich mit der interrupt-auf das warten auf die topLatch.

Ist dies eine Verbesserung gegenüber OP ' s eigene Lösung mit Atomarer Zähler? Ich würde sagen, wahrscheinlich nicht, wennn er ist beharrlich auf der Vollzieher, aber es ist, glaube ich, eine gleichermaßen gültige alternative Ansatz durch die AQS in diesem Fall, und ist verwendbar mit generischen threads sowie.

Crit Weg fellow Hacker.

InformationsquelleAutor alphazero

Muss ich sagen, dass Lösungen, die oben beschrieben problem mit den rekursiven aufrufen von task und warten Sie, bis Ende Unterordnung Aufgaben nicht erfüllen mich. Da ist meine Lösung, inspiriert durch die original-Dokumentation von Oracle gibt es: CountDownLatch und Beispiel: Human resources CountDownLatch.

Den ersten roten Faden im Prozess im Instanz der Klasse HRManagerCompact hat warten Riegel für die zwei Tochter-threads, die darauf warten Riegel für den späteren 2 Tochter threads... usw.

Natürlich", "latch" eingestellt werden können, auf der anderen Wert als 2 (im Konstruktor von CountDownLatch), sowie die Anzahl der runnable-Objekte hergestellt werden können, die in iteration also ArrayList, aber es muss übereinstimmen (Nummer des count-downs gleich sein müssen, die parameter in CountDownLatch Konstruktor).

Vorsichtig sein, die Anzahl der Verriegelungen erhöht sich exponentiell nach Einschränkung Bedingung:
'Ebene.get() < 2', sowie die Anzahl der Objekte. 1, 2, 4, 8, 16... und latches 0, 1, 2, 4... Wie Sie sehen können, für die vier Ebenen (Ebene.get() < 4) es werden 15 wartenden threads und 7 Riegel in der Zeit, als peak-16 threads ausgeführt werden.

package processes.countdownlatch.hr;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/** Recursively latching running classes to wait for the peak threads
 *
 * @author hariprasad
 */
public class HRManagerCompact extends Thread {
  final int N = 2; //number of daughter's tasks for latch
  CountDownLatch countDownLatch;
  CountDownLatch originCountDownLatch;
  AtomicInteger level = new AtomicInteger(0);
  AtomicLong order = new AtomicLong(0); //id latched thread waiting for

  HRManagerCompact techLead1 = null;
  HRManagerCompact techLead2 = null;
  HRManagerCompact techLead3 = null;

//constructor
public HRManagerCompact(CountDownLatch countDownLatch, String name,
    AtomicInteger level, AtomicLong order){
  super(name);
  this.originCountDownLatch=countDownLatch;
  this.level = level;
  this.order = order;
 }

 private void doIt() {
    countDownLatch = new CountDownLatch(N);
    AtomicInteger leveli = new AtomicInteger(level.get() + 1);
    AtomicLong orderi = new AtomicLong(Thread.currentThread().getId());
    techLead1 = new HRManagerCompact(countDownLatch, "first", leveli, orderi);
    techLead2 = new HRManagerCompact(countDownLatch, "second", leveli, orderi);
    //techLead3 = new HRManagerCompact(countDownLatch, "third", leveli);

    techLead1.start();
    techLead2.start();
    //techLead3.start();

    try {
     synchronized (Thread.currentThread()) { //to prevent print and latch in the same thread
       System.out.println("*** HR Manager waiting for recruitment to complete... " + level + ", " + order + ", " + orderi);
       countDownLatch.await(); //wait actual thread
     }
     System.out.println("*** Distribute Offer Letter, it means finished. " + level + ", " + order + ", " + orderi);
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
  }

 @Override
 public void run() {
  try {
   System.out.println(Thread.currentThread().getName() + ": working... " + level + ", " + order + ", " + Thread.currentThread().getId());
   Thread.sleep(10*level.intValue());
   if (level.get() < 2) doIt();
   Thread.yield();
  }
  catch (Exception e) {
   //TODO Auto-generated catch block
   e.printStackTrace();
  }
  /*catch (InterruptedException e) {
   //TODO Auto-generated catch block
   e.printStackTrace();
  }*/
  //TODO Auto-generated method stub
  System.out.println("--- " +Thread.currentThread().getName() + ": recruted " + level + ", " + order + ", " + Thread.currentThread().getId());
  originCountDownLatch.countDown(); //count down
 }

 public static void main(String args[]){
  AtomicInteger levelzero = new AtomicInteger(0);
  HRManagerCompact hr = new HRManagerCompact(null, "zero", levelzero, new AtomicLong(levelzero.longValue()));
  hr.doIt();
 }
}

Möglich kommentierte Ausgabe (mit einiger Wahrscheinlichkeit):

first: working... 1, 1, 10 //thread 1, first daughter's task (10)
second: working... 1, 1, 11 //thread 1, second daughter's task (11)
first: working... 2, 10, 12 //thread 10, first daughter's task (12)
first: working... 2, 11, 14 //thread 11, first daughter's task (14)
second: working... 2, 11, 15 //thread 11, second daughter's task (15)
second: working... 2, 10, 13 //thread 10, second daughter's task (13)
--- first: recruted 2, 10, 12 //finished 12
--- first: recruted 2, 11, 14 //finished 14
--- second: recruted 2, 10, 13  //finished 13 (now can be opened latch 10)
--- second: recruted 2, 11, 15  //finished 15 (now can be opened latch 11)
*** HR Manager waiting for recruitment to complete... 0, 0, 1
*** HR Manager waiting for recruitment to complete... 1, 1, 10
*** Distribute Offer Letter, it means finished. 1, 1, 10 //latch on 10 opened
--- first: recruted 1, 1, 10 //finished 10
*** HR Manager waiting for recruitment to complete... 1, 1, 11
*** Distribute Offer Letter, it means finished. 1, 1, 11 //latch on 11 opened
--- second: recruted 1, 1, 11  //finished 11 (now can be opened latch 1)
*** Distribute Offer Letter, it means finished. 0, 0, 1  //latch on 1 opened

InformationsquelleAutor

Nur unelegant Lösung könnte ich kommen mit, um direkt mit einem ThreadPoolExecutor und Abfrage seiner getPoolSize() jeder einmal in eine Weile. Gibt es wirklich keine bessere Art und Weise tun Sie das?

Verwenden Sie shutdown() ,awaitTermination()and shutdownNow() Methoden in der richtigen Reihenfolge.

shutdown(): Initiiert einen geordneten shutdown, in dem zuvor eingereichten Aufgaben werden ausgeführt, aber es werden keine neuen Aufgaben angenommen werden.

awaitTermination():Blockiert, bis alle Aufgaben abgeschlossen haben, der Ausführung nach eine shutdown-Anforderung, oder das timeout Auftritt, oder der aktuelle thread unterbrochen wird, je nachdem, was zuerst Eintritt.

shutdownNow(): Versuche zu stoppen, alle aktiv ausgeführten Aufgaben, Stoppt die Verarbeitung der wartenden Aufgaben, und gibt eine Liste der Aufgaben, die auf Ihre Hinrichtung warteten.

Empfohlene Methode aus der oracle-Dokumentation Seite ExecutorService:

 void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); //Disable new tasks from being submitted
   try {
     //Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); //Cancel currently executing tasks
       //Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     //(Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     //Preserve interrupt status
     Thread.currentThread().interrupt();
   }

Kann man ersetzen, wenn die Bedingung bei while-Bedingung im Fall von langer Dauer in der Erledigung von Aufgaben wie folgt:

Ändern

if (!pool.awaitTermination(60, TimeUnit.SECONDS))

Zu

 while(!pool.awaitTermination(60, TimeUnit.SECONDS)) {
     Thread.sleep(60000);
 }  

Finden Sie andere alternativen (außer join(), die verwendet werden kann, mit der eigenständigen thread ), in :

warten Sie, bis alle threads Ihre Arbeit beenden im java

InformationsquelleAutor Ravindra babu