Java Thread ve Concurrency Rehberi

byAli Turgut BOZKURT -
0

Java Thread ve Concurrency Detaylı Rehberi

1. Thread'ler

Thread'ler, bir programın aynı anda birden fazla işi yapabilmesini sağlayan hafif alt işlemlerdir. Java'da her uygulama en az bir thread (main thread) ile başlar.

Thread Oluşturma:

  1. Thread sınıfını extend etmek:
java
public class MyThread extends Thread {
    public void run() {
        System.out.println("Thread çalışıyor!");
    }
}

MyThread thread = new MyThread();
thread.start();
  1. Runnable interface'ini implement etmek:
java
public class MyRunnable implements Runnable {
    public void run() {
        System.out.println("Runnable çalışıyor!");
    }
}

Thread thread = new Thread(new MyRunnable());
thread.start();

Thread Yaşam Döngüsü:

  1. New: Thread oluşturuldu ama henüz başlatılmadı.
  2. Runnable: Thread çalışmaya hazır veya çalışıyor.
  3. Blocked: Thread geçici olarak çalışamıyor (örn. bir kaynağın kilidini bekliyor).
  4. Waiting: Thread başka bir thread'in bir işlemi tamamlamasını bekliyor.
  5. Timed Waiting: Thread belirli bir süre bekliyor.
  6. Terminated: Thread çalışmasını tamamladı.

Önemli Thread Metodları:

  • start(): Thread'i başlatır.
  • run(): Thread'in yapacağı işi tanımlar.
  • sleep(long millis): Thread'i belirli bir süre bekletir.
  • join(): Bir thread'in bitmesini bekler.
  • interrupt(): Thread'i kesintiye uğratır.

2. Synchronization

Synchronization, birden fazla thread'in paylaşılan kaynaklara erişimini kontrol etmek için kullanılır. Bu, veri tutarsızlığını ve race condition'ları önler.

synchronized Keyword:

java
public class Counter {
    private int count = 0;

    public synchronized void increment() {
        count++;
    }

    public synchronized int getCount() {
        return count;
    }
}

synchronized Blokları:

java
public void someMethod() {
    synchronized(this) {
        // Kritik bölge
    }
}

Açıklama:

  • synchronized metotlar veya bloklar, aynı anda sadece bir thread tarafından yürütülebilir.
  • Bir thread synchronized bloğa girdiğinde, nesnenin kilidini alır ve bloktan çıkana kadar tutar.
  • Bu, diğer thread'lerin aynı nesne üzerindeki synchronized metotlara veya bloklara erişimini engeller.

3. Thread Güvenliği

Thread güvenliği, çoklu thread ortamında veri bütünlüğünü korumak için kritik öneme sahiptir.

Race Condition:

Birden fazla thread'in aynı anda paylaşılan bir kaynağa erişmeye ve değiştirmeye çalışması durumudur.

Örnek:

java
public class UnsafeCounter {
    private int count = 0;

    public void increment() {
        count++; // Bu işlem atomik değil!
    }
}

Çözüm: synchronized keyword'ü veya AtomicInteger kullanmak.

Deadlock:

İki veya daha fazla thread'in birbirlerinin kilitledikleri kaynakları beklemesi durumudur.

Örnek:

java
public class DeadlockExample {
    private final Object lock1 = new Object();
    private final Object lock2 = new Object();

    public void method1() {
        synchronized(lock1) {
            synchronized(lock2) {
                // İşlem
            }
        }
    }

    public void method2() {
        synchronized(lock2) {
            synchronized(lock1) {
                // İşlem
            }
        }
    }
}

Çözüm: Kilitleri her zaman aynı sırada almak veya tryLock() metodunu kullanmak.

Livelock:

Thread'lerin birbirlerinin işlemlerini engellediği, ancak hiçbirinin ilerleyemediği durumdur.

Çözüm: Rastgele bekleme süreleri eklemek veya retry mekanizması kullanmak.

4. Executor Framework

Executor Framework, thread yönetimini kolaylaştırır ve thread havuzları oluşturmayı sağlar.

ExecutorService:

java
ExecutorService executor = Executors.newFixedThreadPool(5);
executor.submit(() -> {
    System.out.println("Executor ile çalışan görev");
});
executor.shutdown();

Callable ve Future:

java
Callable<Integer> task = () -> {
    TimeUnit.SECONDS.sleep(1);
    return 123;
};

Future<Integer> future = executor.submit(task);
System.out.println(future.get()); // Blocking call

Açıklama:

  • ExecutorService, görevleri yönetir ve thread havuzunu kontrol eder.
  • Callable, bir değer döndüren ve exception fırlatabilen görevler için kullanılır.
  • Future, asenkron hesaplamaların sonucunu temsil eder.

5. Concurrent Collections

Java, thread-safe koleksiyonlar sunar:

ConcurrentHashMap:

java
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.put("key", 1);
map.computeIfAbsent("key2", k -> k.length());

CopyOnWriteArrayList:

java
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("item");
for (String item : list) { // Güvenli iterasyon
    System.out.println(item);
}

BlockingQueue:

java
BlockingQueue<String> queue = new LinkedBlockingQueue<>();
queue.put("item");
String item = queue.take(); // Bloke olabilir

Açıklama:

  • ConcurrentHashMap: Birden çok thread'in aynı anda okuma yapmasına izin verir, yazma işlemlerini senkronize eder.
  • CopyOnWriteArrayList: Her yazma işleminde arrayin bir kopyasını oluşturur, okuma işlemleri için idealdir.
  • BlockingQueue: Üretici-tüketici senaryoları için idealdir, queue boş veya dolu olduğunda thread'leri bloke eder.

6. Lock ve Condition'lar

Lock interface'i, synchronized'dan daha esnek bir kilit mekanizması sağlar.

ReentrantLock:

java
Lock lock = new ReentrantLock();
lock.lock();
try {
    // Kritik bölge
} finally {
    lock.unlock();
}

ReadWriteLock:

java
ReadWriteLock rwLock = new ReentrantReadWriteLock();
Lock readLock = rwLock.readLock();
Lock writeLock = rwLock.writeLock();

// Okuma işlemi
readLock.lock();
try {
    // Okuma işlemleri
} finally {
    readLock.unlock();
}

// Yazma işlemi
writeLock.lock();
try {
    // Yazma işlemleri
} finally {
    writeLock.unlock();
}

Condition:

java
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();

lock.lock();
try {
    while (!condition) {
        condition.await();
    }
    // İşlem yap
} finally {
    lock.unlock();
}

// Başka bir thread'de
lock.lock();
try {
    condition.signal();
} finally {
    lock.unlock();
}

Açıklama:

  • ReentrantLock: Aynı thread'in kilidi birden fazla kez almasına izin verir.
  • ReadWriteLock: Okuma işlemleri için paylaşımlı, yazma işlemleri için özel kilit sağlar.
  • Condition: Thread'lerin belirli koşulları beklemesini ve sinyalleşmesini sağlar.

7. Atomic Değişkenler

Atomic değişkenler, thread-safe ve lock-free operasyonlar sağlar.

AtomicInteger:

java
AtomicInteger counter = new AtomicInteger(0);
int updatedValue = counter.incrementAndGet();
boolean wasSet = counter.compareAndSet(updatedValue, 10);

AtomicReference:

java
AtomicReference<String> ref = new AtomicReference<>("initial");
ref.set("new value");
String oldValue = ref.getAndSet("newer value");

LongAdder:

java
LongAdder adder = new LongAdder();
adder.add(10);
adder.increment();
long sum = adder.sum();

Açıklama:

  • Atomic değişkenler, çoklu thread ortamında race condition'ları önler.
  • incrementAndGet(), compareAndSet() gibi metodlar atomik olarak çalışır.
  • LongAdder, yüksek eşzamanlılık durumlarında AtomicLong'dan daha iyi performans gösterir.

8. Thread Pools

Thread havuzları, thread oluşturma ve yönetme maliyetini azaltır.

FixedThreadPool:

java
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
    executor.submit(() -> {
        System.out.println("Task executed by " + Thread.currentThread().getName());
    });
}
executor.shutdown();

CachedThreadPool:

java
ExecutorService executor = Executors.newCachedThreadPool();
// Kullanım FixedThreadPool ile aynı

ScheduledThreadPool:

java
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
    System.out.println("Scheduled task");
}, 0, 1, TimeUnit.SECONDS);

Açıklama:

  • FixedThreadPool: Sabit sayıda thread içerir.
  • CachedThreadPool: İhtiyaca göre yeni thread'ler oluşturur ve kullanılmayan thread'leri sonlandırır.
  • ScheduledThreadPool: Görevlerin belirli bir gecikme ile veya periyodik olarak çalıştırılmasını sağlar.

9. CompletableFuture

CompletableFuture, asenkron işlemleri zincirlemek ve birleştirmek için kullanılır.

Asenkron İşlem:

java
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // Uzun süren bir işlem
    return "Result";
});

future.thenAccept(System.out::println);

Zincirleme İşlemler:

java
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")
    .thenApply(String::toUpperCase);

System.out.println(future.get()); // HELLO WORLD

Birleştirme:

java
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");

CompletableFuture<String> combined = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2);

System.out.println(combined.get()); // Hello World

Açıklama:

  • supplyAsync(): Asenkron olarak bir değer döndüren işlem başlatır.
  • thenApply(): Bir önceki aşamanın sonucunu alıp işler ve yeni bir sonuç döndürür.
  • thenAccept(): Bir önceki aşamanın sonucunu alıp işler, ama yeni bir sonuç döndürmez.
  • thenCombine(): İki CompletableFuture'ın sonuçlarını birleştirir.

10. Fork/Join Framework

Fork/Join Framework, "böl ve yönet" stratejisini kullanarak büyük görevleri daha küçük alt görevlere böler ve paralel olarak işler.

RecursiveTask:

java
public class FibonacciTask extends RecursiveTask<Integer> {
    final int n;

    FibonacciTask(int n) { this.n = n; }

    @Override
    protected Integer compute() {
        if (n <= 1) return n;
        FibonacciTask f1 = new FibonacciTask(n - 1);
        f1.fork();
        FibonacciTask f2 = new FibonacciTask(n - 2);
        return f2.compute() + f1.join();
    }
}

ForkJoinPool pool = new ForkJoinPool();
System.out.println(pool.invoke(new FibonacciTask(10)));

RecursiveAction:

java
public class SortTask extends RecursiveAction {
    final int[] array;
    final int low, high;

    SortTask(int[] array, int low, int high) {
        this.array = array;
        this.low = low;
        this.high = high;
    }

    @Override
    protected void compute() {
        if (high - low < THRESHOLD) {
            Arrays.sort(array, low, high);
        } else {
            int mid = (low + high) >>> 1;
            invokeAll(new SortTask(array, low, mid),
                      new SortTask(array, mid, high));
            merge(array, low, mid, high);
        }
    }
}

Açıklama:

  • RecursiveTask: Bir değer döndüren görevler için kullanılır.
  • RecursiveAction: Bir değer döndürmeyen görevler için kullanılır.
  • fork(): Alt görevi asenkron olarak başlatır.
  • join(): Alt görevin sonucunu bekler ve alır.
  • invokeAll(): Birden fazla alt görevi paralel olarak çalıştırır.

11. Phaser

Phaser, CountDownLatch ve CyclicBarrier'ın özelliklerini birleştiren esnek bir senkronizasyon bariyeridir.

java
public class PhaserExample {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // "1" is for self

        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            phaser.register();
            new Thread(() -> {
                System.out.println("Thread " + threadId + " arriving");
                phaser.arriveAndAwaitAdvance();
                System.out.println("Thread " + threadId + " after passing barrier");
                phaser.arriveAndDeregister();
            }).start();
        }

        phaser.arriveAndDeregister();
    }
}

Açıklama:

  • register(): Yeni bir katılımcı ekler.
  • arriveAndAwaitAdvance(): Fazın tamamlandığını bildirir ve diğer katılımcıları bekler.
  • arriveAndDeregister(): Fazı tamamlar ve Phaser'dan ayrılır.
  • Phaser, dinamik sayıda thread'in senkronize edilmesine olanak tanır.

12. Semaphore

Semaphore, sınırlı sayıda kaynağa erişimi kontrol etmek için kullanılır.

java
public class SemaphoreExample {
    private static final int MAX_CONNECTIONS = 3;
    private static final Semaphore semaphore = new Semaphore(MAX_CONNECTIONS);

    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + " acquired a connection");
                    Thread.sleep(2000); // Simulating work
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    semaphore.release();
                    System.out.println(Thread.currentThread().getName() + " released a connection");
                }
            }).start();
        }
    }
}

Açıklama:

  • acquire(): Bir izin alır, eğer izin yoksa thread bloklanır.
  • release(): Bir izni serbest bırakır.
  • Semaphore, örneğin veritabanı bağlantı havuzları gibi sınırlı kaynaklara erişimi kontrol etmek için kullanılabilir.

13. CountDownLatch

CountDownLatch, bir veya daha fazla thread'in belirli sayıda olayın gerçekleşmesini beklemesini sağlar.

java
public class CountDownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);

        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getName() + " completed");
                    latch.countDown();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }).start();
        }

        latch.await();
        System.out.println("All threads completed");
    }
}

Açıklama:

  • countDown(): Sayacı bir azaltır.
  • await(): Sayaç sıfıra ulaşana kadar bekler.
  • CountDownLatch, bir kez kullanılabilir. Sıfıra ulaştığında tekrar kullanılamaz.

14. CyclicBarrier

CyclicBarrier, bir grup thread'in belirli bir noktada buluşmasını sağlar.

java
public class CyclicBarrierExample {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("All threads reached the barrier"));

        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " waiting at barrier");
                    barrier.await();
                    System.out.println(Thread.currentThread().getName() + " crossed the barrier");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

Açıklama:

  • await(): Thread'i bariyerde bekletir.
  • Tüm thread'ler bariyere ulaştığında, isteğe bağlı bir eylem gerçekleştirilebilir.
  • CyclicBarrier, adından da anlaşılacağı gibi tekrar kullanılabilir.

15. ThreadLocal

ThreadLocal, her thread için ayrı bir değer saklamanızı sağlar.

java
public class ThreadLocalExample {
    private static final ThreadLocal<Integer> threadLocal = ThreadLocal.withInitial(() -> 0);

    public static void main(String[] args) {
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                threadLocal.set(threadLocal.get() + 1);
                System.out.println(Thread.currentThread().getName() + ": " + threadLocal.get());
            }).start();
        }
    }
}

Açıklama:

  • Her thread kendi değerini görür ve değiştirir.
  • Thread-local değerler, diğer thread'ler tarafından görülmez.
  • Genellikle, her thread için ayrı bir bağlam (context) saklamak için kullanılır.

16. Volatile Keyword

volatile keyword'ü, bir değişkenin her zaman main memory'den okunup yazılmasını sağlar, bu da thread'ler arası görünürlüğü garanti eder.

java
public class VolatileExample {
    private static volatile boolean flag = false;

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            while (!flag) {
                // Do nothing, just wait
            }
            System.out.println("Flag is true, exiting");
        }).start();

        Thread.sleep(1000);
        flag = true;
    }
}

Açıklama:

  • volatile değişkenler, her okuma/yazma işleminde doğrudan main memory ile etkileşime girer.
  • Cache tutarsızlığı problemlerini önler.
  • Ancak, atomik operasyonları garanti etmez. Örneğin, i++ gibi işlemler için yeterli değildir.

17. Java Memory Model ve Happens-Before İlişkisi

Java Memory Model (JMM), thread'lerin nasıl etkileşime girdiğini ve memory operasyonlarının nasıl sıralandığını tanımlar.

Happens-Before İlişkisi:

  1. Program Order Rule: Bir thread içindeki her eylem, programdaki sıralamaya göre önceki eylemlerden sonra gerçekleşir.
  2. Monitor Lock Rule: Bir monitör kilidinin serbest bırakılması, aynı kilidin sonraki edinilmesinden önce gerçekleşir.
  3. Volatile Variable Rule: Bir volatile değişkene yazma, aynı değişkenin sonraki okumalarından önce gerçekleşir.
  4. Thread Start Rule: Bir thread'de thread.start() çağrısı, başlatılan thread'deki herhangi bir eylemden önce gerçekleşir.
  5. Thread Termination Rule: Bir thread'deki tüm eylemler, başka bir thread'in bu thread üzerinde thread.join() dönüşünden önce gerçekleşir.

Örnek:

java
class HappensBefore {
    int a = 0;
    volatile boolean flag = false;

    public void writer() {
        a = 1;                 // 1
        flag = true;           // 2
    }

    public void reader() {
        if (flag) {            // 3
            int i = a;         // 4
        }
    }
}

Bu örnekte, eğer reader thread'i flag'i true olarak okursa (3), writer thread'inin a'ya 1 değerini ataması (1) görünür olacaktır. Bu, volatile write'ın (2) sonraki volatile read'den (3) önce gerçekleşmesi ve 1'in 2'den önce, 3'ün de 4'ten önce gerçekleşmesi (program order rule) nedeniyledir.

18. Concurrency Design Patterns

Concurrency design pattern'ları, çok thread'li uygulamalarda yaygın problemleri çözmek için kullanılan tekniklerdir.

1. Producer-Consumer Pattern:

java
public class ProducerConsumer {
    private static final int BUFFER_SIZE = 5;
    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(BUFFER_SIZE);

    static class Producer implements Runnable {
        public void run() {
            try {
                for (int i = 0; i < 20; i++) {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    static class Consumer implements Runnable {
        public void run() {
            try {
                while (true) {
                    Integer item = queue.take();
                    System.out.println("Consumed: " + item);
                    Thread.sleep(300);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
    }
}

2. Read-Write Lock Pattern:

java
public class ReadWriteLockExample {
    private static final ReadWriteLock lock = new ReentrantReadWriteLock();
    private static final Lock readLock = lock.readLock();
    private static final Lock writeLock = lock.writeLock();
    private static int value = 0;

    public static int read() {
        readLock.lock();
        try {
            return value;
        } finally {
            readLock.unlock();
        }
    }

    public static void write(int newValue) {
        writeLock.lock();
        try {
            value = newValue;
        } finally {
            writeLock.unlock();
        }
    }
}

Açıklama:

  • Producer-Consumer Pattern: Üreticiler ve tüketiciler arasında veri akışını yönetir.
  • Read-Write Lock Pattern: Okuma işlemlerinin paralel, yazma işlemlerinin özel olmasını sağlar.

Bu ileri düzey konular ve örnekler, Java'da concurrency ve multi-threading konularında daha derin bir anlayış kazanmanıza yardımcı olacaktır. Bu kavramları pratikte uygulayarak ve farklı senaryolarda deneyerek, zamanla bir uzman haline geleceksiniz.

Tags

Bu yayınları beğenebilirsiniz

Please Select Embedded Mode To Show The Comment System.*

Daha yeni Daha eski

نموذج الاتصال