Поддержание FIFO при использовании потоков в Java

Question

Я пытаюсь выполнить дорогостоящую задачу переменной длины для элементов, полученных последовательным образом. Крайне важно, чтобы порядок элементов поддерживался при быстрой обработке каждого элемента.

Ниже SSCWE (W для неверного!) И моя попытка распараллеливания обработки. Есть ли способ гарантировать, что каждый вызов processSomething() выполняется в своем потоке, сохраняя при этом FIFO, когда я смотрю на ExecutorCompletionService?

import java.util.concurrent.Callable; 
import java.util.concurrent.CompletionService; 
import java.util.concurrent.ExecutionException; 
import java.util.concurrent.ExecutorCompletionService; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 

public class ThreadingHelper { 

public static void main(String[] args) { 

    ExecutorService pool = Executors.newFixedThreadPool(20); 
    ExecutorCompletionService<String> compService 
     = new ExecutorCompletionService<String>(pool); 

    // process some data 
    processSomething(compService, "1."); 
    processSomething(compService, "2.."); 
    processSomething(compService, "3..."); 
    processSomething(compService, "4...."); 
    processSomething(compService, "5....."); 

    // print out the processed data 
    try { 
     System.out.println(compService.take().get()); 
     System.out.println(compService.take().get()); 
     System.out.println(compService.take().get()); 
     System.out.println(compService.take().get()); 
     System.out.println(compService.take().get()); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } catch (ExecutionException e) { 
     e.printStackTrace(); 
    } 

} 

public static synchronized void processSomething(
     CompletionService<String> compService, final String x) { 
    Callable<String> c = new Callable<String>() { 

     @Override 
     public String call() throws Exception { 
      // this represents the variable and expensive 
          // amount of time it takes to process x 
      long rand = (long) (Math.random() * 100); 
      Thread.sleep(rand); 

      // this represents the processing of x 
      String xProcessed = x.replace(".", "!"); 

      return xProcessed; 
     } 
    }; 
    compService.submit(c); 
} 
} 

Типичный выход

4!!!! 
2!! 
1! 
5!!!!! 
3!!! 

, но я хочу

1! 
2!! 
3!!! 
4!!!! 
5!!!!! 

Answer 1

Используйте Futures вместо CompletionService, чтобы получить результаты в определенном порядке, и по-прежнему пользоваться параллельно Исполнение:

ExecutorService pool = Executors.newFixedThreadPool(20); 
List<Future<String>> futures = new ArrayList<>(); 
futures.add(pool.submit(makeCallable("1."))); 
// ... 
for (Future<String> future : futures) { 
    try { 
     System.out.println(future.get()); 
    } catch (...) { 
     ... 
    } 
} 

public static Callable<String> makeCallable(String x) { 
    Callable<String> c = ...; 
    return c; 
}