Fluxuri paralele

Pe lângă fluxurile secvențiale, Stream API suportă și fluxuri paralele. Paralelizarea fluxurilor permite utilizarea mai multor nuclee ale procesorului (dacă mașina țintă este multi-core), ceea ce poate îmbunătăți performanța și accelera calculele.

Cu toate acestea, nu este întotdeauna corect să afirmăm că utilizarea fluxurilor paralele pe mașinile multi-core va crește performanța în mod garantat. În fiecare caz specific, este necesar să se testeze și să se verifice.

Pentru a face un flux secvențial obișnuit paralel, trebuie să apelăm metoda parallel a obiectului Stream. De asemenea, putem folosi metoda parallelStream() a interfeței Collection pentru a crea un flux paralel dintr-o colecție.

Dacă mașina de lucru nu este multi-core, fluxul se va executa secvențial.

Utilizarea fluxurilor paralele în multe cazuri va fi similară. De exemplu:

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);
       Optional<Integer> result = numbersStream.parallel().reduce((x, y) -> x * y);
       System.out.println(result.get()); // 720
   }
}

Un alt exemplu:

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("Flux secvențial");
       people.stream().filter(p -> p.length() == 3).forEach(System.out::println);

       System.out.println("\nFlux paralel");
       people.parallelStream().filter(p -> p.length() == 3).forEach(System.out::println);
   }
}

În acest caz, mai întâi creăm un flux pentru lista people și efectuăm câteva operațiuni în mod secvențial, căutând în listă șiruri de caractere cu lungimea egală cu 3 și afișându-le în consolă. În acest caz, toate operațiunile vor fi efectuate în ordinea în care elementele apar în listă.

Apoi, folosim metoda people.parallelStream() pentru a crea un flux paralel. Deși aplicăm aceleași operațiuni, ordinea în care elementele listei vor fi procesate nu este determinată.

Output în consolă:

Flux secvențial
Tom
Bob
Sam
Tim

Flux paralel
Sam
Tim
Bob
Tom

În cazul fluxului paralel, rezultatul este nedeterministic și poate varia.

Totuși, nu toate funcțiile pot fi transferate de la fluxurile secvențiale la cele paralele fără a compromite corectitudinea calculelor. Aceste funcții trebuie să fie fără stări și asociative, adică să producă același rezultat indiferent dacă se efectuează de la stânga la dreapta sau de la dreapta la stânga, ca în cazul înmulțirii numerelor. De exemplu:

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);

Aici se realizează înmulțirea numerelor. Nu contează dacă ordinea este 1 * 2 * 3 * 4 * (5 * 6) sau 5 * 6 * 1 * (2 * 3) * 4. Putem pune parantezele în orice mod, iar rezultatul va fi același, deoarece operația este asociativă și poate fi paralelizată.

Probleme de performanță în operațiunile paralele

Utilizarea fluxurilor paralele implică împărțirea datelor în părți, fiecare parte fiind procesată pe un nucleu separat al procesorului, iar la final aceste părți sunt combinate și se aplică operațiile finale. Iată câțiva factori care pot afecta performanța fluxurilor paralele:

• Dimensiunea datelor: Cu cât datele sunt mai mari, cu atât este mai complicat să le împărțim și apoi să le combinăm
• Numărul de nuclee ale procesorului: Teoretic, cu cât sunt mai multe nuclee, cu atât mai rapid va funcționa programul. Dacă mașina are un singur nucleu, nu are sens să folosim fluxuri paralele
• Structura datelor: Cu cât structura datelor este mai simplă, cu atât mai rapid vor fi efectuate operațiile. De exemplu, ArrayList este ușor de utilizat deoarece conține date neconectate. În schimb, LinkedList nu este cea mai bună alegere, deoarece fiecare element din listă este conectat cu cel anterior/următor, ceea ce face dificilă paralelizarea
• Tipuri de date primitive: Operațiile pe tipuri primitive vor fi efectuate mai rapid decât pe obiecte de clasă

Ordinea în fluxurile paralele

De obicei, elementele sunt transmise fluxului în aceeași ordine în care sunt definite în sursa de date. În fluxurile paralele, sistemul menține ordinea elementelor. O excepție este metoda forEach(), care poate afișa elementele într-o ordine aleatorie. Pentru a păstra ordinea, trebuie să folosim metoda forEachOrdered:

phones.parallelStream()
   .sorted()
   .forEachOrdered(s -> System.out.println(s));

Păstrarea ordinii în fluxurile paralele implică costuri suplimentare la execuție. Dar dacă ordinea nu este importantă, o putem dezactiva pentru a crește performanța, folosind metoda unordered:

phones.parallelStream()
   .sorted()
   .unordered()
   .forEach(s -> System.out.println(s));