Yürütücü Servisleri

Önceki ders eşzamanlılık yardımcılarının işi çağırana ne kadar yükleyip ne kadar başka bir tarafa devrettiğini ölçtü: AtomicInteger’da yeniden deneme çağıranda kalıyordu, CountDownLatch ve Semaphore’da bekleyen engelleniyor, sayacı başka bir taraf düşürüyordu — ama görevin hangi iş parçacığında koşacağına hâlâ çağıranın kendisi karar veriyordu, run() ile mi start() ile mi çağıracağını seçerek. Bu ders o seçimi bir servise devrediyor. Yürütücü servisi bir görev alır ve onu kimin koşturacağına kendisi karar verir; soru aynıdır — bu adımı hangi iş parçacığı koşturuyor — ama yanıt artık çağrı satırında değil, yürütücünün o anki durumundadır.

Kursun en sert örneği burada duruyor, çünkü iki durumun kodu birebir aynıdır. Havuz boşken ve doluyken aynı execute çağrısı yazılır; farkı yapan tek şey çağrı yerinde hiç görünmeyen bir şey — havuzun o anki doluluğu.

Bir yürütücü servisi üç parçadan kurulur: sınırlı sayıda iş parçacığı, bekleyen görevleri tutan bir kuyruk ve kuyruk da dolduğunda devreye giren bir reddetme yordamı. Önceki derste run() ile start() arasındaki fark tek bir yöntem adıydı; burada fark daha da küçülüyor — aynı yöntem adı, aynı imza, aynı çağrı satırı, ama iki ayrı sonuç. Bu üç bölüm sırayla o üç parçayı ölçüyor: kaç ayrı iş parçacığı gerçekten koşuyor, doluyken görev kime düşüyor ve doluluk anını hangi taraf yönetiyor.

Görevi Kim Koşturuyor: Ayrık Kimlik Sayısı

ExecutorService, bir görev listesi alan ve her görevi hangi iş parçacığının koşturacağına kendisi karar veren bir soyutlamadır. Çağıran görevi bir Callable ya da Runnable olarak yazar; kimin koşturacağını yürütücünün kendi kuralı belirler. Bu kuralı sayıyla görmek için ayrık iş parçacığı kimliği ölçülüyor: her görev koştuğu iş parçacığını paylaşılan bir kümeye ekliyor.

EZ16. Ayrık iş parçacığı kimliği bir kümeyle sayılır; kümeye eklenen her öge yalnızca nesne kimliğiyle (== eşdeğeri) ayırt edilir, ad ya da numarası hiç basılmaz. EZ17. Sabit boyutlu bir yürütücü, çekirdek iş parçacığı sayısına ulaşana kadar var olan boşta bir iş parçacığını kullanmak yerine yeni bir iş parçacığı oluşturur; bu kural çizelgeleyiciye değil yürütücünün kendisine aittir. EZ18. Çekirdek ile en büyük iş parçacığı sayısı eşit verildiğinde havuz bu sayıyı hiçbir zaman aşmaz; kuyruğa giren görevler var olan iş parçacıklarınca sırayla işlenir.

// Kimlikler.java — ayrik is parcacigi kimligini == esdegeriyle sayan kume (yardimci kaynak, main yok)
class Kimlikler {
    static java.util.Set<Thread> yeniKume() {
        return java.util.Collections.synchronizedSet(
                java.util.Collections.newSetFromMap(new java.util.IdentityHashMap<>()));
    }
}

// AyrikKimlikSayisi.java — sabit boyutlu bir havuzda gorev sayisi buyuse de ayrik kimlik sayisi sabit kalir
import java.util.*;
import java.util.concurrent.*;

public class AyrikKimlikSayisi {
    static int ayrikKimlik(ExecutorService yurutucu, int gorevSayisi) throws Exception {
        Set<Thread> gorulen = Kimlikler.yeniKume();
        List<Callable<Void>> gorevler = new ArrayList<>();
        for (int i = 0; i < gorevSayisi; i++) {
            gorevler.add(() -> { gorulen.add(Thread.currentThread()); return null; });
        }
        yurutucu.invokeAll(gorevler);
        return gorulen.size();
    }

    public static void main(String[] args) throws Exception {
        try (ExecutorService dortluHavuz1 = Executors.newFixedThreadPool(4)) {
            System.out.println("4 gorev, sabit havuz(4)  -> ayrik kimlik: " + ayrikKimlik(dortluHavuz1, 4));
        }
        try (ExecutorService dortluHavuz2 = Executors.newFixedThreadPool(4)) {
            System.out.println("8 gorev, sabit havuz(4)  -> ayrik kimlik: " + ayrikKimlik(dortluHavuz2, 8));
        }
        try (ExecutorService birliHavuz = Executors.newFixedThreadPool(1)) {
            System.out.println("4 gorev, sabit havuz(1)  -> ayrik kimlik: " + ayrikKimlik(birliHavuz, 4));
        }
    }
}

4 gorev, sabit havuz(4)  -> ayrik kimlik: 4
8 gorev, sabit havuz(4)  -> ayrik kimlik: 4
4 gorev, sabit havuz(1)  -> ayrik kimlik: 1

Ölçüm invokeAll yöntemini kullanıyor, tek tek submit çağırıp elde kalan gelecekleri beklemek yerine — çünkü invokeAll verilen bütün görevler bitmeden dönmüyor. Bu, sırayı çizelgeleyiciye bırakmadan üç kurulumu da güvenle karşılaştırmanın yolu: kümenin son boyutu okunduğunda, o kurulumdaki bütün görevlerin gerçekten koştuğu ve kendi kimliğini kümeye eklediği kesin.

try bloğunun parantezindeki ExecutorService, Standart Kütüphane ve Akışlar kursunda kurulan kaynakla deneme kalıbına AutoCloseable gerçekleştirimiyle katılır; blok bitince kapanış çağrısı burada da örtük ve otomatiktir. Üç satır aynı soruyu üç ayrı kurulumda soruyor: dört görev dört büyüklüğündeki havuzda dört ayrı kimlik üretiyor — her görev kendi iş parçacığını buluyor (EZ17). Sekiz görev aynı büyüklükteki havuzda yine dört ayrı kimlik üretiyor; kalan dört görev yeni bir iş parçacığı doğurmuyor, var olan dördünden birine düşüyor (EZ18). Tek iş parçacıklı havuzda dört görev tek bir kimliğe düşüyor — havuz büyüklüğü kaç görev geldiğinden bağımsız bir üst sınırdır. Sayı sabit kalıyor çünkü kural çizelgeleyiciye değil yürütücünün kendi kurucusuna aittir: newFixedThreadPool(4) çekirdek ile en büyük iş parçacığı sayısını eşit ve 4 vererek bu sınırı koyuyor.

Havuz Doluyken Görevi Veren Taraf Koşturur

Aynı soruyu havuz doluyken sormak için sınırlı bir kuyruk ve tek iş parçacıklı bir havuz kuruluyor. İlk görev bir CountDownLatch üzerinde bilinçli olarak bekletiliyor; bu, havuzun tek iş parçacığını meşgul tutmanın ve kuyruğun doluluğunu koşturarak kurgulamanın yolu — sıra çizelgeleyiciye bırakılmıyor.

// HavuzDolulugu.java — ayni execute cagrisi, farki havuzun o anki doluluğu belirliyor
import java.util.concurrent.*;

public class HavuzDolulugu {
    static String kim(Thread cagiran) {
        return Thread.currentThread() == cagiran ? "cagiran" : "olusturulan";
    }

    public static void main(String[] args) throws Exception {
        Thread cagiran = Thread.currentThread();
        CountDownLatch tut = new CountDownLatch(1);
        ThreadPoolExecutor havuz = new ThreadPoolExecutor(
                1, 1, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1),
                gorev -> new Thread(gorev),
                new ThreadPoolExecutor.CallerRunsPolicy());

        havuz.execute(() -> { try { tut.await(); } catch (InterruptedException e) { } });
        havuz.execute(() -> System.out.println("havuza verilen gorev      -> " + kim(cagiran)));
        havuz.execute(() -> System.out.println("dolu havuzda reddedilen   -> " + kim(cagiran)));

        tut.countDown();
        havuz.shutdown();
        havuz.awaitTermination(10, TimeUnit.SECONDS);
    }
}

dolu havuzda reddedilen   -> cagiran
havuza verilen gorev      -> olusturulan

Üç execute çağrısı sözdiziminde birbirinin aynısı. Birinci çağrı havuzun tek iş parçacığını tut.await() ile meşgul ediyor. İkinci çağrı geldiğinde iş parçacığı hâlâ meşgul ama kuyrukta bir yer boş — görev kuyruğa giriyor ve daha sonra, tut.countDown() çağrıldıktan sonra havuzun kendi iş parçacığında koşuyor: oluşturulan. Üçüncü çağrı geldiğinde hem iş parçacığı meşgul hem kuyruk dolu; yürütücünün genişleyecek yeri yok ve görev reddetme yordamına düşüyor. Yordam CallerRunsPolicy olduğu için reddedilen görev execute çağrısını yazan iş parçacığında, hemen orada, senkron olarak koşuyor — çıktıda bu satır önce görünüyor, çünkü üçüncü görev tut.countDown() çağrılmadan, ikinci görev henüz kuyruktan çıkmadan koşturuluyor. Kodda hiçbir dallanma yok; farkı yaratan tek şey çağrı anındaki doluluk.

Reddetme Yordamı Çağıranın Seçtiği Bir Parametredir

CallerRunsPolicy reddedilen görevi sessizce çağırana yüklüyor; bu, dört hazır seçenekten yalnızca biri. Standart kütüphane üç yordamı daha tanımlıyor: biri reddedilen görevi hiç koşturmadan sessizce düşürüyor, biri kuyruktaki en eski bekleyen görevi atıp yeni geleni kuyruğa alıyor, biri de burada gösterilen gibi bir istisna fırlatıyor. Dördü de aynı anı — iş parçacığı meşgul, kuyruk dolu — farklı bir kararla yanıtlıyor; hangisinin seçileceği görevin türünden değil, yürütücüyü kuran tarafın o görevi kaybetmeyi göze alıp alamayacağından geliyor.

EZ19. Sınırlı kuyruk dolduğunda ve havuz üst sınırındayken gelen görev reddetme yordamına düşer; CountDownLatch ile ilk görev bilinçli olarak bekletilerek kuyruğun ve havuzun doluluğu koşturarak kurgulanır. EZ20. Reddetme yordamı yürütücüyü kuran tarafın seçtiği bir parametredir, görev tarafından belirlenmez; aynı doluluk anında iki ayrı yordam iki ayrı sonuç verir.

// ReddetmeYordami.java — ayni dolu havuz aninda iki reddetme yordami iki ayri sonuc veriyor
import java.util.concurrent.*;

public class ReddetmeYordami {
    static String kim(Thread cagiran) {
        return Thread.currentThread() == cagiran ? "cagiran" : "olusturulan";
    }

    static ThreadPoolExecutor doluHavuz(RejectedExecutionHandler yordam, CountDownLatch tut) {
        ThreadPoolExecutor havuz = new ThreadPoolExecutor(
                1, 1, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1),
                gorev -> new Thread(gorev),
                yordam);
        havuz.execute(() -> { try { tut.await(); } catch (InterruptedException e) { } });
        havuz.execute(() -> { });
        return havuz;
    }

    public static void main(String[] args) throws Exception {
        Thread cagiran = Thread.currentThread();

        CountDownLatch tut1 = new CountDownLatch(1);
        ThreadPoolExecutor caginirsaCalisir = doluHavuz(new ThreadPoolExecutor.CallerRunsPolicy(), tut1);
        try {
            caginirsaCalisir.execute(() -> System.out.println("CallerRunsPolicy  -> " + kim(cagiran)));
        } finally {
            tut1.countDown();
            caginirsaCalisir.shutdown();
            caginirsaCalisir.awaitTermination(10, TimeUnit.SECONDS);
        }

        CountDownLatch tut2 = new CountDownLatch(1);
        ThreadPoolExecutor keserAtar = doluHavuz(new ThreadPoolExecutor.AbortPolicy(), tut2);
        try {
            keserAtar.execute(() -> System.out.println("bu satir hic basilmiyor"));
            System.out.println("AbortPolicy       -> dusmedi");
        } catch (RejectedExecutionException e) {
            System.out.println("AbortPolicy       -> " + e.getClass().getSimpleName());
        } finally {
            tut2.countDown();
            keserAtar.shutdown();
            keserAtar.awaitTermination(10, TimeUnit.SECONDS);
        }
    }
}

CallerRunsPolicy  -> cagiran
AbortPolicy       -> RejectedExecutionException

Aynı doluluk kurgusu — bir görev iş parçacığını meşgul ediyor, kuyrukta bir görev bekliyor — iki yürütücüde de birebir aynı. Değişen tek şey kurucuya verilen dördüncü parametre. CallerRunsPolicy reddedilen görevi sessizce çağırana devrediyor; AbortPolicy aynı anı bir istisnaya çeviriyor ve reddedilen görev hiç koşmuyor — üçüncü satır hiçbir zaman basılmıyor, catch bloğu bunu doğruluyor. İkisi de “dolu” durumuna bir yanıt, ama yanıtı seçen görev değil, yürütücüyü kuran koddur.

Kuyruk ve havuz büyüklüğü neden sınırsız bırakılmıyor da bu kararla uğraşılıyor? Çünkü her iş parçacığı bir bedel taşıyor — kendi çağrı çerçevesi, kendi yığıtı, çizelgeleyicinin izlediği bir kayıt. Sınırsız bir kuyruk bu sorunu yalnızca ertelemiş olur: görev üretimi tüketimden hızlıysa kuyruk sonsuza kadar büyür ve bellek tükenir; sınırsız bir havuz ise aynı sorunu iş parçacığı sayısına taşır. Sınırlı kuyruk ve sınırlı havuz, dolma anını görünür kılıp bir karara zorluyor — reddetme yordamı bu kararın adı.

İstisnanın Yolu: execute mi, submit mi

Bir görev çalışırken istisna fırlatırsa bu istisna nereye düşer? Yanıt görevi hangi yöntemle verdiğine bağlı — ve iki yöntemden yalnızca biri istisnayı kendiliğinden gösterir.

EZ21. execute ile verilen bir görevin fırlattığı istisna, o iş parçacığının yakalanmamış istisna işleyicisine düşer. submit ile verilen bir görevin fırlattığı istisna gelecek nesnesinde saklanır ve yalnızca get çağrılırsa çağırana taşınır; get çağrılmazsa istisna hiçbir yerde görünmez.

// IstisnaninYolu.java — execute ile atilan istisna kendiliginden gorunur, submit ile atilan get cagrilmadikca sessizdir
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class IstisnaninYolu {
    public static void main(String[] args) throws Exception {
        AtomicInteger yakalanan = new AtomicInteger(0);
        CountDownLatch bittiA = new CountDownLatch(1);

        ExecutorService havuzA = Executors.newSingleThreadExecutor(gorev -> {
            Thread t = new Thread(gorev);
            t.setUncaughtExceptionHandler((th, hata) -> yakalanan.incrementAndGet());
            return t;
        });
        havuzA.execute(() -> {
            try {
                throw new RuntimeException("gorev hatasi");
            } finally {
                bittiA.countDown();
            }
        });
        bittiA.await();
        havuzA.shutdown();
        havuzA.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("execute()  - yakalanmamis istisna isleyicisine dustu mu: " + (yakalanan.get() == 1));

        ExecutorService havuzB = Executors.newSingleThreadExecutor();
        Future<?> gelecek = havuzB.submit(() -> { throw new RuntimeException("gorev hatasi"); });
        havuzB.shutdown();
        havuzB.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("submit()   - kapanistan sonra hala sessiz mi          : true (get cagrilmadi)");
        try {
            gelecek.get();
            System.out.println("get() sonrasi -> dusmedi");
        } catch (ExecutionException e) {
            System.out.println("get() sonrasi -> " + e.getClass().getSimpleName()
                    + " sarmaliyor: " + e.getCause().getClass().getSimpleName());
        }
    }
}

execute()  - yakalanmamis istisna isleyicisine dustu mu: true
submit()   - kapanistan sonra hala sessiz mi          : true (get cagrilmadi)
get() sonrasi -> ExecutionException sarmaliyor: RuntimeException

execute() ile verilen görev bir istisna fırlattığında bu istisna kendi iş parçacığının yakalanmamış istisna işleyicisine ulaşıyor — hiçbir get çağrılmadan, kendiliğinden. submit() ile verilen aynı görev aynı istisnayı fırlatıyor, ama Future bu istisnayı yutmuyor, saklıyor: havuz kapatılıp beklense bile gelecek nesnesi üzerinde get çağrılmadıkça istisna hiçbir yere düşmüyor. get çağrıldığında istisna ExecutionException içine sarmalanmış olarak, çağıran iş parçacığına taşınıyor. İki verme biçiminin yalnızca biri istisnayı kendiliğinden gösteriyor — sayı: 2’nin 1’i. Bu, yürütücü kapanışının kendisi hakkında da bir uyarı: shutdown() ve awaitTermination() görevlerin bittiğini garanti eder, ama hangi görevin hata verdiğini söylemez; o bilgi yalnızca gelecek nesnelerinden istenirse gelir.

Fark, iki verme biçiminin görevi ne biçimde sardığından geliyor. execute(Runnable) görevi doğrudan çalıştırıyor; iş parçacığının çalıştırma döngüsü görevden çıkan bir istisnayı yakalamıyor, döngünün kendisini sonlandırıp iş parçacığının yakalanmamış istisna işleyicisine taşıyor. submit(...) ise görevi önce bir gelecek nesnesine sarıyor; bu nesnenin çalıştırma yöntemi görevden çıkan her istisnayı kendi içinde yakalayıp saklıyor ve iş parçacığının çalıştırma döngüsüne hiç sızdırmıyor — döngü, görev başarıyla bitmiş gibi devam ediyor. İstisnanın nereye düştüğü, dolayısıyla görevin çağrılma biçiminin bir yan etkisi: aynı istisna, aynı satırdan fırlasa da iki ayrı yoldan geçiyor.

Dört ölçüm de aynı yöne işaret ediyor: bir yürütücü servisine görev vermek, o görevi kimin koşturacağını çağıranın elinden alıyor ama görünmez kılmıyor. Ayrık kimlik sayısı havuzun büyüklüğünü aşmıyor, dolu havuzda görev veren iş parçacığı devreye giriyor, reddetme yordamı bu anı kimin karşıladığını belirliyor, istisnanın nereye düştüğü de verme biçiminin bir sonucu. Dördünde de kod satırında görünmeyen bir taraf kararı veriyor; bu ders o tarafı her seferinde koşturarak açığa çıkardı.

Özet

• ExecutorService, bir görevi hangi iş parçacığının koşturacağına kendisi karar veren bir soyutlamadır; ayrık iş parçacığı kimliği sabit havuzda çekirdek sayısıyla sınırlı kalır, görev sayısı bu sınırı aşsa da aşmaz.
• Havuz boşken görevi havuzun kendi iş parçacığı koşturur; havuz doluyken aynı execute çağrısı görevi veren iş parçacığında koşturur — kod aynıdır, farkı yaratan çağrı anındaki doluluktur.
• Reddetme yordamı görevin değil, yürütücüyü kuran tarafın seçtiği bir parametredir; CallerRunsPolicy görevi çağırana devreder, AbortPolicy aynı anı bir istisnaya çevirir ve görev hiç koşmaz.
• execute ile atılan bir istisna yakalanmamış istisna işleyicisine kendiliğinden düşer; submit ile atılan aynı istisna gelecek nesnesinde saklanır ve yalnızca get çağrılırsa çağırana taşınır.
• ExecutorService’in AutoCloseable gerçekleştirimi kaynakla deneme kalıbına katılır; kapanış çağrısı yürütücü için de otomatiktir.

Sonraki Adım

Bu derste havuzun iş parçacıkları sınırlı bir kaynaktı: sabit sayıda vardı, doldular, sırayla yeniden kullanıldılar ve doluluk çağıranı işe koştu. Bu sınır, iş parçacığının kendisinin pahalı olmasından geliyor. Sıradaki ders bu öncülü kaldırıyor: platform iş parçacığı yerine çok daha ucuz bir iş parçacığı türü kullanan bir yürütücüyle aynı soru soruluyor — havuz yerine iş parçacığının kendisi ucuzlarsa görevi kimin koşturacağı sorusu nasıl değişiyor?