Java-da çoxaxınlılıq və paralel iş

Bu dərsdə öyrənəcəksən

Thread yaratmaq, start() ilə işə salmaq və join() ilə gözləmək
Yarış vəziyyətini tanımaq və onu synchronized və ya AtomicInteger ilə aradan qaldırmaq
Tapşırıqları ExecutorService, CompletableFuture və virtual thread-lərlə icra etmək

Tək kassası olan mağazada növbə uzanır, beş kassa açılanda isə müştərilərə tez xidmət göstərilir. Kompüterin prosessorunda da bir neçə nüvə var və server eyni anda minlərlə sorğuya cavab verməlidir. Bir neçə işi eyni vaxtda görmək bacarığına çoxaxınlılıq (concurrency) deyilir. Java bu sahədə ən güclü dillərdəndir: sadə thread-lərdən tutmuş Java 21-in virtual thread-lərinə qədər.

Tərif

Thread (icra axını)

Proqramın daxilində müstəqil icra xətti. Hər Java proqramı main adlı bir thread ilə başlayır; yeni thread-lər yaradıb işi onlar arasında bölmək olar. Thread-lər eyni yaddaşı — eyni obyektləri — paylaşır: bu həm rahatdır, həm də təhlükəlidir.

Thread yaratmaq: start və join

Görüləcək işi new Thread(...) konstruktoruna lambda kimi veririk. start() yeni thread-i işə salır və dərhal geri qayıdır — main öz işini davam etdirir. join() isə həmin thread bitənə qədər gözləyir. Aşağıda 1-dən 1 000 000-a qədər ədədlərin cəmi iki yarıya bölünüb və hər yarını ayrı thread hesablayır; nəticələri yalnız join()-dən sonra oxuyuruq.

Java

public class Main {
    public static void main(String[] args) throws InterruptedException {
        long[] sums = new long[2];
        Thread first = new Thread(() -> sums[0] = sum(1, 500_000));
        Thread second = new Thread(() -> sums[1] = sum(500_001, 1_000_000));
        first.start();
        second.start();
        first.join();
        second.join();
        System.out.println("Part 1: " + sums[0]);
        System.out.println("Part 2: " + sums[1]);
        System.out.println("Total:  " + (sums[0] + sums[1]));
        System.out.println("Printed by: " + Thread.currentThread().getName());
    }

    static long sum(long from, long to) {
        long total = 0;
        for (long i = from; i <= to; i++) total += i;
        return total;
    }
}

Gözlənilən nəticə

Part 1: 125000250000
Part 2: 375000250000
Total:  500000500000
Printed by: main

İki thread işi bölüşür, main isə onları join() ilə gözləyir

Səhv: `run()` adi metod çağırışıdır

Thread t = new Thread(() -> System.out.println(Thread.currentThread().getName()));
t.run();     // prints main: no new thread was created!

Düzgün: `start()` yeni thread açır

Thread t = new Thread(() -> System.out.println(Thread.currentThread().getName()));
t.start();   // prints Thread-0: the code runs in a new thread
t.join();

Ortaq verilənlər: yarış vəziyyəti

count++ bir əməliyyat kimi görünür, əslində isə üç addımdır: qiyməti oxu, 1 əlavə et, geri yaz. İki thread eyni anda oxusa, hər ikisi eyni qiyməti yazacaq və bir artırma itəcək. Bu səhvə yarış vəziyyəti (race condition) deyilir: nəticə thread-lərin təsadüfi növbəsindən asılıdır. Sınaqda 100 000-lik dövr işlədən iki thread 200 000 əvəzinə, məsələn, 121 962 verdi — hər işə salmada başqa ədəd çıxır.

Yarış vəziyyəti

static int count = 0;

// each of two threads runs:
for (int i = 0; i < 100_000; i++) {
    count++;                    // read + add + write: not atomic
}
// result: often less than 200000

Atomik sayğac

static final AtomicInteger count = new AtomicInteger();

// each of two threads runs:
for (int i = 0; i < 100_000; i++) {
    count.incrementAndGet();    // one indivisible operation
}
// result: always 200000

Problemi həll etməyin iki əsas yolu var. synchronized açar sözü metodu kilidləyir: eyni anda onun içində yalnız bir thread ola bilər, digərləri növbə gözləyir. java.util.concurrent.atomic paketindəki AtomicInteger, AtomicLong kimi siniflər isə artırmanı kilidsiz, bölünməz bir prosessor əməliyyatı ilə edir və adətən daha sürətlidir. Kolleksiyalar üçün hazır təhlükəsiz variantlar var: ConcurrentHashMap, CopyOnWriteArrayList.

Java

import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    static int syncCount = 0;
    static final AtomicInteger atomicCount = new AtomicInteger();

    static synchronized void incrementSync() {
        syncCount++;
    }

    public static void main(String[] args) throws InterruptedException {
        Runnable task = () -> {
            for (int i = 0; i < 100_000; i++) {
                incrementSync();
                atomicCount.incrementAndGet();
            }
        };
        Thread a = new Thread(task);
        Thread b = new Thread(task);
        a.start();
        b.start();
        a.join();
        b.join();
        System.out.println("synchronized: " + syncCount);
        System.out.println("AtomicInteger: " + atomicCount.get());
    }
}

Gözlənilən nəticə

synchronized: 200000
AtomicInteger: 200000

Hər iki üsul həmişə 200 000 verir

ExecutorService: thread hovuzu

Hər tapşırıq üçün yeni thread yaratmaq bahalıdır. ExecutorService hazır thread-lərdən ibarət hovuz (pool) saxlayır və tapşırıqları onlara paylayır. submit bir tapşırıq verir və Future qaytarır — gələcək nəticənin «qəbzi»; invokeAll isə tapşırıqlar siyahısını verir və hamısını gözləyir. Future.get() nəticə hazır olana qədər gözləyir. Java 19-dan bəri ExecutorService-i try-with-resources ilə bağlamaq olar: blok bitəndə bütün tapşırıqlar gözlənilir və hovuz söndürülür.

Java

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main {
    public static void main(String[] args) throws Exception {
        List<String> classes = List.of("10A", "10B", "11A");
        long start = System.currentTimeMillis();
        try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
            List<Callable<String>> jobs = classes.stream()
                    .map(c -> (Callable<String>) () -> buildReport(c))
                    .toList();
            for (Future<String> f : pool.invokeAll(jobs)) {
                System.out.println(f.get());
            }
        }
        long ms = System.currentTimeMillis() - start;
        System.out.println("Parallel was faster: " + (ms < 900));
    }

    static String buildReport(String name) throws InterruptedException {
        Thread.sleep(300);   // pretend to read the database
        return "Report for " + name + " is ready";
    }
}

Gözlənilən nəticə

Report for 10A is ready
Report for 10B is ready
Report for 11A is ready
Parallel was faster: true

Hər hesabat 300 ms çəkir, amma üçü birlikdə təxminən 300 ms-ə hazır olur

CompletableFuture: asinxron zəncirlər

Future.get() thread-i bloklayır. CompletableFuture isə nəticə gələndə nə ediləcəyini əvvəlcədən təsvir etməyə imkan verir: supplyAsync işi fonda başladır, thenApply nəticəni çevirir, thenCombine iki müstəqil nəticəni birləşdirir, exceptionally isə xəta baş verərsə ehtiyat qiymət verir. Sonda join() yekun nəticəni götürür. Bu, JavaScript-dəki Promise-ə çox bənzəyir.

Java

import java.util.concurrent.CompletableFuture;

public class Main {
    public static void main(String[] args) {
        CompletableFuture<Integer> exam = CompletableFuture.supplyAsync(() -> slow(92, 300));
        CompletableFuture<Integer> bonus = CompletableFuture.supplyAsync(() -> slow(5, 200));

        CompletableFuture<String> report = exam
                .thenCombine(bonus, Integer::sum)
                .thenApply(total -> Math.min(total, 100))
                .thenApply(total -> "Final score: " + total);
        System.out.println(report.join());

        CompletableFuture<Integer> broken = CompletableFuture
                .supplyAsync(() -> Integer.parseInt("ninety"))
                .exceptionally(ex -> -1);
        System.out.println("With fallback: " + broken.join());
    }

    static int slow(int value, long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return value;
    }
}

Gözlənilən nəticə

Final score: 97
With fallback: -1

İmtahan balı və bonus paralel hesablanır; ninety ədədə çevrilmədiyi üçün ehtiyat qiymət -1 işə düşür

Virtual thread-lər (Java 21)

Adi (platforma) thread əməliyyat sisteminin thread-inə bağlıdır və stek üçün adətən təxminən 1 MB yaddaş ayırır, ona görə minlərlə belə thread yaratmaq çətindir. Java 21-də gələn virtual thread-lər isə JVM-in idarə etdiyi yüngül thread-lərdir: gözləyərkən (şəbəkə, fayl, verilənlər bazası) onlar altdakı thread-i boşaldır və milyonlarla yaradıla bilər. Aşağıda 10 000 tapşırığın hər biri 1 saniyə «gözləyir». 100 thread-lik adi hovuzda bu, təxminən 100 saniyə çəkərdi, virtual thread-lərlə isə bir-iki saniyə çəkir.

Java

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    public static void main(String[] args) {
        AtomicInteger done = new AtomicInteger();
        Instant start = Instant.now();
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 10_000; i++) {
                executor.submit(() -> {
                    Thread.sleep(Duration.ofSeconds(1));   // a slow network call
                    done.incrementAndGet();
                    return null;
                });
            }
        }   // close() waits until every task has finished
        Duration took = Duration.between(start, Instant.now());
        System.out.println("Finished tasks: " + done.get());
        System.out.println("Took less than 5 s: " + (took.toSeconds() < 5));

        Thread vt = Thread.ofVirtual().start(() -> {});
        System.out.println("Is virtual: " + vt.isVirtual());
    }
}

Gözlənilən nəticə

Finished tasks: 10000
Took less than 5 s: true
Is virtual: true

Hər tapşırıq üçün ayrıca virtual thread: newVirtualThreadPerTaskExecutor()

Virtual thread-lə yazılan kod adi, ardıcıl koda bənzəyir: Thread.sleep, fayl oxuma və verilənlər bazası sorğusu olduğu kimi yazılır, callback və zəncirlərə ehtiyac yoxdur. Tək thread-i Thread.ofVirtual().start(...) və ya Thread.startVirtualThread(...) ilə də başlatmaq olar. Veb-serverlər də bundan faydalanır: məsələn, Spring Boot 3.2-dən bəri spring.threads.virtual.enabled=true parametri hər HTTP sorğusunu virtual thread-də emal edir.

Vəziyyət	Alət
Çoxlu gözləmə: HTTP sorğuları, verilənlər bazası, fayllar	`newVirtualThreadPerTaskExecutor()`
Ağır hesablamalar (CPU)	`newFixedThreadPool(cores)`
Nəticələri zəncirləmək və birləşdirmək	`CompletableFuture`
Ortaq sayğac və ya xəritə	`AtomicInteger`, `ConcurrentHashMap`

Əsas fikirlər

start() yeni thread açır, join() onu gözləyir; run() isə yeni thread yaratmır.
Ortaq dəyişən verilənlər yarış vəziyyəti yaradır; həlli synchronized, atomik tiplər və ConcurrentHashMap-dir.
ExecutorService thread-ləri təkrar istifadə edir; submit və invokeAll Future qaytarır.
CompletableFuture zənciri: supplyAsync işi başladır, thenApply / thenCombine nəticələri çevirir və birləşdirir, exceptionally xətanı tutur, join yekunu götürür.
Java 21-in virtual thread-ləri gözləmə ilə dolu işlər üçündür; hesablama üçün nüvə sayı qədər thread-li hovuz seç.

Özünü yoxla

10 sual. Hər düzgün cavab XP qazandırır.

1 / 10

t.start() əvəzinə t.run() çağırsan, nə baş verər?