Wie kann ich mit ExecutorService warten, bis alle Threads abgeschlossen sind?

381

Ich muss eine bestimmte Anzahl von Aufgaben 4 gleichzeitig ausführen, ungefähr so:

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
    taskExecutor.execute(new MyTask());
}
//...wait for completion somehow

Wie kann ich benachrichtigt werden, wenn alle abgeschlossen sind? Im Moment kann ich mir nichts Besseres vorstellen, als einen globalen Aufgabenzähler zu setzen und ihn am Ende jeder Aufgabe zu verringern. Dann diesen Zähler in einer Endlosschleife zu überwachen, um 0 zu werden. oder erhalten Sie eine Liste von Futures und im Endlosschleifenmonitor isDone für alle. Was sind bessere Lösungen ohne Endlosschleifen?

Vielen Dank.

serg
quelle

Antworten:

446

Grundsätzlich auf einen ExecutorServiceAnruf shutdown()und dann awaitTermination():

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}
taskExecutor.shutdown();
try {
  taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
  ...
}
Cletus
quelle
9
Genau dafür ist shutdown / awaitTermination gedacht
matt b
31
Es ist ein gutes Muster, wenn diese Aufgabenbehandlung ein einmaliges Ereignis ist. Wenn dies wiederholt zur gleichen Laufzeit durchgeführt wird, ist dies jedoch nicht optimal, da Sie Threads bei jeder Ausführung wiederholt erstellen und abreißen würden.
Sjlee
44
Ich suche nach offiziellen Unterlagen, Long.MAX_VALUE, TimeUnit.NANOSECONDSdie einer Zeitüberschreitung entsprechen.
Sam Harwell
15
Ich kann nicht glauben, dass Sie shutdown verwenden müssen, um alle aktuellen Threads zu verbinden (nach der Verwendung von shutdown können Sie den Executor nie wieder verwenden). Schlagen Sie vor, stattdessen eine Liste von Future's zu verwenden ...
Rogerdpack
20
@SamHarwell die siehe java.util.concurrentPaket - Dokumentation unter dem TimingAbschnitt: „ für immer“ warten, können Sie einen Wert verwenden könnenLong.MAX_VALUE
beluchin
174

Verwenden Sie einen CountDownLatch :

CountDownLatch latch = new CountDownLatch(totalNumberOfTasks);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
while(...) {
  taskExecutor.execute(new MyTask());
}

try {
  latch.await();
} catch (InterruptedException E) {
   // handle
}

und innerhalb Ihrer Aufgabe (in try / finally einschließen)

latch.countDown();
ChssPly76
quelle
3
Es gibt keine 4 Aufgaben. Es werden "einige Aufgaben" 4 gleichzeitig ausgeführt.
Cletus
2
Entschuldigung, ich habe die Frage falsch verstanden. Ja, die Anzahl der Aufgaben sollte das Argument für den CountDownLatch-Konstruktor sein
ChssPly76
3
Ich finde diese Lösung eleganter als die anderen, sie scheint für diesen Zweck entwickelt worden zu sein und ist einfach und unkompliziert.
wvdschel
3
Was ist, wenn Sie die Anzahl der Aufgaben vor dem Start nicht kennen?
Cletus
11
@cletus - dann verwendest du keinen CountDownLatch :-) Wohlgemerkt, ich behaupte nicht, dass dieser Ansatz besser ist als deiner. Allerdings fand ich, dass in realen Situationen ich tun die Anzahl der Aufgaben kennen, Thread - Pool - Einstellungen Sie pro Einsatz konfigurierbar sein müssen, und Pools können wiederverwendet werden. Daher habe ich normalerweise Thread-Pools, die von Spring injiziert wurden, und das Festlegen von Prototypen als Prototypen und das manuelle Herunterfahren nur , um auf das Ende der Threads zu warten, scheint nicht ideal.
ChssPly76
82

ExecutorService.invokeAll() tut es für dich.

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
List<Callable<?>> tasks; // your tasks
// invokeAll() returns when all tasks are complete
List<Future<?>> futures = taskExecutor.invokeAll(tasks);
sjlee
quelle
Die Schwierigkeit tritt auf, wenn Sie die "4" Threads
einzeln
@rogerdpack: Ich lerne immer noch diese Executoren und so. Aber als Antwort auf Ihre Fragen. Sollten die 4 Threads gleichzeitig nicht Teil einer Batch-Aufgabe sein, die mit der obigen Antwort ausgeführt wird?
Mukul Goel
3
Diese Methode funktioniert nur, wenn Sie die Anzahl der Aufgaben vorher kennen.
Konstantin
3
Ich denke, wenn die futureszurückgegeben werden, sind die Aufgaben noch nicht erledigt. Sie werden möglicherweise in Zukunft abgeschlossen und Sie haben einen Link zum Ergebnis. Deshalb heißt es a Future. Sie haben die Methode Future.get () , die darauf wartet, dass die Aufgabe abgeschlossen ist, um ein Ergebnis zu erhalten.
AlikElzin-Kilaka
5
@ AlikElzin-kilaka Zitat aus den JavaDocs (in der Antwort verlinkt): "Führt die angegebenen Aufgaben aus und gibt eine Liste von Futures zurück, die ihren Status und ihre Ergebnisse enthalten, wenn alle abgeschlossen sind. Future.isDone () gilt für jedes Element der zurückgegebenen Liste. ""
Hulk
47

Sie können auch Futures-Listen verwenden:

List<Future> futures = new ArrayList<Future>();
// now add to it:
futures.add(executorInstance.submit(new Callable<Void>() {
  public Void call() throws IOException {
     // do something
    return null;
  }
}));

Wenn Sie dann allen beitreten möchten, entspricht dies im Wesentlichen dem Beitritt zu jedem (mit dem zusätzlichen Vorteil, dass Ausnahmen von untergeordneten Threads zum Haupt-Thread erneut ausgelöst werden):

for(Future f: this.futures) { f.get(); }

Grundsätzlich besteht der Trick darin, .get () für jede Zukunft einzeln aufzurufen, anstatt in einer Endlosschleife isDone () für (alle oder jede) aufzurufen. Sobald der letzte Thread beendet ist, können Sie diesen Block garantiert durchlaufen und passieren. Die Einschränkung besteht darin, dass Sie, da der Aufruf von .get () erneut Ausnahmen auslöst, wenn einer der Threads stirbt, dies möglicherweise auslösen würden, bevor die anderen Threads vollständig abgeschlossen sind [um dies zu vermeiden, können Sie ein https: // hinzufügen stackoverflow.com/a/31885029/32453catch ExecutionException Aufruf um den get-Aufruf ]. Die andere Einschränkung ist, dass ein Verweis auf alle Threads beibehalten wird. Wenn sie also lokale Thread-Variablen haben, werden sie erst erfasst, nachdem Sie diesen Block überschritten haben (obwohl Sie dies möglicherweise umgehen können, wenn es zu einem Problem wurde, indem Sie es entfernen Zukunft ist aus der ArrayList). Wenn Sie wissen möchten, welche Zukunft "zuerst endet"

Rogerdpack
quelle
3
Die „Oberflächen erste“ Zu wissen, Verwendung ExecutorCompletionService.take: stackoverflow.com/a/11872604/199364
ToolmakerSteve
34

In Java8 können Sie dies mit CompletableFuture tun :

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();
AdamSkywalker
quelle
3
Dies ist eine sehr elegante Lösung.
Mattvonb
ExecutorService es = Executors.newFixedThreadPool(4); List< Future<?>> futures = new ArrayList<>(); for(Runnable task : taskList) { futures.add(es.submit(task)); } for(Future<?> future : futures) { try { future.get(); }catch(Exception e){ // do logging and nothing else } }
user2862544
@AdamSkywalker wartet aufTermination () nach es.shutdown ()?
Gaurav
@gaurav Wenn Sie shutdown aufrufen, sind einige Aufgaben möglicherweise noch nicht abgeschlossen. Mit awaitTermination wird der aufrufende Thread blockiert, bis alles erledigt ist. Es hängt davon ab, ob Sie auf Ergebnisse in diesem Thread warten müssen oder nicht.
AdamSkywalker
@AdamSkywalker tolle Antwort. Es ist sinnvoll, awaitTermination () nicht aufzurufen, wenn ich nicht auf die Ergebnisse warten muss.
Gaurav
26

Nur meine zwei Cent. Um die Anforderung zu überwinden CountDownLatch, die Anzahl der Aufgaben im Voraus zu kennen, können Sie dies auf die altmodische Weise mithilfe einer einfachen Aufgabe tun Semaphore.

ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
int numberOfTasks=0;
Semaphore s=new Semaphore(0);
while(...) {
    taskExecutor.execute(new MyTask());
    numberOfTasks++;
}

try {
    s.aquire(numberOfTasks);
...

Rufen s.release()Sie bei Ihrer Aufgabe einfach so an, wie Sie es möchtenlatch.countDown();

Stryba
quelle
Als ich das sah, fragte ich mich zuerst, ob es ein Problem wäre, wenn einige releaseAnrufe vor dem acquireAnruf stattfinden, aber nachdem ich die Semaphore-Dokumentation gelesen habe, sehe ich, dass das in Ordnung ist.
ToolmakerSteve
13

Ein bisschen spät zum Spiel, aber der Vollendung halber ...

Anstatt darauf zu warten, dass alle Aufgaben erledigt sind, können Sie nach dem Hollywood-Prinzip denken: "Rufen Sie mich nicht an, ich rufe Sie an" - wenn ich fertig bin. Ich denke, der resultierende Code ist eleganter ...

Guave bietet einige interessante Werkzeuge, um dies zu erreichen.

Ein Beispiel ::

Wickeln Sie einen ExecutorService in einen ListeningExecutorService ::

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));

Senden Sie eine Sammlung von Callables zur Ausführung ::

for (Callable<Integer> callable : callables) {
  ListenableFuture<Integer> lf = service.submit(callable);
  // listenableFutures is a collection
  listenableFutures.add(lf)
});

Nun zum wesentlichen Teil:

ListenableFuture<List<Integer>> lf = Futures.successfulAsList(listenableFutures);

Fügen Sie der ListenableFuture einen Rückruf hinzu, mit dem Sie benachrichtigt werden können, wenn alle Futures abgeschlossen sind:

        Futures.addCallback(lf, new FutureCallback<List<Integer>>() {
        @Override
        public void onSuccess(List<Integer> result) {
            log.info("@@ finished processing {} elements", Iterables.size(result));
            // do something with all the results
        }

        @Override
        public void onFailure(Throwable t) {
            log.info("@@ failed because of :: {}", t);
        }
    });

Dies bietet auch den Vorteil, dass Sie alle Ergebnisse nach Abschluss der Verarbeitung an einem Ort sammeln können ...

Weitere Informationen hier

Răzvan Petruescu
quelle
2
Sehr sauber. Funktioniert auch unter Android einwandfrei. Musste einfach nutzen runOnUiThread()in onSuccess().
DSchmidt
12

Die CyclicBarrier- Klasse in Java 5 und höher ist für diese Art von Dingen konzipiert.

Pekka Enberg
quelle
7
Cool, kann mich nie an den Namen dieser Datenstruktur erinnern. Nur geeignet, wenn Sie im Voraus wissen, wie viele Aufgaben in die Warteschlange gestellt werden.
2 ᆺ ᆼ
Ja, Sie würden denken, Sie könnten die Barriere mit dem aktuellen Thread und allen
untergeordneten
Eigentlich ist es eine falsche Antwort. CyclicBarrier für Portionen. CountDownLatch für wartende Veranstaltung
gstackoverflow
7

Folgen Sie einem der folgenden Ansätze.

  1. Durchlaufen Sie alle zukünftigen Aufgaben, kehren Sie von Ein submitzurück ExecutorServiceund überprüfen Sie den Status mit blockierendem Aufruf get()des FutureObjekts, wie von vorgeschlagenKiran
  2. Verwenden Sie invokeAll()auf ExecutorService
  3. CountDownLatch
  4. ForkJoinPool oder Executors.html # newWorkStealingPool
  5. Verwenden Sie die shutdown, awaitTermination, shutdownNowAPIs von ThreadPoolExecutor in der richtigen Reihenfolge

Verwandte SE-Fragen:

Wie wird CountDownLatch in Java Multithreading verwendet?

So beenden Sie Java ExecutorService ordnungsgemäß

Ravindra Babu
quelle
6

Hier sind zwei Optionen, nur ein bisschen verwirren, welche am besten ist.

Option 1:

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();

Option 2:

ExecutorService es = Executors.newFixedThreadPool(4);
List< Future<?>> futures = new ArrayList<>();
for(Runnable task : taskList) {
    futures.add(es.submit(task));
}

for(Future<?> future : futures) {
    try {
        future.get();
    }catch(Exception e){
        // do logging and nothing else
    }
}
es.shutdown();

Hier setzen future.get (); in try catch ist eine gute idee oder?

user2862544
quelle
5

Sie können Ihre Aufgaben in eine andere ausführbare Datei einbinden, die Benachrichtigungen sendet:

taskExecutor.execute(new Runnable() {
  public void run() {
    taskStartedNotification();
    new MyTask().run();
    taskFinishedNotification();
  }
});
Zed
quelle
1
Ich habe eine Weile gebraucht, um zu sehen, wie dies die Frage von OP lösen würde. Beachten Sie zunächst, dass diese Umhüllung für jede Aufgabe gilt und nicht für den Code, mit dem alle Aufgaben gestartet werden. Vermutlich würde jeder Start einen Zähler erhöhen, und jedes Ende würde diesen Zähler verringern oder einen completedZähler erhöhen . Nachdem Sie alle gestartet haben, können Sie bei jeder Benachrichtigung feststellen, ob alle Aufgaben abgeschlossen sind. Beachten Sie, dass die Verwendung unbedingt erforderlich ist, try/finallydamit eine fertige Benachrichtigung (oder eine alternative Benachrichtigung im catchBlock) auch dann angezeigt wird, wenn eine Aufgabe fehlschlägt. Sonst würde ich ewig warten.
ToolmakerSteve
3

Ich habe gerade ein Beispielprogramm geschrieben, das Ihr Problem löst. Es wurde keine präzise Implementierung angegeben, daher werde ich eine hinzufügen. Obwohl Sie executor.shutdown()und verwenden können executor.awaitTermination(), ist dies nicht die beste Vorgehensweise, da die von verschiedenen Threads benötigte Zeit nicht vorhersehbar wäre.

ExecutorService es = Executors.newCachedThreadPool();
    List<Callable<Integer>> tasks = new ArrayList<>();

    for (int j = 1; j <= 10; j++) {
        tasks.add(new Callable<Integer>() {

            @Override
            public Integer call() throws Exception {
                int sum = 0;
                System.out.println("Starting Thread "
                        + Thread.currentThread().getId());

                for (int i = 0; i < 1000000; i++) {
                    sum += i;
                }

                System.out.println("Stopping Thread "
                        + Thread.currentThread().getId());
                return sum;
            }

        });
    }

    try {
        List<Future<Integer>> futures = es.invokeAll(tasks);
        int flag = 0;

        for (Future<Integer> f : futures) {
            Integer res = f.get();
            System.out.println("Sum: " + res);
            if (!f.isDone()) 
                flag = 1;
        }

        if (flag == 0)
            System.out.println("SUCCESS");
        else
            System.out.println("FAILED");

    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
Kiran
quelle
Es ist gut, dass Sie die Verwendung von future.get zeigen - eine gute Alternative, über die Sie Bescheid wissen sollten. Aber warum halten Sie es für besser, ewig zu warten , als ein maximal akzeptables Zeitlimit festzulegen? Noch wichtiger ist, dass es keinen Grund gibt, all diese Logik zu verwenden, wenn man einfach eine sehr, sehr lange Zeit zum Warten auf die Beendigung geben kann, wenn man (im Wesentlichen für immer) warten möchte, bis alle Aufgaben abgeschlossen sind.
ToolmakerSteve
Dies unterscheidet sich nicht von den hier bereits vorgestellten Lösungen. Ihre gerechte Lösung ist die gleiche wie von @sjlee
pulp_fiction
Nicht sicher, warum Sie überprüfen müssen, ob erledigt, wenn laut oracle doc invokeAll nur zurückgibt, "wenn alles abgeschlossen ist oder das Timeout abläuft, je nachdem, was zuerst
eintritt
3

Nur um hier mehr Alternativen zur Verwendung von Riegeln / Barrieren anzubieten. Sie können die Teilergebnisse auch abrufen, bis alle mit CompletionService abgeschlossen sind .

Von Java Concurrency in der Praxis: "Wenn Sie eine Reihe von Berechnungen an einen Executor senden müssen und deren Ergebnisse abrufen möchten, sobald sie verfügbar sind, können Sie die mit jeder Aufgabe verknüpfte Zukunft beibehalten und wiederholt nach Abschluss abrufen, indem Sie get with a aufrufen Timeout von Null. Dies ist möglich, aber langwierig . Glücklicherweise gibt es einen besseren Weg : einen Abschlussservice. "

Hier die Umsetzung

public class TaskSubmiter {
    private final ExecutorService executor;
    TaskSubmiter(ExecutorService executor) { this.executor = executor; }
    void doSomethingLarge(AnySourceClass source) {
        final List<InterestedResult> info = doPartialAsyncProcess(source);
        CompletionService<PartialResult> completionService = new ExecutorCompletionService<PartialResult>(executor);
        for (final InterestedResult interestedResultItem : info)
            completionService.submit(new Callable<PartialResult>() {
                public PartialResult call() {
                    return InterestedResult.doAnOperationToGetPartialResult();
                }
        });

    try {
        for (int t = 0, n = info.size(); t < n; t++) {
            Future<PartialResult> f = completionService.take();
            PartialResult PartialResult = f.get();
            processThisSegment(PartialResult);
            }
        } 
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } 
        catch (ExecutionException e) {
            throw somethinghrowable(e.getCause());
        }
    }
}
Alberto Gurrion
quelle
3

Dies ist meine Lösung, basierend auf dem Tipp "AdamSkywalker", und sie funktioniert

package frss.main;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestHilos {

    void procesar() {
        ExecutorService es = Executors.newFixedThreadPool(4);
        List<Runnable> tasks = getTasks();
        CompletableFuture<?>[] futures = tasks.stream().map(task -> CompletableFuture.runAsync(task, es)).toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(futures).join();
        es.shutdown();

        System.out.println("FIN DEL PROCESO DE HILOS");
    }

    private List<Runnable> getTasks() {
        List<Runnable> tasks = new ArrayList<Runnable>();

        Hilo01 task1 = new Hilo01();
        tasks.add(task1);

        Hilo02 task2 = new Hilo02();
        tasks.add(task2);
        return tasks;
    }

    private class Hilo01 extends Thread {

        @Override
        public void run() {
            System.out.println("HILO 1");
        }

    }

    private class Hilo02 extends Thread {

        @Override
        public void run() {
            try {
                sleep(2000);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("HILO 2");
        }

    }


    public static void main(String[] args) {
        TestHilos test = new TestHilos();
        test.procesar();
    }
}
frss-soft.com
quelle
2

Sie könnten diesen Code verwenden:

public class MyTask implements Runnable {

    private CountDownLatch countDownLatch;

    public MyTask(CountDownLatch countDownLatch {
         this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
         try {
             //Do somethings
             //
             this.countDownLatch.countDown();//important
         } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
         }
     }
}

CountDownLatch countDownLatch = new CountDownLatch(NUMBER_OF_TASKS);
ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
for (int i = 0; i < NUMBER_OF_TASKS; i++){
     taskExecutor.execute(new MyTask(countDownLatch));
}
countDownLatch.await();
System.out.println("Finish tasks");
Tuan Pham
quelle
2

Ich habe das folgende Arbeitsbeispiel erstellt. Die Idee ist, eine Möglichkeit zu haben, einen Aufgabenpool (ich verwende eine Warteschlange als Beispiel) mit vielen Threads (programmgesteuert durch die Anzahl der Aufgaben / Schwellenwert bestimmt) zu verarbeiten und zu warten, bis alle Threads abgeschlossen sind, um mit einer anderen Verarbeitung fortzufahren.

import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** Testing CountDownLatch and ExecutorService to manage scenario where
 * multiple Threads work together to complete tasks from a single
 * resource provider, so the processing can be faster. */
public class ThreadCountDown {

private CountDownLatch threadsCountdown = null;
private static Queue<Integer> tasks = new PriorityQueue<>();

public static void main(String[] args) {
    // Create a queue with "Tasks"
    int numberOfTasks = 2000;
    while(numberOfTasks-- > 0) {
        tasks.add(numberOfTasks);
    }

    // Initiate Processing of Tasks
    ThreadCountDown main = new ThreadCountDown();
    main.process(tasks);
}

/* Receiving the Tasks to process, and creating multiple Threads
* to process in parallel. */
private void process(Queue<Integer> tasks) {
    int numberOfThreads = getNumberOfThreadsRequired(tasks.size());
    threadsCountdown = new CountDownLatch(numberOfThreads);
    ExecutorService threadExecutor = Executors.newFixedThreadPool(numberOfThreads);

    //Initialize each Thread
    while(numberOfThreads-- > 0) {
        System.out.println("Initializing Thread: "+numberOfThreads);
        threadExecutor.execute(new MyThread("Thread "+numberOfThreads));
    }

    try {
        //Shutdown the Executor, so it cannot receive more Threads.
        threadExecutor.shutdown();
        threadsCountdown.await();
        System.out.println("ALL THREADS COMPLETED!");
        //continue With Some Other Process Here
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
}

/* Determine the number of Threads to create */
private int getNumberOfThreadsRequired(int size) {
    int threshold = 100;
    int threads = size / threshold;
    if( size > (threads*threshold) ){
        threads++;
    }
    return threads;
}

/* Task Provider. All Threads will get their task from here */
private synchronized static Integer getTask(){
    return tasks.poll();
}

/* The Threads will get Tasks and process them, while still available.
* When no more tasks available, the thread will complete and reduce the threadsCountdown */
private class MyThread implements Runnable {

    private String threadName;

    protected MyThread(String threadName) {
        super();
        this.threadName = threadName;
    }

    @Override
    public void run() {
        Integer task;
        try{
            //Check in the Task pool if anything pending to process
            while( (task = getTask()) != null ){
                processTask(task);
            }
        }catch (Exception ex){
            ex.printStackTrace();
        }finally {
            /*Reduce count when no more tasks to process. Eventually all
            Threads will end-up here, reducing the count to 0, allowing
            the flow to continue after threadsCountdown.await(); */
            threadsCountdown.countDown();
        }
    }

    private void processTask(Integer task){
        try{
            System.out.println(this.threadName+" is Working on Task: "+ task);
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }
}
}

Ich hoffe es hilft!

Fernando Gil
quelle
1

Sie können Ihre eigene Unterklasse von ExecutorCompletionService zum Umschließen taskExecutorund Ihre eigene Implementierung von BlockingQueue verwenden , um informiert zu werden, wenn jede Aufgabe abgeschlossen ist, und einen beliebigen Rückruf oder eine andere Aktion ausführen, wenn die Anzahl der abgeschlossenen Aufgaben Ihr gewünschtes Ziel erreicht.

Alex Martelli
quelle
1

Sie sollten verwenden executorService.shutdown()und executorService.awaitTerminationMethode.

Ein Beispiel wie folgt:

public class ScheduledThreadPoolExample {

    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
        executorService.scheduleAtFixedRate(() -> System.out.println("process task."),
                0, 1, TimeUnit.SECONDS);

        TimeUnit.SECONDS.sleep(10);
        executorService.shutdown();
        executorService.awaitTermination(1, TimeUnit.DAYS);
    }

}
Rollen Holt
quelle
ist abwartenTermination () notwendig nach dem Herunterfahren () /
gaurav
1

Also poste ich meine Antwort von der verknüpften Frage hier, falls jemand einen einfacheren Weg möchte, dies zu tun

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture[] futures = new CompletableFuture[10];
int i = 0;
while (...) {
    futures[i++] =  CompletableFuture.runAsync(runner, executor);
}

CompletableFuture.allOf(futures).join(); // THis will wait until all future ready.
Mạnh Quyết Nguyễn
quelle
0

Java 8 - Wir können die Stream-API verwenden, um Streams zu verarbeiten. Bitte sehen Sie den Ausschnitt unten

final List<Runnable> tasks = ...; //or any other functional interface
tasks.stream().parallel().forEach(Runnable::run) // Uses default pool

//alternatively to specify parallelism 
new ForkJoinPool(15).submit(
          () -> tasks.stream().parallel().forEach(Runnable::run) 
    ).get();
Vlad
quelle
2
Hallo Vlad, willkommen bei StackOverflow. Können Sie bitte Ihre Antwort bearbeiten, um zu erklären, wie dies die Frage beantwortet und was der Code bewirkt? Von Nur-Code-Antworten wird hier abgeraten. Vielen Dank!
Tim Malone
Dieser Beitrag befasst sich mit Parallelität. Parallelität! = Parallelität
GabrielBB
0

ExecutorService WORKER_THREAD_POOL 
  = Executors.newFixedThreadPool(10);
CountDownLatch latch = new CountDownLatch(2);
for (int i = 0; i < 2; i++) {
    WORKER_THREAD_POOL.submit(() -> {
        try {
            // doSomething();
            latch.countDown();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    });
}

// wait for the latch to be decremented by the two remaining threads
latch.await();

Wenn Sie doSomething()einige andere Ausnahmen auslösen, wird der latch.countDown()Schein nicht ausgeführt. Was soll ich also tun?

Pengfei Zhan
quelle
0

Wenn Sie mehr Thread ExecutionServices SEQUENTIALLY verwenden und warten möchten, bis JEDER EXECUTIONSERVICE abgeschlossen ist. Der beste Weg ist wie unten;

ExecutorService executer1 = Executors.newFixedThreadPool(THREAD_SIZE1);
for (<loop>) {
   executer1.execute(new Runnable() {
            @Override
            public void run() {
                ...
            }
        });
} 
executer1.shutdown();

try{
   executer1.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);

   ExecutorService executer2 = Executors.newFixedThreadPool(THREAD_SIZE2);
   for (true) {
      executer2.execute(new Runnable() {
            @Override
            public void run() {
                 ...
            }
        });
   } 
   executer2.shutdown();
} catch (Exception e){
 ...
}
Dr. X.
quelle
-1

Dies könnte helfen

Log.i(LOG_TAG, "shutting down executor...");
executor.shutdown();
while (true) {
                try {
                    Log.i(LOG_TAG, "Waiting for executor to terminate...");
                    if (executor.isTerminated())
                        break;
                    if (executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                        break;
                    }
                } catch (InterruptedException ignored) {}
            }
Amol Desai
quelle
-1

Sie können waitTillDone () für diese Runner- Klasse aufrufen :

Runner runner = Runner.runner(4); // create pool with 4 threads in thread pool

while(...) {
    runner.run(new MyTask()); // here you submit your task
}


runner.waitTillDone(); // and this blocks until all tasks are finished (or failed)


runner.shutdown(); // once you done you can shutdown the runner

Sie können diese Klasse wiederverwenden und waitTillDone () so oft aufrufen, wie Sie möchten, bevor Sie shutdown () aufrufen . Außerdem ist Ihr Code äußerst einfach . Auch Sie müssen nicht wissen , die Anzahl der Aufgaben im Voraus.

Um es zu benutzen, füge einfach diesen Gradle / Maven hinzu compile 'com.github.matejtymes:javafixes:1.3.1' Abhängigkeit zu Ihrem Projekt hinzu.

Weitere Details finden Sie hier:

https://github.com/MatejTymes/JavaFixes

Matej Tymes
quelle
-2

In Executor gibt es eine Methode getActiveCount(), die die Anzahl der aktiven Threads angibt.

Nachdem wir den Thread überspannt haben, können wir überprüfen, ob der activeCount()Wert ist 0. Sobald der Wert Null ist, werden derzeit keine aktiven Threads ausgeführt, was bedeutet, dass die Aufgabe abgeschlossen ist:

while (true) {
    if (executor.getActiveCount() == 0) {
    //ur own piece of code
    break;
    }
}
Benutzer
quelle
3
Keine gute Idee, siehe stackoverflow.com/a/7271685/1166992 und javadoc: "Gibt die ungefähre Anzahl von Threads zurück, die aktiv Aufgaben ausführen."
Olivier Faucheux