Streams Paralelos em Java

A Stream API oferece uma maneira notavelmente simples de paralelizar operações: os streams paralelos. A ideia é dividir o trabalho de processamento de uma coleção de dados entre os múltiplos núcleos de um processador, com o objetivo de acelerar os cálculos e melhorar a performance.

No entanto, o paralelismo não é uma "bala de prata". Usá-lo indiscriminadamente pode, na verdade, piorar o desempenho, pois envolve um custo de dividir os dados, gerenciar as threads e combinar os resultados. O paralelismo é uma ferramenta de otimização que deve ser usada com critério.

Criando um Stream Paralelo

Existem duas maneiras principais de obter um stream paralelo:

A partir de um stream sequencial existente, chamando o método parallel().
A partir de uma coleção, chamando o método parallelStream(), disponível na interface Collection.

Se a máquina onde o código é executado não for multicore, o stream paralelo será executado como um stream sequencial.

O uso de streams paralelos é sintaticamente similar ao de streams sequenciais. Por exemplo:

import java.util.Optional;
import java.util.stream.Stream;

public class Program {
    public static void main(String[] args) {
        Stream<Integer> numbersStream = Stream.of(1, 2, 3, 4, 5, 6);

        // Converte o stream para paralelo antes de aplicar a redução
        Optional<Integer> result = numbersStream.parallel().reduce(1, (x, y) -> x * y);

        result.ifPresent(System.out::println); // 720
    }
}

Ordem e Determinismo

A principal diferença entre a execução sequencial e a paralela é a ordem de processamento. Em um stream sequencial, os elementos são processados na mesma ordem em que aparecem na fonte de dados. Em streams paralelos, a ordem pode mudar porque os elementos são processados em blocos independentes.

O exemplo abaixo ilustra essa diferença.

import java.util.Arrays;
import java.util.List;

public class Program {
    public static void main(String[] args) {
        List<String> people = Arrays.asList("Tom", "Bob", "Sam", "Kate", "Tim");

        System.out.println("Stream Sequencial");
        people.stream()
              .filter(p -> p.length() == 3)
              .forEach(System.out::println);

        System.out.println("\nStream Paralelo - varia a cada execução");
        people.parallelStream()
              .filter(p -> p.length() == 3)
              .forEach(System.out::println);
    }
}

Primeiro, criamos um stream sequencial, filtramos as strings com 3 caracteres e as imprimimos. Neste caso, as operações são aplicadas aos elementos na ordem em que eles aparecem na lista.

Em seguida, criamos um stream paralelo com parallelStream() e aplicamos as mesmas operações. Agora, a ordem em que os elementos são processados não é determinística.

Saída possível:

Stream Sequencial
Tom
Bob
Sam
Tim

Stream Paralelo - varia a cada execução
Sam
Tim
Bob
Tom

Enquanto a saída do stream sequencial é sempre a mesma, a saída do stream paralelo é não determinística e pode variar a cada execução.

Requisitos para Operações Paralelas Seguras

Para que uma operação em um stream paralelo produza um resultado correto e consistente, a função aplicada (especialmente em reduções) deve seguir duas regras:

Ser sem estado (Stateless): A função não deve depender de nenhum estado externo que possa ser modificado durante a execução.
Ser associativa (Associative): A ordem de agrupamento dos operandos não deve alterar o resultado final. Ou seja, (a op b) op c deve ser igual a a op (b op c).

A multiplicação é um bom exemplo de uma operação associativa. Não importa como agrupamos os números, o produto final é o mesmo. É essa propriedade que permite que o stream seja dividido, processado em partes e combinado sem erros.

Exemplo seguro:

Stream<Integer> numbersStream = Stream.of(1, 2, 3, 4, 5, 6);
Integer result = numbersStream.parallel().reduce(1, (x, y) -> x * y);
System.out.println(result); // 720

Exemplo não seguro:

Stream<Integer> numbersStream = Stream.of(1, 2, 3, 4, 5, 6);
Integer result = numbersStream.parallel().reduce(0, (x, y) -> x - y);
System.out.println(result); // Resultado matematicamente incorreto

Fatores que Influenciam a Performance

Tamanho dos Dados (N): Para volumes pequenos de dados, a execução sequencial é quase sempre mais rápida. O paralelismo brilha em coleções com milhares ou milhões de itens.
Número de Núcleos: O ganho de performance é limitado pelo número de núcleos de processador disponíveis.
Estrutura da Fonte de Dados: A eficiência com que os dados podem ser divididos (splittable) é crucial. Fontes baseadas em arrays, como ArrayList, são ideais. Estruturas como LinkedList são péssimas para paralelização.
Operações sobre Tipos Primitivos: Operações em IntStream, LongStream, etc., são mais rápidas do que em Stream<Integer> devido à ausência de boxing/unboxing.

Preservando a Ordem em Streams Paralelos com `forEachOrdered`

O forEach pode produzir saídas em qualquer ordem, pois ele expõe a ordem de processamento bruto para maximizar a velocidade.

Para forçar a iteração na ordem original, mesmo em um contexto paralelo, use o método forEachOrdered.

// Supondo uma lista 'phones'
phones.parallelStream()
       .sorted()
       .forEachOrdered(s -> System.out.println(s));

Manter a ordem em um stream paralelo tem um custo de performance. Se a ordem não for importante, você pode relaxar essa restrição usando o método unordered(), o que pode aumentar o desempenho.

// Supondo uma lista 'phones'
phones.parallelStream()
       .sorted()
       .unordered()
       .forEach(s -> System.out.println(s));

Resumo

Streams paralelos são uma ferramenta de otimização, não um padrão de uso geral.
Para criar, use stream.parallel() ou collection.parallelStream().
O ganho de performance não é garantido e depende do volume de dados, da operação, da fonte e do hardware.
Operações de redução em paralelo devem ser sem estado e associativas para garantir a correção do resultado.
Fontes de dados facilmente divisíveis (como ArrayList) e streams de tipos primitivos (IntStream) são ideais para paralelismo.
Use forEachOrdered para garantir a ordem na iteração final, ciente do custo de performance.