Метод newWorkStealingPool

Keling, bizga ExecutorService tayyorlaydigan navbatdagi metodni ko'rib chiqamiz — newWorkStealingPool.

Bu oqimlar havzasining maxsusligi – uning ishlash konsepsiyasi “ishni o'g'irlash”da turibdi.

Vazifalar navbatga yig'iladi va protsessorlar bo'ylab taqsimlanadi. Ammo agar protsessor band bo'lsa, boshqa bo'sh protsessor undan vazifani o'g'irlab, uni bajarishi mumkin. Bunday format Java'ga ko'p oqimli dasturlarda nizolarni kamaytirish uchun kiritilgan. Asosida fork/join freymvorki yotadi.

fork/join

fork/join freymvorkida vazifalar rekursiv ravishda dekompoziya qilinadi, ya'ni kichik vazifalarga ajratiladi. Keyin vazifalar individual bajariladi va kichik vazifalar natijalari asosiy vazifalar natijalarini shakllantirish uchun birlashtiriladi.

fork metodi ma’lum bir oqimda vazifani asinxron ravishda ishga tushiradi, va join metodi bu vazifa ustida ish yakunlanishini kutishga imkon beradi.

newWorkStealingPool

newWorkStealingPool metodining ikki marta amalga oshirilishi mavjud:

public static ExecutorService newWorkStealingPool(int parallelism) {
        return new ForkJoinPool
            (parallelism,
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }
 
public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

Biz darhol ko'ramizki, biz ThreadPoolExecutor konstruktorini chaqirmaymiz, bu erda biz ForkJoinPool entiteti bilan ishlayapmiz. Bu ham ThreadPoolExecutor kabi AbstractExecutorService ning amalga oshirilishidir.

Bizga 2 metodni tanlash imkonini berishadi. Ular orasidagi farq shundaki, birinchi holda biz qaysi darajadagi parallelizmni ko'rishni xohlaymizligini o'zimiz belgilaymiz. Agar biz bu qiymatni ko'rsatmasak, bizning havzamizda parallelizm darajasi ayni paytda Java virtual mashinasiga mavjud bo'lgan protsessor yadrolari soniga teng bo'ladi.

Bu qanday ishlashini amalda tushundingizmi:

Collection<Callable<Void>> tasks = new ArrayList<>();
        ExecutorService executorService = Executors.newWorkStealingPool(10);
 
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            Callable<Void> callable = () -> {
                System.out.println("Foydalanuvchi №" + taskNumber + " so'rovi " + Thread.currentThread().getName() + " oqimida qayta ishlangan.");
                return null;
            };
            tasks.add(callable);
        }
        executorService.invokeAll(tasks);

Biz 10 ta vazifa yaratamiz, ular bajarilish holatini chiqaradi. Shundan so'ng, barcha vazifalarni invokeAll metodi yordamida ishga tushiramiz.

10 ta oqimda 10 ta vazifa bajarilganda natijalar:

Biz ko'ramizki, navbat shakllangandan so'ng, oqimlar vazifalarni bajarish uchun olishgan. Shuningdek, havzadagi oqimlar orasidagi taqsimot qanday bo'lishini tekshirish mumkin, masalan, 20 ta vazifa 10 ta oqimda.

Chiqishdan ko'rinib turibdiki, ba'zi oqimlar bir nechta vazifalarni bajarishga ulgurdi (ForkJoinPool-1-worker-6 4 ta vazifani bajardi), ba'zilari esa faqat birini (ForkJoinPool-1-worker-2). Agar call metodining amalga oshirilishiga 1 soniyalik kechikma qo'shsak, manzara o'zgaradi.

Callable<Void> callable = () -> {
   System.out.println("Foyдалануvчи №" + taskNumber + " so'roви " + Thread.currentThread().getName() + " oqимида кайта ишланган.");
   TimeUnit.SECONDS.sleep(1);
   return null;
};

Tajriba uchun, bu kodni boshqa mashinada bajaramiz. Olingan chiqish:

Bu chiqishdagi qiziqarli narsa shundaki, biz havzadagi oqimlarni “buyurtma” qildik. Ammo ishchilar nomlari birdan o'nga kiritilgani emas, balki kattaroq. Agar noyob nomlarga qarasak, ishchilar haqiqatan ham o'n (3, 5, 9, 13, 17, 19, 21, 23, 27 va 31). Mantiqiy savol tug'iladi: nega bunday bo'ldi? Har qanday noma'lum vaziyatda debugdan foydalan.

final ForkJoinPool forkJoinPool = (ForkJoinPool) executorService;

Biz bu ob'ektni aynan invokeAll metodi chaqirilgandan keyin Evaluate Expression rejimida ko'rib chiqamiz. Buning uchun invokeAll metodidan keyin har qanday buyruqni qo'shamiz, masalan, bo'sh sout, va unda breakpoin qo'yamiz.

Biz ko'ramizki, havzada 10 ta oqim bor, lekin oqimlar massivi (ishchilar) o'lchami 32 ga teng. G'alati, lekin yaxshi, yanada chuqurroq o'rganamiz. Havza yaratishda parallelizm parametrini 32 dan ko'proq qo'yishga harakat qilamiz, masalan, 40.

ExecutorService executorService = Executors.newWorkStealingPool(40);

Va debuggingda yana bir bor forkJoinPool ob'ektini ko'ramiz.

Endi ishchilar massivi o'lchami 128. Bu JVM ichki optimizatsiyasi ekanligini taxmin qilish mumkin. O'zingizni JDK (openjdk-14) kodida topishga harakat qilaylik:

Ha, haqiqatan ham: ishchilar massivining o'lchami parallelizm qiymatidan kelib chiqqan holda hisoblab chiqiladi va unga bitli manipulatsiyalar amalga oshiriladi. Bu yerda aynan nima sodir bo'layotganini tushunishga harakat qilishning hojati yo'q. Faqat bunday optimallaşdırmanın mavjudligini bilish kifoya.

Bizning misolimizning yana bir xususiyati — bu invokeAll metodining ishlatilishi. Ta'kidlash kerakki, invokeAll metodi bizga natijani qaytarishi mumkin, aniqrog'i, natijalarni saqlovchi ro'yxat (List<Future<Void>>) qaerda biz har bir vazifaning natijasini olishimiz mumkin.

var results = executorService.invokeAll(tasks);
        for (Future<Void> result : results) {
            // Vazifani bajarish natijasini qayta ishlash
        }

Bunday maxsus xizmat va oqimlar havzasini parallelizm darajasini oldindan yoki oldindan aytib bo'lmaydigan, lekin nazarda tutilgan vazifalarda ishlatish mumkin.