Streams paralelos

parallelStream()

El método parallelStream() es una característica introducida en Java 8 que permite procesar colecciones de datos de forma concurrente, aprovechando los múltiples núcleos de procesamiento disponibles en los equipos modernos. Esta funcionalidad forma parte del paradigma de la programación funcional en Java y está diseñada para mejorar el rendimiento en operaciones sobre grandes conjuntos de datos.

La API de Stream en Java ofrece dos tipos de procesamiento: secuencial (donde los elementos se procesan uno tras otro) y paralelo (donde los elementos se procesan concurrentemente). El método parallelStream() es una forma directa de crear un stream paralelo a partir de una colección.

Creación de Streams paralelos con parallelStream()

Para crear un stream paralelo a partir de una colección, simplemente invocamos el método parallelStream() en lugar del tradicional stream():

List<Integer> numeros = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

// Creación de un stream paralelo
Stream<Integer> streamParalelo = numeros.parallelStream();

Este método está disponible en todas las clases que implementan la interfaz Collection, lo que lo hace muy accesible para la mayoría de las estructuras de datos en Java.

Uso básico de parallelStream()

Veamos un ejemplo sencillo donde calculamos la suma de los cuadrados de una lista de números:

List<Integer> numeros = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

// Usando parallelStream para calcular la suma de cuadrados
int suma = numeros.parallelStream()
                 .map(n -> n * n)
                 .reduce(0, Integer::sum);

System.out.println("La suma de los cuadrados es: " + suma);

En este ejemplo, la operación map que calcula el cuadrado de cada número se ejecuta en paralelo, potencialmente en diferentes hilos, y luego los resultados se combinan mediante la operación reduce.

Operaciones terminales comunes con parallelStream()

Las operaciones terminales funcionan de la misma manera que con streams secuenciales, pero se ejecutan en paralelo:

List<String> nombres = Arrays.asList("Ana", "Juan", "Carlos", "María", "Pedro");

// forEach en paralelo (el orden no está garantizado)
nombres.parallelStream()
       .forEach(nombre -> System.out.println("Procesando: " + nombre));

// Filtrado y recolección en paralelo
List<String> nombresFiltrados = nombres.parallelStream()
                                      .filter(nombre -> nombre.length() > 4)
                                      .collect(Collectors.toList());

// Búsqueda en paralelo
Optional<String> primerNombreLargo = nombres.parallelStream()
                                          .filter(nombre -> nombre.length() > 4)
                                          .findAny();

Consideraciones de rendimiento

El uso de parallelStream() no siempre garantiza un mejor rendimiento. Hay varios factores a considerar:

• Tamaño de la colección: Para colecciones pequeñas, el costo de dividir el trabajo y combinar resultados puede superar los beneficios.

// Para colecciones pequeñas, el stream secuencial puede ser más eficiente
List<Integer> pequeña = Arrays.asList(1, 2, 3, 4, 5);
int resultado = pequeña.stream() // Mejor usar stream() en lugar de parallelStream()
                     .map(n -> n * 2)
                     .reduce(0, Integer::sum);

• Tipo de operaciones: Las operaciones que requieren mucho cálculo por elemento se benefician más del paralelismo.

// Operación computacionalmente intensiva que se beneficia del paralelismo
List<BigInteger> numeros = obtenerListaGrande();
List<BigInteger> resultados = numeros.parallelStream()
                                   .map(n -> calcularFactorial(n)) // Operación costosa
                                   .collect(Collectors.toList());

• Estructura de datos subyacente: Algunas colecciones como ArrayList se dividen más fácilmente que otras como LinkedList.

// ArrayList es más eficiente para streams paralelos que LinkedList
ArrayList<Integer> arrayList = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
int suma = arrayList.parallelStream()
                  .reduce(0, Integer::sum);

Ejemplo práctico: Procesamiento de una gran colección

Veamos un ejemplo más completo donde procesamos una colección grande de datos:

// Creamos una lista grande de números
List<Integer> numerosGrandes = new ArrayList<>();
for (int i = 0; i < 10_000_000; i++) {
    numerosGrandes.add(i);
}

// Medimos el tiempo con stream secuencial
long inicio = System.currentTimeMillis();
long sumaSecuencial = numerosGrandes.stream()
                                  .filter(n -> n % 2 == 0)
                                  .mapToLong(n -> (long) n * n)
                                  .sum();
long finSecuencial = System.currentTimeMillis();

// Medimos el tiempo con parallelStream
inicio = System.currentTimeMillis();
long sumaParalela = numerosGrandes.parallelStream()
                                .filter(n -> n % 2 == 0)
                                .mapToLong(n -> (long) n * n)
                                .sum();
long finParalelo = System.currentTimeMillis();

System.out.println("Tiempo secuencial: " + (finSecuencial - inicio) + " ms");
System.out.println("Tiempo paralelo: " + (finParalelo - inicio) + " ms");

Este ejemplo muestra cómo el procesamiento paralelo puede mejorar significativamente el rendimiento en operaciones sobre grandes conjuntos de datos.

Operaciones con estado y paralelismo

Es importante tener cuidado con las operaciones que mantienen estado cuando se utilizan streams paralelos, ya que pueden producir resultados inesperados:

// Las operaciones con estado como sorted pueden comportarse diferente en paralelo
List<Integer> desordenados = Arrays.asList(5, 2, 8, 1, 9, 3, 7, 4, 6);

// En un stream paralelo, el orden de ejecución de forEach no está garantizado
desordenados.parallelStream()
           .sorted()
           .forEach(n -> System.out.print(n + " ")); // Puede no mostrar en orden ascendente

System.out.println();

// Para garantizar el orden, usar forEachOrdered
desordenados.parallelStream()
           .sorted()
           .forEachOrdered(n -> System.out.print(n + " ")); // Garantiza el orden

El método parallelStream() proporciona una forma sencilla pero potente de aprovechar el procesamiento paralelo en Java, permitiendo mejorar el rendimiento de operaciones sobre colecciones grandes cuando se utiliza correctamente.

Diferencia entre Stream paralelo y Stream secuencial

Los Streams en Java representan una secuencia de elementos sobre los que se pueden realizar diversas operaciones. La principal diferencia entre un Stream paralelo y uno secuencial radica en cómo se procesan estos elementos y el impacto que esto tiene en el rendimiento, comportamiento y resultados de las operaciones.

Modelo de ejecución

Un Stream secuencial procesa los elementos uno tras otro en un único hilo de ejecución. Las operaciones se aplican a cada elemento en orden, siguiendo una secuencia predecible:

List<Integer> numeros = Arrays.asList(1, 2, 3, 4, 5);
numeros.stream()  // Stream secuencial
       .map(n -> n * 2)
       .forEach(n -> System.out.println(Thread.currentThread().getName() + ": " + n));

Por otro lado, un Stream paralelo divide automáticamente los datos en múltiples partes que se procesan simultáneamente en diferentes hilos:

List<Integer> numeros = Arrays.asList(1, 2, 3, 4, 5);
numeros.parallelStream()  // Stream paralelo
       .map(n -> n * 2)
       .forEach(n -> System.out.println(Thread.currentThread().getName() + ": " + n));

Al ejecutar este código, notarás que el Stream paralelo utiliza diferentes hilos del pool común ForkJoinPool, mientras que el secuencial usa solo el hilo principal.

Determinismo y orden de ejecución

Una diferencia fundamental es el determinismo en el orden de procesamiento:

• Los Streams secuenciales garantizan que los elementos se procesan en el orden definido por la fuente de datos.
• Los Streams paralelos no ofrecen garantías sobre el orden de procesamiento, lo que puede afectar a operaciones como forEach:

// Stream secuencial - orden predecible
List<String> nombres = Arrays.asList("Ana", "Carlos", "Elena", "David");
nombres.stream()
       .forEach(System.out::println);  // Imprime: Ana, Carlos, Elena, David

// Stream paralelo - orden impredecible
nombres.parallelStream()
       .forEach(System.out::println);  // El orden puede variar en cada ejecución

Para mantener el orden en Streams paralelos, podemos usar forEachOrdered:

nombres.parallelStream()
       .forEachOrdered(System.out::println);  // Mantiene el orden original

Conversión entre tipos de Streams

Es posible convertir entre ambos tipos de Streams durante una cadena de operaciones:

// De secuencial a paralelo
Stream<Integer> streamParalelo = Stream.of(1, 2, 3, 4, 5)
                                      .parallel();

// De paralelo a secuencial
Stream<Integer> streamSecuencial = streamParalelo.sequential();

// Verificar si un Stream es paralelo
boolean esParalelo = streamParalelo.isParallel();  // true

Impacto en operaciones de reducción

Las operaciones de reducción como reduce() se comportan de manera diferente según el tipo de Stream:

// Reducción secuencial - predecible
int sumaSecuencial = IntStream.range(1, 1000)
                             .reduce(0, (a, b) -> {
                                 System.out.println("Combinando " + a + " y " + b);
                                 return a + b;
                             });

// Reducción paralela - orden de combinación impredecible
int sumaParalela = IntStream.range(1, 1000)
                           .parallel()
                           .reduce(0, (a, b) -> {
                                 System.out.println("Combinando " + a + " y " + b);
                                 return a + b;
                             });

En el Stream paralelo, las combinaciones ocurren en un orden diferente y posiblemente con valores intermedios distintos.

Diferencias de rendimiento según el escenario

El rendimiento relativo entre ambos tipos de Streams varía según diferentes factores:

• Tamaño de los datos: Los Streams paralelos suelen ser más eficientes con colecciones grandes:

// Para colecciones pequeñas, el secuencial puede ser más rápido
List<Integer> pequeña = IntStream.range(1, 100).boxed().collect(Collectors.toList());

// Para colecciones grandes, el paralelo puede ofrecer mejor rendimiento
List<Integer> grande = IntStream.range(1, 10_000_000).boxed().collect(Collectors.toList());

• Tipo de operación: Las operaciones con alta intensidad computacional se benefician más del paralelismo:

// Operación simple - el paralelismo puede no compensar
List<Integer> numeros = obtenerNumeros();
long conteoSecuencial = numeros.stream().count();

// Operación compleja - el paralelismo puede mejorar significativamente
long sumaComplejaParalela = numeros.parallelStream()
                                  .map(n -> calcularOperacionCostosa(n))
                                  .reduce(0L, Long::sum);

• Estructura de datos subyacente: Algunas colecciones son más adecuadas para la división paralela:

// ArrayList - eficiente para paralelismo (acceso aleatorio rápido)
ArrayList<Integer> arrayList = new ArrayList<>(numeros);

// LinkedList - menos eficiente para paralelismo (división más costosa)
LinkedList<Integer> linkedList = new LinkedList<>(numeros);

Efectos secundarios y operaciones con estado

Las operaciones con efectos secundarios pueden comportarse de manera impredecible en Streams paralelos:

// Problema con efectos secundarios en Stream paralelo
List<Integer> resultados = new ArrayList<>();
numeros.parallelStream()
       .map(n -> n * 2)
       .forEach(resultados::add);  // Problema: ArrayList no es thread-safe

// Solución: usar un collector thread-safe
List<Integer> resultadosCorrectos = numeros.parallelStream()
                                         .map(n -> n * 2)
                                         .collect(Collectors.toList());

Las operaciones con estado como sorted(), distinct() o limit() pueden tener diferentes características de rendimiento:

// En streams secuenciales, limit es una operación de corto circuito eficiente
long primerosSecuencial = IntStream.range(1, 1_000_000)
                                  .filter(n -> n % 2 == 0)
                                  .limit(10)
                                  .count();  // Procesa solo los elementos necesarios

// En streams paralelos, limit puede procesar más elementos de los necesarios
long primerosParalelo = IntStream.range(1, 1_000_000)
                                .parallel()
                                .filter(n -> n % 2 == 0)
                                .limit(10)
                                .count();  // Puede procesar muchos más elementos

Escenarios de uso recomendados

La elección entre Streams secuenciales y paralelos debe basarse en el contexto específico:

• Usar Streams secuenciales cuando:

• La colección es pequeña (menos de miles de elementos)
• Las operaciones son simples y rápidas
• El orden de procesamiento es crítico
• Se trabaja con operaciones con estado como limit() o findFirst()

• Usar Streams paralelos cuando:

• La colección es grande (miles o millones de elementos)
• Las operaciones son computacionalmente intensivas
• La fuente de datos se puede dividir eficientemente (como ArrayList o arrays)
• Las operaciones son independientes entre sí (sin efectos secundarios)

Ejemplo comparativo de rendimiento

Este ejemplo muestra la diferencia de rendimiento entre ambos enfoques en un caso realista:

// Generamos una lista grande de números
List<Integer> numeros = IntStream.range(0, 10_000_000)
                               .boxed()
                               .collect(Collectors.toList());

// Función que simula un cálculo intensivo
Function<Integer, Double> calculoIntensivo = num -> {
    double resultado = 0;
    for (int i = 0; i < 100; i++) {
        resultado += Math.sin(Math.sqrt(num * Math.PI / i+1));
    }
    return resultado;
};

// Medición con stream secuencial
long inicioSec = System.currentTimeMillis();
double resultadoSec = numeros.stream()
                           .filter(n -> n % 2 == 0)
                           .mapToDouble(calculoIntensivo::apply)
                           .average()
                           .orElse(0);
long tiempoSec = System.currentTimeMillis() - inicioSec;

// Medición con stream paralelo
long inicioPar = System.currentTimeMillis();
double resultadoPar = numeros.parallelStream()
                           .filter(n -> n % 2 == 0)
                           .mapToDouble(calculoIntensivo::apply)
                           .average()
                           .orElse(0);
long tiempoPar = System.currentTimeMillis() - inicioPar;

System.out.println("Secuencial: " + tiempoSec + " ms");
System.out.println("Paralelo: " + tiempoPar + " ms");
System.out.println("Mejora: " + (tiempoSec / (double)tiempoPar) + "x");

En este tipo de escenario, con una colección grande y operaciones intensivas, el Stream paralelo puede ofrecer una mejora de rendimiento significativa, potencialmente proporcional al número de núcleos disponibles en el sistema.