ThreadPoolExecutor – דוגמה לבריכת תהליכים ב-Java

הבריכה של החוטים ב- Java ניהול הבריכה של החוטים העובדים. זה מכיל תור ששומר משימות הממתינות לביצוע. אנו יכולים להשתמש ב- ThreadPoolExecutor כדי ליצור בריכת חוטים ב-Java. בריכת חוטים ב-Java ניהול את אוסף החוטים הרצים. החוטים העובדים מבצעים את החוטים הרצים מהתור. java.util.concurrent.Executors מספק יצרנים ושיטות תמיכה עבור java.util.concurrent.Executor ממשק כדי ליצור את בריכת החוטים ב-Java. 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, שבה אנו יוצרים בריכת חוטים קבועה מ- מסגרת ה- Executors.

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. קרא עוד עליהם ב-מתזמן Java לבריכת תהליכים לניהול משימות.