ExecutorCompletionService ждет n секунд max для завершения

Question

Я использую ExecutorCompletionService для отправки нескольких задач. Я хочу подождать max, скажем 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); 

Я использовал shutdown и awaitTermination на executorService, но мой основной поток все еще ждет всех представленных задач для завершения и занимает 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; 
}