ThreadPoolExecutor – مثال على حوض توThread بل الجافا

حوض موضوعات جافا يدير حوض موضوعات العمال. يحتوي على طابور انتظار يحتفظ بالمهام في انتظار التنفيذ. يمكننا استخدام ThreadPoolExecutor لإنشاء حوض موضوعات في جافا. يدير حوض موضوعات جافا مجموعة من الموضوعات القابلة للتشغيل. تنفذ الموضوعات العاملة الموضوعات القابلة للتشغيل من الطابور. java.util.concurrent.Executors توفر طرق المصنع والدعم لـ java.util.concurrent.Executor لإنشاء حوض موضوعات في جافا. Executors هو فئة مساعدة توفر أيضًا طرقًا مفيدة للعمل مع ExecutorService وScheduledExecutorService وThreadFactory وCallable classes من خلال طرق مصنع مختلفة. دعونا نكتب برنامجًا بسيطًا لشرح كيفية عمله. أولاً، نحتاج إلى وجود فئة Runnable، تسمى WorkerThread.java

package com.journaldev.threadpool;

public class WorkerThread implements Runnable {
  
    private String command;
    
    public WorkerThread(String s){
        this.command=s;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+" Start. Command = "+command);
        processCommand();
        System.out.println(Thread.currentThread().getName()+" End.");
    }

    private void processCommand() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString(){
        return this.command;
    }
}

مثال على ExecutorService

إليك فئة البرنامج الاختباري SimpleThreadPool.java، حيث نقوم بإنشاء حوض موضوعات ثابت من إطار العمال.

package com.journaldev.threadpool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SimpleThreadPool {

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 10; i++) {
            Runnable worker = new WorkerThread("" + i);
            executor.execute(worker);
          }
        executor.shutdown();
        while (!executor.isTerminated()) {
        }
        System.out.println("Finished all threads");
    }
}

في البرنامج أعلاه، نقوم بإنشاء حوض موضوعات ثابت بحجم 5 موضوعات عاملة. ثم نقوم بتقديم 10 وظائف إلى هذا الحوض، نظرًا لأن حجم الحوض هو 5، سيبدأ في العمل على 5 وظائف وسيكون الوظائف الأخرى في حالة الانتظار، وبمجرد الانتهاء من إحدى الوظائف، ستتم اختيار وظيفة أخرى من طابور الانتظار بواسطة الموضوع العامل وستتم تنفيذها. إليك ناتج البرنامج أعلاه.

pool-1-thread-2 Start. Command = 1
pool-1-thread-4 Start. Command = 3
pool-1-thread-1 Start. Command = 0
pool-1-thread-3 Start. Command = 2
pool-1-thread-5 Start. Command = 4
pool-1-thread-4 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
pool-1-thread-3 End.
pool-1-thread-3 Start. Command = 8
pool-1-thread-2 End.
pool-1-thread-2 Start. Command = 9
pool-1-thread-1 Start. Command = 7
pool-1-thread-5 Start. Command = 6
pool-1-thread-4 Start. Command = 5
pool-1-thread-2 End.
pool-1-thread-4 End.
pool-1-thread-3 End.
pool-1-thread-5 End.
pool-1-thread-1 End.
Finished all threads

الناتج يؤكد وجود خمس خيوط في المجموعة تحمل أسماء من “pool-1-thread-1” إلى “pool-1-thread-5” وهي مسؤولة عن تنفيذ المهام المقدمة إلى المجموعة.

مثال على ThreadPoolExecutor

يوفر الفئة Executors تنفيذًا بسيطًا لـ ExecutorService باستخدام ThreadPoolExecutor ولكن ThreadPoolExecutor يوفر المزيد من الميزات بكثير. يمكننا تحديد عدد الخيوط التي ستكون حية عند إنشاء مثيل ThreadPoolExecutor ويمكننا تحديد حجم حوض الخيوط وإنشاء تنفيذ RejectedExecutionHandler الخاص بنا للتعامل مع المهام التي لا يمكن أن تتناسب مع طابور العمل. إليك تنفيذنا المخصص لواجهة RejectedExecutionHandler.

package com.journaldev.threadpool;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println(r.toString() + " is rejected");
    }

}

ThreadPoolExecutor يوفر العديد من الطرق التي يمكننا من خلالها معرفة الحالة الحالية للمنفذ، حجم حوض الخيوط، عدد الخيوط النشطة، وعدد المهام. لذلك لدي خيط رصد سيقوم بطباعة معلومات المنفذ في فاصل زمني معين.

package com.journaldev.threadpool;

import java.util.concurrent.ThreadPoolExecutor;

public class MyMonitorThread implements Runnable
{
    private ThreadPoolExecutor executor;
    private int seconds;
    private boolean run=true;

    public MyMonitorThread(ThreadPoolExecutor executor, int delay)
    {
        this.executor = executor;
        this.seconds=delay;
    }
    public void shutdown(){
        this.run=false;
    }
    @Override
    public void run()
    {
        while(run){
                System.out.println(
                    String.format("[monitor] [%d/%d] Active: %d, Completed: %d, Task: %d, isShutdown: %s, isTerminated: %s",
                        this.executor.getPoolSize(),
                        this.executor.getCorePoolSize(),
                        this.executor.getActiveCount(),
                        this.executor.getCompletedTaskCount(),
                        this.executor.getTaskCount(),
                        this.executor.isShutdown(),
                        this.executor.isTerminated()));
                try {
                    Thread.sleep(seconds*1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }
            
    }
}

إليك مثال على تنفيذ حوض الخيوط باستخدام ThreadPoolExecutor.

package com.journaldev.threadpool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class WorkerPool {

    public static void main(String args[]) throws InterruptedException{
        //تنفيذ المعالج RejectedExecutionHandler
        RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
        //الحصول على تنفيذ ThreadFactory للاستخدام
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        //إنشاء ThreadPoolExecutor
        ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue(2), threadFactory, rejectionHandler);
        //بدء موضوع المراقبة
        MyMonitorThread monitor = new MyMonitorThread(executorPool, 3);
        Thread monitorThread = new Thread(monitor);
        monitorThread.start();
        //تقديم العمل إلى حوض الموضوع
        for(int i=0; i<10; i++){
            executorPool.execute(new WorkerThread("cmd"+i));
        }
        
        Thread.sleep(30000);
        //إيقاف التشغيل الحوض
        executorPool.shutdown();
        //إيقاف تشغيل موضوع المراقبة
        Thread.sleep(5000);
        monitor.shutdown();
        
    }
}

يرجى ملاحظة أنه أثناء تهيئة ThreadPoolExecutor، نحن نحتفظ بحجم الحوض الأولي على أنه 2، والحد الأقصى لحجم الحوض على أنه 4، وحجم طابور العمل على أنه 2. لذا إذا كان هناك 4 مهام تعمل وتم تقديم مهام أخرى، سيحتفظ طابور العمل بـ 2 منها فقط وسيتم التعامل مع باقيها بواسطة RejectedExecutionHandlerImpl. هنا إخراج البرنامج السابق الذي يؤكد البيان أعلاه.

pool-1-thread-1 Start. Command = cmd0
pool-1-thread-4 Start. Command = cmd5
cmd6 is rejected
pool-1-thread-3 Start. Command = cmd4
pool-1-thread-2 Start. Command = cmd1
cmd7 is rejected
cmd8 is rejected
cmd9 is rejected
[monitor] [0/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 4, Completed: 0, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-4 End.
pool-1-thread-1 End.
pool-1-thread-2 End.
pool-1-thread-3 End.
pool-1-thread-1 Start. Command = cmd3
pool-1-thread-4 Start. Command = cmd2
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
[monitor] [4/2] Active: 2, Completed: 4, Task: 6, isShutdown: false, isTerminated: false
pool-1-thread-1 End.
pool-1-thread-4 End.
[monitor] [4/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [2/2] Active: 0, Completed: 6, Task: 6, isShutdown: false, isTerminated: false
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true
[monitor] [0/2] Active: 0, Completed: 6, Task: 6, isShutdown: true, isTerminated: true

لاحظ التغيير في عدد المهام النشطة والمكتملة وإجمالي عدد المهام المكتملة للمنفذ. يمكننا استدعاء الطريقة shutdown() لإنهاء تنفيذ جميع المهام المقدمة وإنهاء حوض الأسلاك. إذا كنت ترغب في جدولة مهمة للتشغيل مع تأخير أو بشكل دوري، فيمكنك استخدام فئة ScheduledThreadPoolExecutor. اقرأ المزيد حولهم في مخطط جافا لتنفيذ حوض الأسلاك.