Faça stream paralelo no resultado de flatMap

Question

Considere o seguinte código simples:

Stream.of(1)
  .flatMap(x -> IntStream.range(0, 1024).boxed())
  .parallel() // Moving this before flatMap has the same effect because it's just a property of the entire stream
  .forEach(x -> {
     System.out.println("Thread: " + Thread.currentThread().getName());
  });

Por muito tempo, pensei que Java teria execução paralela para elementos mesmo depois flatMap. Mas o código acima imprime todos "Thread: main", o que prova que meu pensamento está errado.

Uma maneira simples de torná-lo paralelo depois flatMapseria coletar e transmitir novamente:

Stream.of(1)
  .flatMap(x -> IntStream.range(0, 1024).boxed())
  .parallel() // Moving this before flatMap has the same effect because it's just a property of the entire stream
  .collect(Collectors.toList())
  .parallelStream()
  .forEach(x -> {
     System.out.println("Thread: " + Thread.currentThread().getName());
  });

Eu queria saber se existe uma maneira melhor, e sobre a escolha do design de flatMapque só paralelize o stream antes da chamada, mas não depois da chamada.

========= Mais esclarecimentos sobre a questão ========

A partir de algumas respostas, parece que minha pergunta não foi totalmente transmitida. Como @Andreas disse, se eu começar com um Stream de 3 elementos, pode haver 3 threads em execução.

Mas minha pergunta realmente é: Java Stream usa um ForkJoinPool comum que tem um tamanho padrão igual a um a menos que o número de núcleos, de acordo com este post . Agora suponha que eu tenha 64 núcleos, então espero que meu código acima veja muitos threads diferentes depois flatMap, mas na verdade, ele vê apenas um (ou 3 no caso de Andreas). A propósito, eu costumava isParallelobservar que o riacho é paralelo.

Para ser honesto, eu não estava fazendo essa pergunta por puro interesse acadêmico. Encontrei esse problema em um projeto que apresenta uma longa cadeia de operações de fluxo para transformar um conjunto de dados. A cadeia começa com um único arquivo e explode em vários elementos flatMap. Mas, aparentemente, em meu experimento, ele NÃO explora totalmente minha máquina (que tem 64 núcleos), mas usa apenas um núcleo (a partir da observação do uso da CPU).

Answer 1

Eu estava pensando [...] na escolha do design de flatMapque apenas paraleliza o stream antes da chamada, mas não depois da chamada.

Você está enganado. Todas as etapas antes e depois do flatMapsão executadas em paralelo, mas só divide o fluxo original entre os encadeamentos. A flatMapoperação é então tratada por um desses encadeamentos e seu fluxo não é dividido.

Como seu fluxo original tem apenas 1 elemento, ele não pode ser dividido e, portanto parallel, não tem efeito.

Tente mudar para Stream.of(1, 2, 3)e você verá que o forEach, que está após o flatMap, na verdade é executado em 3 threads diferentes.

Answer 2

A documentação paraforEach especifica:

Para qualquer elemento, a ação pode ser executada a qualquer momento e em qualquer thread que a biblioteca escolher.

Em particular, "executar todas as operações no encadeamento de chamada" parece uma boa implementação amplamente segura.

Observe que sua tentativa de paralelizar o fluxo não requer nenhum paralelismo específico, mas é muito mais provável que você veja um efeito com isso:

IntStream.range(0, 1024).boxed()
  .parallel()
  .map(i -> "Thread: " + Thread.currentThread().getName())
  .forEach(System.out::println);