ExecutorCompletionService等待最多n秒完成

Question

我正在使用ExecutorCompletionService提交幾個任務。然後我想等待最長時間，例如5秒，然後停止處理。

ExecutorService executorService = Executors.newFixedThreadPool(10);  
CompletionService<String> completionService = new ExecutorCompletionService<String>(
      executorService); 
List<Callable<String>> callables = createCallables(); //each callable sleeps randomly between 1-10 seconds and then prints the thread name 
for (Callable<String> callable : callables) 
    taskCompletionService.submit(callable); 
for (int i = 0; i < callables.size(); i++) { 
    Future<String> result = completionService.take(); 
    System.out.println(result.get()); 
} 

現在我不想等待超過5秒鐘完成所有任務。我只想收集5秒內完成的任務結果。我怎樣才能做到這一點？

executorService.shutdown(); 
executorService.awaitTermination(5, TimeUnit.SECONDS); 

我已經使用上executorServiceshutdown和awaitTermination，但我的主線程仍然等待所有提交的任務來完成，它需要所有的任務10秒內完成，並打印每個線程的名字。如何在5秒內停止處理？

Answer 1

這裏提到的主要問題是您的代碼正在等待任務完成，然後可以調用shutdown()。基本上這是因爲CompletionService.take()將阻塞，直到任務完成。此外，由於CompletionService不適合您，因此您需要跟蹤獲取任務結果所需的累計時間。

這個想法是使用poll(long, TimeUnit)而不是並將null結果解釋爲超時過期，之後您可以立即關閉執行程序服務。例如，這可以做到這樣的：

try { 
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
    CompletionService<String> completionService = new ExecutorCompletionService<>(executorService); 
    // each callable sleeps randomly between 1-10 seconds and then prints the thread name 
    List<Callable<String>> callables = createCallables(); 
    for (Callable<String> callable : callables) { 
    completionService.submit(callable); 
    } 
    final long timeout = 5_000_000_000L; // 5 seconds in nanos 
    long elapsed = 0L; 
    int count = 0; 
    final long start = System.nanoTime(); 
    // while not timed out and not all tasks have completed 
    while (((elapsed = System.nanoTime() - start) < timeout) && (count < callables.size())) { 
    // wait for at most the remaining time before timeout 
    Future<String> result = completionService.poll(timeout - elapsed, TimeUnit.NANOSECONDS); 
    if (result == null) { 
     System.out.println("timed out after " + count + " tasks and " + ((System.nanoTime() - start)/1_000_000L) + " ms"); 
     break; 
    } 
    count++; 
    System.out.println(result.get()); 
    } 
    executorService.shutdownNow(); 
    System.out.println("done"); 
} catch (Exception e) { 
    e.printStackTrace(); 
} 

我能夠測試它與createCallables()是這樣實現的：

private static List<Callable<String>> createCallables() { 
    Random rand = new Random(System.nanoTime()); 
    List<Callable<String>> list = new ArrayList<>(); 
    for (int i=0; i<10; i++) { 
    // between 1 and 10s 
    final long time = 1000L * (1L + rand.nextInt(10)); 
    list.add(new Callable<String>() { 
     @Override 
     public String call() throws Exception { 
     Thread.sleep(time); 
     return "ok after " + time + "s on thread " + Thread.currentThread(); 
     } 
    }); 
    } 
    return list; 
}