Paralel stream-lər: sintaksis, tətbiq

1. Stream API-ni xatırlayaq

Artıq Stream API ilə tanışsınız — bu, kolleksiyalarla işləmək üçün rahat üsuldur və məlumatların emalı (filtrasiya, çeşidləmə, sayma və s.) üçün yığcam və anlaşılan kod yazmağa imkan verir.

Budur klassik bir nümunə:

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)

Bu nümunədə kolleksiya axına (stream()) çevrilir, oradan yalnız cüt ədədlər seçilir, sonra onlar int-ə çevrilir və nəticə sum() çağırışı ilə cəmlənir.

Stream API kodu daha qısa və ifadəli edir: addım-addım hər şeyin necə baş verdiyini təsvir etmək əvəzinə, sadəcə nə almaq istədiyinizi deyirsiniz. Üstəlik, lazım gəldikdə cəmi bir sətir ilə paralel emala asanlıqla keçmək olar.

2. Paralel stream-lər: sintaksis və iş prinsipi

Stream-i paralel necə edək?

Çox sadədir: stream() əvəzinə parallelStream() istifadə edin. Və ya mövcud axın üçün .parallel() metodunu çağırın.

List<Integer> numbers = ...;

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

Və ya belə:

numbers.stream()
    .parallel() // paralel stream-ə çeviririk
    .filter(...)
    .map(...)
    .sum();

Pərdə arxasında nə baş verir?

• Kolleksiya avtomatik olaraq hissələrə bölünür.
• Hər hissə ayrı bir thread-də emal olunur (ForkJoinPool — xüsusi thread hovuzudur — istifadə edilir).
• Nəticələr yekun dəyərə birləşdirilir.

Yəni çoxnüvəli prosessorunuz varsa, emal həqiqətən paralel gedir — məsələn, filtrasiya və cəmləmə eyni vaxtda bir neçə nüvədə icra oluna bilər.

Bu harada xüsusilə faydalıdır?

• Böyük kolleksiyaların emalı (on minlərlə element və daha çox).
• Hər element üçün mürəkkəb hesablamalar.
• Emalın dəqiq sıra ilə aparılmasına ehtiyac olmadıqda.

Nümunə: ardıcıl və paralel stream-in müqayisəsi

Gəlin böyük massivlə sadə bir nümunəyə baxaq.

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

        // Ardıcıl stream
        long time1 = System.currentTimeMillis();
        long count1 = numbers.stream()
            .filter(n -> n % 2 == 0)
            .count();
        long time2 = System.currentTimeMillis();
        System.out.println("Ardıcıl: " + (time2 - time1) + " ms, cütlərin sayı: " + count1);

        // Paralel stream
        long time3 = System.currentTimeMillis();
        long count2 = numbers.parallelStream()
            .filter(n -> n % 2 == 0)
            .count();
        long time4 = System.currentTimeMillis();
        System.out.println("Paralel: " + (time4 - time3) + " ms, cütlərin sayı: " + count2);
    }
}

Bu kodu öz kompüterinizdə yoxlayın — çox güman ki, paralel stream kolleksiyanı daha tez emal edəcək (xüsusən də çoxnüvəli prosessorunuz varsa). Amma həmişə yox! Nüanslar barədə — aşağıda.

3. Bu necə işləyir: ForkJoinPool və avtomatik bölmə

Paralel stream-lər pərdə arxasında ForkJoinPool.commonPool()-dan istifadə edir; o, thread-lərin sayını avtomatik idarə edir (adətən — mövcud prosessor nüvələrinin sayına görə).

Sxematik olaraq:

+-----------------------------+
|        Kolleksiyanız        |
+-----------------------------+
| 1  | 2  | 3  | ... | 10 mln |
+----+----+----+-----+--------+
   |    |    |           |
   v    v    v           v
[Thread1][Thread2]...[ThreadN]
   |    |    |           |
   +----+----+-----------+
        |
   [Nəticələrin birləşdirilməsi]

Hər thread öz hissəsini emal edir, sonra nəticələr toplanır.

4. Məhdudiyyətlər və tələlər

Paralel stream-lər — “hər şeyi sürətləndir” sehrli düyməsi deyil. Bəzən onlar hətta işi ləngidir!

Paralelləşdirmək nə zaman sərfəli deyil:

• Kolleksiya kiçikdirsə (təxminən 1000 elementə qədər).
• Hər element üzərində əməliyyat çox sürətlidirsə (məsələn, sadəcə n * 2).
• Sıralamanın ciddi qorunması vacibdirsə (məsələn, fayla ardıcıl yazma üçün).

Niyə? Thread-lərin yaradılması və sinxronlaşdırılması da vaxt aparır. Tapşırıq “xırda”dırsa, əlavə xərclər paralelliyin gətirdiyi qazancı üstələyə bilər.

Yan təsirlər — paralelliyin düşməni

Əgər stream daxilindəki əməliyyatlar xarici dəyişənləri dəyişirsə, ehtiyatlı olun!

Pis nümunə:

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]); // ??? (15 gözləyirsiniz, amma hər nə ola bilər)

Niyə? Çünki bir neçə thread eyni dəyişəni eyni vaxtda dəyişir — bu, race condition vəziyyətidir. Yekun dəyər səhv ola bilər.

Düzgün yanaşma — nəticə qaytaran stream metodlarından istifadə etməkdir:

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

Bütün kolleksiyalar eyni dərəcədə “paralelləşmir”

Bəzi kolleksiyalar (məsələn, adi ArrayList) hissələrə yaxşı bölünür. Amma LinkedList və ya sonsuz elementli axınlar (məsələn, Stream.generate(...)) üçün bunu demək olmur.

5. Təcrübə: performansın müqayisəsi

Nümunə: maksimal ədədin tapılması

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

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

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

Nə görəcəyik? Müasir çoxnüvəli prosessorlarda paralel stream adətən daha sürətlidir. Amma 30_000_000-ı 1000 ilə əvəz etsəniz, fərq olmayacaq — bəzən paralel hətta daha ləng işləyəcək!

6. İstifadə nümunələri: filtrasiya, aqreqasiya, çeşidləmə

Filtrasiya və sayma

List<String> names = Arrays.asList("Anya", "Boris", "Vasya", "Grisha", "Dasha", "Egor", "Zhenya");

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

System.out.println("Uzunluğu 4 olan adların sayı: " + count);

Qruplaşma

List<String> words = Arrays.asList("pişik", "balina", "pişik", "it", "balina", "pişik");

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

System.out.println(freq); // {it=1, balina=2, pişik=3}

Çeşidləmə (amma burada paralellik həmişə qazanc vermir!)

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("Paralel çeşidləmə: " + (t2 - t1) + " ms");

7. Vacib nüanslar və tövsiyələr

Nə zaman parallelStream()-dən istifadə etmək olar

• Kolleksiya böyükdürsə (on minlərlə element və daha çox).
• Element üzərində əməliyyat “ağır”dır (mürəkkəb hesablamalar, fayl/şəbəkə işi).
• Elementlərin sırasından asılılıq yoxdursa.
• Yan təsirlər yoxdursa (xarici dəyişənlər dəyişdirilmir).

Nə zaman parallelStream()-dən istifadə ETMƏK lazım deyil

• Kolleksiya kiçikdirsə.
• Əməliyyat sürətlidirsə.
• Sıra ciddi şəkildə qorunmalıdırsa.
• Ümumi dəyişənlərə giriş varsa (thread-safe kolleksiyaları və ya başqa yanaşmaları nəzərdən keçirin).

Neçə thread istifadə olunduğunu necə bilmək olar?

Standartda — prosessor nüvələrinin sayına görə: Runtime.getRuntime().availableProcessors(). Bu davranışı sistem xassəsi ilə dəyişmək olar:

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

Bunu yalnız nəticələrini başa düşürsünüzsə edin — əks halda prosessoru “boğa” və ləngimələr əldə edə bilərsiniz.

8. Paralel stream-lərlə işləyərkən tipik səhvlər

Səhv №1: forEach daxilində yan təsirlər
Çoxları düşünür: “İndi siyahını paralel dolduraram!”

List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
    .parallel()
    .forEach(result::add); // TEHLÜKƏLİ!
System.out.println(result.size()); // Nəticə — təsadüfidir!

Niyə bu pisdir? ArrayList thread-safe deyil və müxtəlif thread-lərdən eyni vaxtda əlavə edildikdə nəticə qeyri-müəyyəndir: buraxmalar, dublikatlar, istisnalar ola bilər.

Həll: Stream-in toplama metodlarından (collect) istifadə edin — onlar təhlükəsizliyi özləri təmin edir — və ya xüsusi kolleksiyalardan istifadə edin.

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

Səhv №2: Kiçik tapşırıqlarda sürətlənmə gözləmək
Paralellik — pulsuz deyil! Kolleksiya kiçikdirsə, planlaşdırma və sinxronlaşdırma əlavə xərclərinə görə paralel stream daha yavaş işləyə bilər.

Səhv №3: Sıranın pozulması
Elementlərin sırası sizin üçün vacibdirsə (məsələn, fayla yazarkən), paralel stream istifadə etməyin — sıra zəmanətli deyil (və ya daha yavaş işləyəcək).

Səhv №4: “Uğursuz” kolleksiyalardan istifadə
Bəzi kolleksiyalar (məsələn, LinkedList, qeyri-standart strukturlar) hissələrə pis bölünür — paralelliyin səmərəliliyi azalır.

Səhv №5: Nəticələrin toplanmasında thread-safety-ni görməməzlikdən gəlmək
Nəticələri əl ilə toplayırsınızsa (məsələn, siyahıya əlavə edirsinizsə), thread-safe kolleksiyalardan (CopyOnWriteArrayList, ConcurrentLinkedQueue) və ya toplama metodlarından istifadə edin.