В параллельных потоках Java 8 используется один и тот же поток для последовательности

Предположим, что у нас есть что-то вроде этого:

LongStream.range(0, 10).parallel()
.filter(l -> {
  System.out.format("filter: %s [%s]\n", l, Thread.currentThread().getName());
  return l % 2 == 0;
})
.map(l -> {
  System.out.format("map:    %s [%s]\n", l, Thread.currentThread().getName());
  return l;
});

Если вы запустите этот выпуск программы, будет что-то вроде:

filter: 6 [main]
map:    6 [main]
filter: 5 [main]
filter: 4 [ForkJoinPool.commonPool-worker-2]
map:    4 [ForkJoinPool.commonPool-worker-2]
filter: 1 [ForkJoinPool.commonPool-worker-3]
filter: 2 [ForkJoinPool.commonPool-worker-1]
filter: 0 [ForkJoinPool.commonPool-worker-3]
filter: 3 [ForkJoinPool.commonPool-worker-2]
filter: 8 [main]
filter: 7 [ForkJoinPool.commonPool-worker-2]
filter: 9 [ForkJoinPool.commonPool-worker-2]
map:    0 [ForkJoinPool.commonPool-worker-3]
map:    2 [ForkJoinPool.commonPool-worker-1]
map:    8 [main]`

Как мы видим, каждая последовательность задач для каждой длинной выполняется точно одним и тем же потоком. На чем мы можем положиться, или это просто совпадение? Могут ли потоки "делиться" задачами во время выполнения?

Ответ 1

Из описание пакета потока раздела "Побочные эффекты":

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

Ответ 2

Это не совпадение, так как Stream API в настоящее время реализован в OracleJDK/OpenJDK: операции без состояния (например, filter, map, peek и flatMap) объединяются в одну операцию, которая выполняет этапы последовательно в одном потоке. Однако введение какой-либо операции с состоянием может изменить ситуацию. Например, добавьте limit:

LongStream.range(0, 10).parallel()
.filter(l -> {
  System.out.format("filter: %s [%s]\n", l, Thread.currentThread().getName());
  return l % 2 == 0;
})
.limit(10)
.map(l -> {
  System.out.format("map:    %s [%s]\n", l, Thread.currentThread().getName());
  return l;
})
.forEach(x -> {});

Теперь предел вводит барьер, который разбивает трубопровод на две части. Результат выглядит следующим образом:

filter: 8 [ForkJoinPool.commonPool-worker-2]
filter: 9 [ForkJoinPool.commonPool-worker-7]
filter: 0 [ForkJoinPool.commonPool-worker-6]
filter: 1 [ForkJoinPool.commonPool-worker-3]
filter: 4 [ForkJoinPool.commonPool-worker-5]
filter: 2 [ForkJoinPool.commonPool-worker-1]
filter: 6 [main]
filter: 7 [ForkJoinPool.commonPool-worker-4]
filter: 3 [ForkJoinPool.commonPool-worker-6]
filter: 5 [ForkJoinPool.commonPool-worker-2]
map:    0 [ForkJoinPool.commonPool-worker-6]
map:    2 [ForkJoinPool.commonPool-worker-2]
map:    8 [ForkJoinPool.commonPool-worker-4]
map:    6 [main]
map:    4 [ForkJoinPool.commonPool-worker-6]

Смотрите, что элемент # 2 был отфильтрован в потоке FJP-1, но отображается в потоке FJP-2.

Обратите внимание, что, как @Misha правильно процитировал, даже для операций без гражданства нет гарантии, что тот же поток будет использоваться. Возможно, что будущие или альтернативные реализации Stream API изменят это поведение (например, используя подход "производитель-потребитель" ).