Streams paralelos: sintaxis y uso

1. Repaso de Stream API

Ya conoces la Stream API: es una forma cómoda de trabajar con colecciones que permite escribir código compacto y claro para procesar datos: filtrado, ordenación, recuento, etc.

He aquí un ejemplo clásico:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

int sum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

System.out.println(sum); // 6 (2 + 4)

En este ejemplo, la colección se convierte en un stream (stream()), de él se seleccionan solo los números pares, luego se transforman en int, y el resultado se suma con sum().

Stream API hace el código más corto y expresivo: en lugar de describir paso a paso cómo ocurre todo, simplemente dices qué quieres obtener. Además, si es necesario, puedes cambiar fácilmente a procesamiento paralelo — con una sola línea.

2. Streams paralelos: sintaxis y principio de funcionamiento

¿Cómo hacer paralelo un stream?

Es sencillo: en lugar de stream() usa parallelStream(). O llama al método .parallel() sobre un stream existente.

List<Integer> numbers = ...;

int sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

O así:

numbers.stream()
    .parallel() // lo convertimos en un stream paralelo
    .filter(...)
    .map(...)
    .sum();

¿Qué ocurre bajo el capó?

• La colección se divide automáticamente en partes.
• Cada parte se procesa en un hilo separado (se usa ForkJoinPool — un pool especial de hilos).
• Los resultados se combinan en un valor final.

Es decir, si tienes un procesador multinúcleo, el procesamiento va realmente en paralelo — por ejemplo, el filtrado y el cálculo de la suma pueden ejecutarse al mismo tiempo en varios núcleos.

¿Dónde es especialmente útil?

• Procesamiento de colecciones grandes (decenas de miles de elementos o más).
• Cálculos complejos para cada elemento.
• Ausencia de necesidad de preservar un orden estricto de procesamiento.

Ejemplo: comparación de stream secuencial y paralelo

Veamos un ejemplo sencillo con el procesamiento de una lista grande.

import java.util.*;
import java.util.stream.*;

public class ParallelStreamDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Stream secuencial
        long time1 = System.currentTimeMillis();
        long count1 = numbers.stream()
            .filter(n -> n % 2 == 0)
            .count();
        long time2 = System.currentTimeMillis();
        System.out.println("Secuencial: " + (time2 - time1) + " ms, pares: " + count1);

        // Stream paralelo
        long time3 = System.currentTimeMillis();
        long count2 = numbers.parallelStream()
            .filter(n -> n % 2 == 0)
            .count();
        long time4 = System.currentTimeMillis();
        System.out.println("En paralelo: " + (time4 - time3) + " ms, pares: " + count2);
    }
}

Prueba este código en tu equipo: lo más probable es que el stream paralelo procese la colección más rápido (especialmente si tu procesador tiene varios núcleos). ¡Pero no siempre! Más detalles, abajo.

3. Cómo funciona: ForkJoinPool y particionado automático

Los streams paralelos usan bajo el capó ForkJoinPool.commonPool(), que gestiona automáticamente la cantidad de hilos (normalmente — según el número de núcleos de CPU disponibles).

Esquemáticamente:

+-----------------------------+
|          Tu colección       |
+-----------------------------+
| 1  | 2  | 3  | ... | 10 mill.|
+----+----+----+-----+--------+
   |    |    |           |
   v    v    v           v
[Hilo1][Hilo2]...[HiloN]
   |    |    |           |
   +----+----+-----------+
        |
   [Combinación del resultado]

Cada hilo procesa su parte y luego se combinan los resultados.

4. Limitaciones y trampas

Los streams paralelos no son un botón mágico «acelerarlo todo». ¡A veces incluso ralentizan la ejecución!

Cuándo no conviene paralelizar:

• La colección es pequeña (hasta ~1000 elementos).
• La operación por elemento es muy rápida (por ejemplo, solo n * 2).
• Necesitas un orden estricto de procesamiento (por ejemplo, para escritura secuencial en un archivo).

¿Por qué? Crear y sincronizar hilos también lleva tiempo. Si la tarea es «pequeña», la sobrecarga puede superar el beneficio del paralelismo.

Los efectos secundarios: el enemigo del paralelismo

Si tus operaciones dentro del stream modifican variables externas, ¡ten cuidado!

Mal ejemplo:

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int[] sum = {0};

numbers.parallelStream().forEach(n -> sum[0] += n);

System.out.println(sum[0]); // ??? (esperas 15, pero puedes obtener cualquier cosa)

¿Por qué? Porque varios hilos cambian la misma variable a la vez — aparece una race condition (condición de carrera). El valor final puede ser incorrecto.

La manera correcta es usar métodos del stream que devuelven el resultado:

int sum = numbers.parallelStream().mapToInt(n -> n).sum();

No todas las colecciones se «paralelizan» igual

Algunas colecciones (por ejemplo, un ArrayList) se particionan bien. En cambio, LinkedList o un stream con número infinito de elementos (por ejemplo, Stream.generate(...)) — no tanto.

5. Práctica: comparación de rendimiento

Ejemplo: búsqueda del número máximo

import java.util.*;
import java.util.stream.*;

public class ParallelMaxDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 30_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Secuencialmente
        long t1 = System.currentTimeMillis();
        int max1 = numbers.stream().max(Integer::compareTo).get();
        long t2 = System.currentTimeMillis();
        System.out.println("Secuencialmente: " + (t2 - t1) + " ms, max = " + max1);

        // En paralelo
        long t3 = System.currentTimeMillis();
        int max2 = numbers.parallelStream().max(Integer::compareTo).get();
        long t4 = System.currentTimeMillis();
        System.out.println("En paralelo: " + (t4 - t3) + " ms, max = " + max2);
    }
}

¿Qué veremos? En procesadores multinúcleo modernos, el stream paralelo suele ser más rápido. Pero si sustituyes 30_000_000 por 1000, no habrá diferencia — ¡e incluso a veces el paralelo será más lento!

6. Casos de uso: filtrado, agregación y ordenación

Filtrado y conteo

List<String> names = Arrays.asList("Anna", "Boris", "Vasily", "Grigory", "Daria", "Yegor", "Eugene");

long count = names.parallelStream()
    .filter(name -> name.length() == 4)
    .count();

System.out.println("Nombres de longitud 4: " + count);

Agrupación

List<String> words = Arrays.asList("kot", "kit", "kot", "pes", "kit", "kot");

Map<String, Long> freq = words.parallelStream()
    .collect(Collectors.groupingBy(
        w -> w,
        Collectors.counting()
    ));

System.out.println(freq); // {pes=1, kit=2, kot=3}

Ordenación (¡pero aquí el paralelismo no siempre aporta mejoras!)

List<Integer> bigList = IntStream.rangeClosed(1, 5_000_000)
                                 .boxed()
                                 .collect(Collectors.toList());

long t1 = System.currentTimeMillis();
List<Integer> sorted = bigList.parallelStream()
    .sorted()
    .collect(Collectors.toList());
long t2 = System.currentTimeMillis();

System.out.println("Ordenación paralela: " + (t2 - t1) + " ms");

7. Matices importantes y recomendaciones

Cuándo conviene usar parallelStream()

• La colección es grande (decenas de miles de elementos o más).
• La operación por elemento es «pesada» (cálculos complejos, trabajo con archivos/red).
• No hay dependencia del orden de los elementos.
• No hay efectos secundarios (no se modifican variables externas).

Cuándo NO conviene usar parallelStream()

• La colección es pequeña.
• La operación es rápida.
• Necesitas preservar el orden estricto.
• Hay acceso a variables compartidas (considera colecciones thread-safe u otros enfoques).

¿Cómo saber cuántos hilos se utilizan?

Por defecto — según el número de núcleos del procesador: Runtime.getRuntime().availableProcessors(). Puedes cambiar este comportamiento mediante una propiedad del sistema:

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

Haz esto solo si entiendes las consecuencias — de lo contrario, puedes saturar la CPU y obtener un sistema más lento.

8. Errores típicos al trabajar con streams paralelos

Error n.º 1: efectos secundarios dentro de forEach
Muchos piensan: «¡Ahora rellenaré la lista en paralelo!»

List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
    .parallel()
    .forEach(result::add); // ¡PELIGROSO!
System.out.println(result.size()); // ¡El resultado es aleatorio!

¿Por qué es malo? ArrayList no es thread-safe, y al añadir simultáneamente desde distintos hilos el resultado es impredecible: puede haber omisiones, duplicados, excepciones.

Solución: Usa los métodos de recogida del stream (collect), que garantizan la seguridad por sí mismos, o colecciones especiales.

List<Integer> result = IntStream.range(0, 1_000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());

Error n.º 2: esperar aceleración en tareas pequeñas
¡El paralelismo no es gratis! Si la colección es pequeña, el stream paralelo puede funcionar más lento debido a la sobrecarga de planificación y sincronización.

Error n.º 3: pérdida del orden
Si te importa el orden de los elementos (por ejemplo, al escribir en un archivo), no uses streams paralelos: el orden no está garantizado (o funcionará más lento).

Error n.º 4: uso de colecciones «poco afortunadas»
Algunas colecciones (por ejemplo, LinkedList, estructuras no estándar) se particionan mal — la eficacia del paralelismo se reduce.

Error n.º 5: ignorar la seguridad de hilos al recolectar resultados
Si recoges los resultados manualmente (por ejemplo, añadiendo a una lista), usa colecciones thread-safe (CopyOnWriteArrayList, ConcurrentLinkedQueue) o métodos de recogida del stream.