Параллельный поток из HashSet не работает параллельно

Question

У меня есть набор элементов, которые я хочу обрабатывать параллельно. Когда я использую List, работает параллелизм. Однако, когда я использую Set, он не запускается параллельно.

Я написал пример кода, который показывает проблему:

public static void main(String[] args) { 
    ParallelTest test = new ParallelTest(); 

    List<Integer> list = Arrays.asList(1,2); 
    Set<Integer> set = new HashSet<>(list); 

    ForkJoinPool forkJoinPool = new ForkJoinPool(4); 

    System.out.println("set print"); 
    try { 
     forkJoinPool.submit(() -> 
      set.parallelStream().forEach(test::print) 
     ).get(); 
    } catch (Exception e) { 
     return; 
    } 

    System.out.println("\n\nlist print"); 
    try { 
     forkJoinPool.submit(() -> 
      list.parallelStream().forEach(test::print) 
     ).get(); 
    } catch (Exception e) { 
     return; 
    } 
} 

private void print(int i){ 
    System.out.println("start: " + i); 
    try { 
     TimeUnit.SECONDS.sleep(1); 
    } catch (InterruptedException e) { 
    } 
    System.out.println("end: " + i); 
} 

Это выход, который я получаю на Windows 7

set print 
start: 1 
end: 1 
start: 2 
end: 2 

list print 
start: 2 
start: 1 
end: 1 
end: 2 

Мы можем видеть, что первый элемент из Set должен был до того, как будет обработан второй элемент. Для List второй элемент начинается до завершения первого элемента.

Можете ли вы рассказать мне, что вызывает эту проблему, и как ее избежать, используя коллекцию Set?

Answer 1

Я могу воспроизвести поведение, которое вы видите, где параллелизм не совпадает с параллелизмом параллелизма пула форвардов, который вы указали. После установки параллелизма пула fork-join равным 10 и увеличения количества элементов в коллекции до 50, я вижу, что параллелизм потока на основе списка возрастает только до 6, тогда как параллелизм потока на основе набора никогда не становится выше 2.

Обратите внимание, однако, что эта методика отправки задачи в пул fork-join для запуска параллельного потока в этом пуле представляет собой «трюк» реализации и не гарантируется для работы. Действительно, потоки или пул потоков, которые используются для выполнения параллельных потоков, это неуказанный. По умолчанию используется общий пул fork-join, но в разных средах могут использоваться различные пулы потоков. (Рассмотрим контейнер внутри сервера приложений.)

В классе java.util.stream.AbstractTask поле LEAF_TARGET определяет величину выполненного разделения, что, в свою очередь, определяет величину параллелизма, которая может быть достигнута. Значение этого поля основано на ForkJoinPool.getCommonPoolParallelism(), который, конечно же, использует параллелизм общего пула, а не какой-либо пул, выполняющий задачи.

Возможно, это ошибка (см. Вопрос OpenJDK JDK-8190974), однако вся эта область не указана. Тем не менее, эта область системы определенно нуждается в развитии, например, с точки зрения политики разделения, количества доступных параллелизма, решения задач блокировки, среди других вопросов. В будущем выпуске JDK могут быть рассмотрены некоторые из этих проблем.

Между тем, можно управлять параллелизмом общего пула вил-пулов с использованием системных свойств. Если добавить эту строку в вашу программу,

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "10"); 

и запускать потоки в общем пуле (или, если вы отправляете их на свой собственный бассейн, который имеет достаточно высокий уровень набора параллелизма), вы заметите, что многие больше задач выполняется параллельно.

Вы также можете установить это свойство в командной строке, используя опцию -D.

Опять же, это не гарантированное поведение, и оно может измениться в будущем. Но этот метод, вероятно, будет работать для реализации JDK 8 в обозримом будущем.