@Async verhindern, dass ein Thread fortgesetzt wird, bis andere Threads beendet sind

Question

Ich habe eine Anwendung, wo für eine bestimmte Anzahl von Zeiten etwas berechnet werden muss. Diese Berechnungsfunktion hat die Annotation @Async (aus dem Spring Framework), die es ermöglicht, diese Berechnungen auf 4 Threads auszuführen. Das Problem ist, dass ich ungefähr 40000 dieser Berechnungen benötige und ich möchte die Zeit zwischen dem Anfang und dem Ende aller Berechnungen wissen, also sehe ich, wie spät es vor und nach der for-Schleife ist, die die Berechnungsfunktionen aufruft. Aber jetzt werden alle Berechnungen in eine Warteschlange gestellt, so dass die for-Schleife sofort beendet wird und die Zeit ungefähr 1 Sekunde beträgt, während es einige Stunden dauert, bis die Berechnungen abgeschlossen sind. Ich habe versucht, eine maximale Warteschlangengröße auf etwa 100 zu setzen (auch gut, um die Speichernutzung zu reduzieren), aber dies ist auch keine Lösung, da ich die letzten 100 Berechnungen in der Gesamtzeit verpassen werde. Gibt es eine Möglichkeit, den ausführenden Code unmittelbar nach der for-Schleife anzuhalten, bis alle Threads ihre Arbeit erledigt haben, aber immer noch die @ Async-Annotation verwenden können? Dies würde in der folgenden Ausgabe

@Service 
public class Bar { 
    @Async 
    public void executeBla() { 
     System.out.println("Bla!"); 
    } 
} 

:

Dies ist ein Code, der das gleiche Problem zeigt:

Executing Klasse:

public class Foo { 
    public void executeBlaALotOfTimes() { 
     long before = System.currentTimeMillis(); 

     for (int i = 0; i<40000; i++) { 
      executeBla(); 
     } 

     long after = System.currentTimeMillis(); 

     System.out.println("Time it took for a lot of bla to execute: " + (after - before)/1000.0 + " seconds."); 
    } 
} 

Und die Klasse, die die Berechnungen durchführt (unter der Annahme, dass der Code in Foo unendlich schnell ausgeführt wird):

 
Time it took for a lot of bla to execute: 0.0 seconds. 
Bla! 
Bla! 
Bla! 
Bla! 
. 
. 
. 
etc 

Answer 1

Wenn Sie warten müssen, bis die Ausführungen abgeschlossen sind, können Sie einen Future als Rückgabewert zurückgeben, z.

@Async 
public Future<Void> executeBla() { 
    System.out.println("Bla!"); 
    return new AsyncResult<Void>(null); 
} 

Dies ist etwas künstlich, da gibt es keinen tatsächlichen Wert zurückgegeben werden, aber es wird immer noch der Angerufene Code erlauben zu warten, für alle Ausführungen beenden zu:

public void executeBlaALotOfTimes() { 
    long before = System.currentTimeMillis(); 

    Collection<Future<Void>> futures = new ArrayList<Future<Void>>(); 

    for (int i = 0; i<40000; i++) { 
     futures.add(executeBla()); 
    } 

    for (Future<Void> future : futures) { 
     future.get(); 
    } 

    long after = System.currentTimeMillis(); 

    System.out.println("Time it took for a lot of bla to execute: " + (after - before)/1000.0 + " seconds."); 
} 

Hier die erste Schleife feuert die asynchronen Aufgaben und speichert die Futures in einer Liste. Die Sekundenschleife iteriert dann über die Futures und wartet darauf, dass jeder beendet wird.

Answer 2

Eine Alternative besteht darin, eine ListenableFuture zurückzugeben und eine CountDownLatch zu verwenden.

@Async 
public ListenableFuture<Void> executeBla() { 
    try { 
     System.out.println("Bla!"); 
     return AsyncResult.forValue(null); 
    } catch (Throwable t) { 
     return AsyncResult.forExecutionException(t); 
    } 
} 

Dieses Szenario ermöglicht Ihnen ausdrücklich future.get() für jeden zukünftigen Aufruf zu vermeiden. Sie erreichen dies, indem Sie Erfolgs- und Misserfolg-Callbacks hinzufügen, die wiederum den CountDownLatch verringern, der genau für diesen Zweck erstellt wurde.

public void executeBlaALotOfTimes() { 
    long before = System.currentTimeMillis(); 

    int numExecutions = 40000; 
    CountDownLatch countDownLatch = new CountDownLatch(numExecutions); 

    for (int i = 0; i<numExecutions; i++) { 
     ListenableFuture<Void> future = executeBla(); 
     future.addCallback(
      aVoid -> countDownLatch.countDown(), 
      throwable -> countDownLatch.countDown() 
     ); 
    } 

    try { 
     countDownLatch.await(); 
    } catch (InterruptedException e) { 
     // Handle exception 
    } finally { 
     long after = System.currentTimeMillis(); 
     System.out.println("Time it took for a lot of bla to execute: " + (after - before)/1000.0 + " seconds."); 
    } 

}