استخر نخ (Thread Pool) ‌در جاوا — راهنمای مقدماتی

در برنامه‌نویسی می‌توان جهت بهبود عملکرد و سرعت برنامه، هر یک از Taskها و بخش‌های مختلف اپلیکیشن را به یک Thread سپرد. در این مقاله به بررسی مفهوم استخر نخ در جاوا یا همان Thread Pool در جاوا می‌پردازیم. به این منظور کار خود را با معرفی پیاده‌سازی‌های مختلف آن در کتابخانه استاندارد جاوا آغاز کرده و سپس به بررسی کتابخانه Guava گوگل می‌پردازیم.

استخر نخ

نخ‌ها در جاوا به نخ‌های سطح سیستمی نگاشت می‌شوند که جزء منابع سیستم عامل محسوب می‌شوند. اگر نخ‌ها را به شیوه‌ای غیر قابل کنترل ایجاد کنید، ممکن است به سرعت همه این منابع را مصرف کنید.

فیلم آموزش برنامه نویسی چند نخی یا مالتی ترد در سی پلاس پلاس C++‎ برای اجرای همزمان چندین فرایند در فرادرس

کلیک کنید

«سوئیچ زمینه» (context switching) بین این نخ‌ها از سوی سیستم‌ عامل به خوبی انجام می‌گیرد تا حالت «محاسبات موازی» (parallelism) شبیه‌سازی شود. به بیان ساده‌تر هر چه نخ‌های بیشتری ایجاد شوند، هر نخ زمان کمتری برای انجام کار عملی صرف می‌کند.

الگوی «استخر نخ» (Thread Pool) به حفظ منابع در یک اپلیکیشن چندنخی کمک می‌کند و همچنین محاسبات موازی را یک چارچوب از پیش تعریف‌شده خاص قرار می‌دهد.

زمانی که از استخر نخ استفاده می‌کنیم، می‌توانیم کد «هم‌زمان» (concurrent) را به شکل وظایف موازی بنویسیم و آن‌ها را برای اجرا به یک وهله (Instance) از یک استخر نخ تحویل دهیم. این وهله چند نخ با استفاده مجدد را برای اجرای این وظایف کنترل می‌کند.

این الگو به ما امکان می‌دهد که تعداد نخ‌هایی که اپلیکیشن ایجاد می‌کند و چرخه عمر آن‌ها را کنترل کنیم و همچنین اجرای وظایف را با نگه‌داشتن وظایف ورودی در صف، زمان‌بندی نماییم.

استخر‌های نخ در جاوا

در این بخش به بررسی انواع استخر‌های نخ موجود در جاوا می‌پردازیم.

Executors ،Executor و ExecutorService

فیلم مجموعه آموزش رایگان جاوا – مقدماتی تا پیشرفته در فرادرس

کلیک کنید

کلاس کمکی Executors شامل چند متد برای ایجاد وهله‌های استخر نخ از پیش پیکر‌بندی‌شده است. این کلاس‌ها محل خوبی برای شروع استفاده از استخر‌های نخ محسوب می‌شوند. در صورتی که نیاز به تنظیم دقیق موارد مختلف ندارید، می‌توانید از این کلاس‌ها بهره بگیرید.

اینترفیس‌های Executor و ExecutorService برای کار با پیاده‌سازی‌های مختلف استخر‌ نخ در جاوا مورد استفاده قرار می‌گیرند. به طور معمول شما باید کد خود را از پیاده‌سازی عملی استخر نخ جدا نگه دارید و از طریق اپلیکیشن از این اینترفیس استفاده کنید.

اینترفیس Executor یک متد منفرد به نام execute دارد که وهله‌های Runnable برای اجرا به آن تحویل داده می‌شوند.

در این بخش یک مثال ساده از شیوه استفاده از Executors API برای به دست آوردن یک وهله Executor ارائه می‌شود. این وهله Executor از سوی یک استخر نخ منفرد پشتیبانی می‌شود و صف بی‌کرانی برای اجرای وظایف به صورت ترتیبی دارد. در این مقاله یک وظیفه منفرد را اجرا می‌کنیم که به سادگی عبارت Hello World را روی صفحه پرینت می‌کند. این وظیفه به صورت یک لامبدا تحویل می‌شود که جاوا آن را به صورت Executors API استنباط می‌کند.

1Executor executor = Executors.newSingleThreadExecutor();
2executor.execute(() -> System.out.println("Hello World"));

اینترفیس ExecutorService شامل تعداد زیادی متد برای کنترل کردن پیشرفت وظایف و مدیریت خاتمه سرویس است. با استفاده از این اینترفیس می‌توانید وظایف را برای اجرا تحویل دهید و همچنین اجرای آن‌ها را با بهره‌گیری از وهله Future بازگشتی کنترل کنید.

در مثال زیر، یک Future ایجاد می‌کنیم، یک وظیفه تحویل می‌دهیم و سپس از متد get مربوط به Future که بازگشت یافته استفاده می‌کنیم تا زمان پایان یافتن وظیفه منتظر شویم و مقدار بازگشت یابد:

1ExecutorService executorService = Executors.newFixedThreadPool(10);
2Future<String> future = executorService.submit(() -> "Hello World");
3// some operations
4String result = future.get();

البته در سناریو‌های واقعی به طور معمول لزومی ندارد که ()future.get را بی‌درنگ فراخوانی کنیم، بلکه فراخوانی آن را تا زمانی که عملاً به مقدار محاسبه نیاز داشته باشیم به تعویق می‌اندازیم. متد submit با گرفتن Runnable یا Callable که هر دوی آن‌ها اینترفیس‌های تابعی هستند overload شده است و می‌تواند به صورت لامبدا ارسال شود.

متد منفرد Runnable یک استثنا ایجاد نمی‌کند و مقداری بازگشت نمی‌دهد. اینترفیس Callable می‌تواند راحت‌تر باشد، چون به ما امکان می‌دهد که یک استثنا ایجاد کرده و مقداری را بازگشت دهیم. در نهایت باید اشاره کنیم که بهتر است به کامپایلر اجازه دهیم که نوع Callable را استنباط کند و تنها یک مقدار از لامبدا بازگشت دهد.

ThreadPoolExecutor

ThreadPoolExecutor یک پیاده‌سازی بسط‌پذیر استخر نخ با پارامترها و قابلیت‌های زیادی است که امکان تنظیم دقیق آن را فراهم می‌سازند. پارامترهای اصلی پیکربندی که در اینجا بررسی می‌کنیم شامل corePoolSize maximumPoolSize، و keepAliveTime هستند.

این استخر شامل تعداد ثابتی از نخ‌های هسته است که همواره در درون آن نگهداری می‌شوند و برخی نخ‌های اضافی نیز وجود دارند که می‌توانند ایجاد شده و پس از این که دیگر لازم نبودند، خاتمه یابند. پارامتر corePoolSize برابر با تعداد نخ‌های هسته است که درون استخر وهله‌سازی شده و نگهداری می‌شوند. زمانی که وظیفه جدیدی از راه می‌رسد، اگر همه نخ‌ها مشغول باشند و صف درونی پر شده باشد در این صورت استخر می‌تواند maximumPoolSize را افزایش دهد.

پارامتر keepAliveTime یک بازه زمانی است که نخ‌های مازاد مجاز هستند تا در حالت بیکار بمانند. به صورت پیش‌فرض ThreadPoolExecutor تنها نخ‌های غیر هسته را حذف می‌کند. برای به‌کارگیری برخی سیاست‌های حذف نخ‌های هسته می‌توان از متد allowCoreThreadTimeOut(true) استفاده کرد. این پارامتر طیف وسیعی از حالت‌های مختلف را پوشش می‌دهد، اما معمول‌ترین پیکربندی در متد‌های استاتیک Executors از پیش‌ تعریف‌شده است.

برای نمونه متد Executors یک ThreadPoolExecutor با مقادیر برابر برای پارامترهای corePoolSize و maximumPoolSize می‌سازد و مقدار keepAliveTime نیز صفر است. این بدان معنی است که تعداد نخ‌ها در این استخر نخ همواره یکسان است:

1ThreadPoolExecutor executor = 
2  (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
3executor.submit(() -> {
4    Thread.sleep(1000);
5    return null;
6});
7executor.submit(() -> {
8    Thread.sleep(1000);
9    return null;
10});
11executor.submit(() -> {
12    Thread.sleep(1000);
13    return null;
14});
15 
16assertEquals(2, executor.getPoolSize());
17assertEquals(1, executor.getQueue().size());

در مثال فوق یک ThreadPoolExecutor را با تعداد ثابتی نخ وهله‌سازی کردیم. این بدان معنی است که اگر تعداد وظایفی که به صورت هم‌زمان در حال اجرا هستند در همه حال کمتر یا مساوی دو باشد، در این صورت بی‌درنگ اجرا می‌شوند. در غیر این صورت برخی از این وظایف را می‌توان در صف قرار داد تا منتظر اجرا بمانند.

این وظیفه‌‌های ThreadPoolExecutor را برای شبیه‌سازی کارهای سنگین طوری تنظیم کردیم که 1000 میلی‌ثانیه به حالت sleep بروند. دو وظیفه اول هم‌زمان اجرا می‌شوند و وظیفه سوم باید در صف منتظر بماند. این وضعیت را می‌توان با فراخوانی متدهای ()getPoolSize و ()getQueue().size بی‌درنگ پس از تحویل وظایف بررسی کرد.

یک ThreadPoolExecutor از پیش پیکربندی‌شده دیگر را می‌توان با استفاده از متد ()Executors.newCachedThreadPool ایجاد کرد. این متد اصلاً هیچ تعداد نخی نمی‌‌گیرد. در واقع مقدار corePoolSize برابر با 0 است و maximumPoolSize نیز روی Integer.MAX_VALUE تنظیم شده است. در این مورد keepAliveTime روی 60 ثانیه قرار دارد.

این مقادیر پارامترها به آن معنی است که استخر نخ کَش‌شده می‌تواند بدون حد و مرز برای اجرای هر تعداد وظیفه تحویلی گسترش یابد. اما زمانی که نخ‌ها دیگر مورد نیاز نباشند، پس از این که 60 ثانیه بیکار ماندند حذف می‌شوند. یک کاربرد معمول این سناریو زمانی است که وظایف با عمر کوتاه زیادی در اپلیکیشن داشته باشید.

1ThreadPoolExecutor executor = 
2  (ThreadPoolExecutor) Executors.newCachedThreadPool();
3executor.submit(() -> {
4    Thread.sleep(1000);
5    return null;
6});
7executor.submit(() -> {
8    Thread.sleep(1000);
9    return null;
10});
11executor.submit(() -> {
12    Thread.sleep(1000);
13    return null;
14});
15 
16assertEquals(3, executor.getPoolSize());
17assertEquals(0, executor.getQueue().size());

در مثال فوق، اندازه صف همواره صفر است، زیرا به صورت داخلی از یک وهله از SynchronousQueue استفاده می‌کند. در SynchronousQueue جفت عملگر‌های insert و remove همواره به صورت هم‌زمان اجرا می‌شوند، از این رو صف عملاً هرگز شامل هیچ چیزی نیست.

Executors.newSingleThreadExecutor() API یک شکل معمول دیگر از ThreadPoolExecutor را می‌سازد که شامل یک نخ منفرد است. اجراکننده با یک نخ منفرد برای ایجاد یک حلقه رویداد مناسب است. پارامترهای corePoolSize و maximumPoolSize برابر با 1 هستند و keepAliveTime مقدار keepAliveTime نیز برابر با صفر است. وظایف موجود در مثال فوق به صورت ترتیبی اجرا خواهند شد، بنابراین مقدار فلگ پس از پایان یافتن وظیفه برابر با 2 است:

1AtomicInteger counter = new AtomicInteger();
2 
3ExecutorService executor = Executors.newSingleThreadExecutor();
4executor.submit(() -> {
5    counter.set(1);
6});
7executor.submit(() -> {
8    counter.compareAndSet(1, 2);
9});

به علاوه ThreadPoolExecutor با یک پوشش تغییرناپذیر تزیین یافته است، بنابراین پس از ایجاد شدن نمی‌تواند تغییر یابد. همچنین توجه کنید که به همین دلیل نمی‌توانیم آن را به یک ThreadPoolExecutor تبدیل کنیم.

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor اقدام به بسط ThreadPoolExecutor می‌کند و همچنین اینترفیس ScheduledExecutorService را با چند متد اضافی پیاده‌سازی می‌کند.

• متد schedule امکان اجرای یک وظیفه را پس از مدت تأخیر مشخص فراهم می‌سازد.
• متد scheduleAtFixedRate امکان اجرای یک وظیفه را پس از یک تاخیر مشخص‌شده اولیه و در ادامه اجرای مکرر با دوره زمانی معین فراهم می‌سازد. این دوره زمانی بر اساس زمان آغاز شدن وظایف اندازه‌گیری می‌شود و از این رو نرخ اجرا ثابت است.
• متد scheduleWithFixedDelay مشابه متد scheduleAtFixedRate است و یک وظیفه معین را به صورت مکرر اجرا می‌کند اما تأخیر مشخص شده بین انتهای وظیفه قبلی و آغاز وظیفه بعدی اندازه‌گیری می‌شود. از این رو نرخ اجرا، بسته به زمانی که طول می‌کشد تا آن وظیفه معین اجرا شود، ممکن است متفاوت باشد.

متد ()Executors.newScheduledThreadPool به طور معمول برای ایجاد یک ()Executors.newScheduledThreadPool با corePoolSize معین، maximumPoolSize نامتناهی و keepAliveTime صفر استفاده می‌شود. روش زمان‌بندی یک وظیفه برای اجرا در 500 میلی‌ثانیه به صورت زیر است:

1ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
2executor.schedule(() -> {
3    System.out.println("Hello World");
4}, 500, TimeUnit.MILLISECONDS);

کد زیر شیوه اجرای یک وظیفه پس از یک تأخیر 300 میلی‌ثانیه‌ای و سپس تکرار آن در هر 100 میلی‌ثانیه یک بار را نمایش می‌‌دهد. پس از زمان‌بندی یک وظیفه، با بهره‌گیری از قفل CountDownLatch صبر می‌کنیم تا سه بار اجرا شود و سپس آن را با استفاده از متد ()Future.cancel لغو می‌کنیم.

1CountDownLatch lock = new CountDownLatch(3);
2 
3ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
4ScheduledFuture<?> future = executor.scheduleAtFixedRate(() -> {
5    System.out.println("Hello World");
6    lock.countDown();
7}, 500, 100, TimeUnit.MILLISECONDS);
8 
9lock.await(1000, TimeUnit.MILLISECONDS);
10future.cancel(true);

ForkJoinPool

ForkJoinPool بخش مرکزی فریمورک fork/join است که در جاوا 7 معرفی شده است. این فریمورک به حل یک مشکل رایج به صورت ایجاد چند وظیفه در الگوریتم‌های بازگشتی کمک می‌کند. با استفاده از یک ThreadPoolExecutor ساده می‌توانید نخ‌ها را به سرعت اجرا کنید، چون هر وظیفه یا وظیفه فرعی روی نخ خاص خود اجرا می‌شود.

در یک فریمورک fork/join هر وظیفه می‌تواند چند وظیفه فرعی ایجاد کند و با استفاده از متد join منتظر تکمیل شدن آن‌ها بماند. مزیت فریمورک fork/join این است که یک نخ جدید برای هر وظیفه یا وظیفه فرعی ایجاد نمی‌کند. به جای آن الگوریتم fork/join را پیاده‌سازی می‌کند.

در این بخش به بررسی یک مثال ساده استفاده از ForkJoinPool برای پیمایش یک درخت از گره‌ها و محاسبه مجموع همه مقادیر برگ می‌پردازیم. در این مثال یک پیاده‌سازی ساده از یک درخت را می‌بینید که شامل یک گره، یک مقدار int و یک مجموعه از گره‌های فرزند است:

1static class TreeNode {
2 
3    int value;
4 
5    Set<TreeNode> children;
6 
7    TreeNode(int value, TreeNode... children) {
8        this.value = value;
9        this.children = Sets.newHashSet(children);
10    }
11}

اگر بخواهیم همه مقادیر را در یک درخت به صورت موازی جمع بزنیم، باید یک اینترفیس RecursiveTask<Integer>‎ پیاده‌سازی کنیم. هنر وظیفه گره خاص خود را دریافت می‌کند و آن را به مقدار مجموع مقادیر فرزندانش اضافه می‌کند. برای محاسبه مجموع مقادیر فرزندان، پیاده‌سازی وظیفه کارهای زیر را انجام می‌دهد:

• مجموعه children را استریم می‌کند.
• روی این استریم یک map ایجاد کرده و یک CountingTask جدید برای هر عنصر می‌سازد.
• هر وظیفه فرعی را با فورک کردن آن اجرا می‌کند.
• نتایج را با فراخوانی متد join روی هر وظیفه فورک‌شده گردآوری می‌کند.
• در نهایت نتیجه با استفاده از کلکتور Collectors.summingInt جمع زده می‌شود.

1public static class CountingTask extends RecursiveTask<Integer> {
2 
3    private final TreeNode node;
4 
5    public CountingTask(TreeNode node) {
6        this.node = node;
7    }
8 
9    @Override
10    protected Integer compute() {
11        return node.value + node.children.stream()
12          .map(childNode -> new CountingTask(childNode).fork())
13          .collect(Collectors.summingInt(ForkJoinTask::join));
14    }
15}

کد مورد نیاز برای اجرای محاسبات روی درخت واقعی بسیار ساده است:

1TreeNode tree = new TreeNode(5,
2  new TreeNode(3), new TreeNode(2,
3    new TreeNode(2), new TreeNode(8)));
4 
5ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
6int sum = forkJoinPool.invoke(new CountingTask(tree));

پیاده‌سازی استخر نخ در Guava

Guava یک کتابخانه محبوب گوگل برای اجرای کارهای مختلف است. این کتابخانه کلاس‌های هم‌زمانی مفید زیادی دارد که شامل پیاده‌سازی‌های مختلف کارآمدی از ExecutorService است. امکان وهله‌سازی یا زیرکلاس‌سازی از پیاده‌سازی کلاس‌ها وجود ندارد و از این رو تنها نقطه ورودی برای ایجاد وهله‌های آن‌ها کلاس کمکی MoreExecutors است.

فیلم آموزش برنامه نویسی جاوا Java در فرادرس

کلیک کنید

افزودن Guava به صورت یک وابستگی Maven

وابستگی زیر را به فایل Maven pom اضافه کنید تا کتابخانه Guava به پروژه شما افزوده شود. جدیدترین نسخه کتابخانه Guava را می‌توانید در ریپازیتوری Maven Central (+) پیدا کنید:

1<dependency>
2    <groupId>com.google.guava</groupId>
3    <artifactId>guava</artifactId>
4    <version>19.0</version>
5</dependency>

Direct Executor و سرویس Direct Executor

برخی اوقات می‌خواهیم یک وظیفه را بسته به برخی شرایط، یا در نخ کنونی و یا در استخر نخ اجرا کنیم. در این حالت بهتر است از یک اینترفیس Direct Executor منفرد استفاده کرده و تنها بین پیاد‌ه‌سازی‌ها سوئیچ کنیم. با این که دست یافتن به یک پیاده‌سازی Executor یا ExecutorService که وظیفه را در نخ جاری اجرا کند، کار چندان دشواری نیست، اما با این حال نیازمند نوشتن مقداری کد تکراری است.

خوشبختانه Guava وهله‌های از پیش تعریف‌شده‌ای به این منظور در اختیار ما قرار می‌دهد. در ادامه مثالی را مشاهده می‌کنید که اجرای یک وظیفه را در همان نخ نمایش می‌دهد. با این که وظیفه ارائه شده به مدت 500 میلی‌ثانیه به خواب می‌رود، اما نخ جاری را مسدود می‌سازد و نتیجه بی‌درنگ پس از پایان یافتن فراخوانی اجرا در اختیار ما قرار می‌گیرد:

1Executor executor = MoreExecutors.directExecutor();
2 
3AtomicBoolean executed = new AtomicBoolean();
4 
5executor.execute(() -> {
6    try {
7        Thread.sleep(500);
8    } catch (InterruptedException e) {
9        e.printStackTrace();
10    }
11    executed.set(true);
12});
13 
14assertTrue(executed.get());

وهله بازگشت یافته از سوی متد ()directExecutor در عمل یک سینگلتون استاتیک است، از این رو استفاده از این متد موجب هیچ گونه سرباری برای ایجاد شیء نمی‌شود.

این متد نسبت به ()MoreExecutors.newDirectExecutorService ترجیح دارد، زیرا آن API یک پیاده‌سازی سرویس executor را امکانات کامل در هر بار فراخوانی ایجاد می‌کند.

سرویس‌های Exiting Executor

یکی از مشکلات رایج دیگر، خاموش کردن ماشین مجازی در زمانی است که استخر نخ همچنان مشغول اجرای وظایف خود است. حتی با وجود یک «سازوکار لغو»، هیچ تضمینی وجود ندارد که وظیفه مورد نظر به درستی کار کند و زمانی که سرویس executor خاموش می‌شود، متوقف شود. این امر موجب می‌شود که JVM در زمانی که وظایف مشغول اجرا هستند، به صورت نامعینی معلق بماند.

برای حل این مشکل، کتابخانه Guava یک خانواده از سرویس‌های exiting executor معرفی کرده است که بر اساس نخ‌های daemon عمل می‌کند که همگی همراه با JVM خاتمه می‌یابند.

این سرویس‌ها همچنین قلاب را با استفاده از متد ()Runtime.getRuntime().addShutdownHook خاموش می‌کنند و از خاتمه یافتن ماشین مجازی برای یک زمان از پیش تعیین شده در حالی که هنوز وظایفی مشغول اجرا هستند جلوگیری می‌کند.

در مثال زیر یک وظیفه را تحویل می‌دهیم که شامل یک حلقه نامتناهی است، اما از سرویس exiting executor با یک زمان پیکر‌بندی‌شده 100 میلی‌ثانیه‌ای استفاده کرده‌ایم تا منتظر خاتمه وظایف در زمان خاموش شدن ماشین مجازی بماند. بدون وجود این exiting executor، این وظیفه موجب می‌شود که ماشین مجازی به صورت نامعینی تعلیق شود.

1ThreadPoolExecutor executor = 
2  (ThreadPoolExecutor) Executors.newFixedThreadPool(5);
3ExecutorService executorService = 
4  MoreExecutors.getExitingExecutorService(executor, 
5    100, TimeUnit.MILLISECONDS);
6 
7executorService.submit(() -> {
8    while (true) {
9    }
10});

گوش دادن به دکوراتورها

قابلیت گوش دادن به دکوراتورها به ما امکان می‌دهد که یک پوشش پیرامون ExecutorService ایجاد کرده و وهله‌های ListenableFuture را به محض تحویل وظیفه به جای وهله‌های ساده Future تحویل بگیریم. این متد امکان افزودن یک شنونده را فراهم ساخته است که به محض تکمیل شدن وظیفه فراخوانی می‌شود.

شاید بخواهید از متد ()ListenableFuture.addListener به صورت مستقیم استفاده کنید، اما این متد برای بسیاری از متدهای کمکی در کلاس کاربردی Futures ضروری است. برای نمونه با استفاده از متد ()Futures.allAsList می‌توانید چند وهله ListenableFuture را با هم در یک ListenableFuture منفرد ترکیب کنید که به محض تکمیل شدن موفق همه future-ها ترکیب می‌شود:

1ExecutorService executorService = Executors.newCachedThreadPool();
2ListeningExecutorService listeningExecutorService = 
3  MoreExecutors.listeningDecorator(executorService);
4 
5ListenableFuture<String> future1 = 
6  listeningExecutorService.submit(() -> "Hello");
7ListenableFuture<String> future2 = 
8  listeningExecutorService.submit(() -> "World");
9 
10String greeting = Futures.allAsList(future1, future2).get()
11  .stream()
12  .collect(Collectors.joining(" "));
13assertEquals("Hello World", greeting);

سخن پایانی

در این مقاله با موضوع بررسی قابلیت‌های مختلف استخر نخ در جاوا به بررسی الگوی Thread Pool و پیاده‌سازی‌های آن در کتابخانه استاندارد این زبان و همچنین کتابخانه Guava گوگل پرداختیم.

فیلم مجموعه آموزش جاوا Java – مقدماتی تا پیشرفته در فرادرس

کلیک کنید