@Async防止線程繼續，直到其他線程完成

Question

我有一個應用程序需要計算一定的次數。這個計算函數的註解@Async（來自Spring框架），可以在4個線程上運行這些計算。問題是我需要大約40000個這樣的計算，我想知道所有計算的開始和結束之間的時間，所以我看到在調用計算函數的for循環之前和之後的時間。但是現在所有的計算都放在一個隊列中，所以for循環立即完成，時間大約是1秒，而計算需要幾個小時才能完成。我已經嘗試將最大隊列大小設置爲大約100（也很好地減少內存使用量），但這也不是解決方案，因爲我會錯過最後100次計算所花費的總時間。有沒有辦法在for循環之後暫停執行的代碼，直到所有線程完成其工作，但仍然能夠使用@Async批註？

這是一些代碼，示出了同樣的問題：

執行類：

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."); 
    } 
} 

和進行計算的類：

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

這將導致以下的輸出（假設Foo中的代碼無限快地執行）：

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

Answer 1

如果您需要等待執行完成，則可以返回Future作爲返回值，例如，

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

這是虛假造作的，因爲有返回沒有實際價值，但它仍然會允許調用代碼等所有執行來完成：

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."); 
} 

在這裏，第一循環打完異步任務並將期貨存儲在列表中。秒循環然後迭代期貨，等待每一個完成。

Answer 2

另一種方法是返回ListenableFuture並使用CountDownLatch。

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

這種情況下，您可以避免明確調用每個未來的future.get()。您可以通過添加成功和失敗回調來實現這一目標，然後遞減CountDownLatch，這是爲此目的而創建的。

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."); 
    } 

}