Dùng ThreadPoolExecutor trong Java để tạo Thread pool

Reading Time: 5 minutes

Thread pool trong Java quản lý một nhóm các worker thread (luồng xử lý công việc cụ thể). Nó chứa một queue (hàng đợi) để giữ các task đang chờ được thực thi.

Java thread pool quản lý một tập hợp các thread Runnable, và các worker thread sẽ thực thi các thread Runnable này từ queue. Để tạo thread pool trong Java, ta có thể dùng ThreadPoolExecutor. Đây là một class tiện ích (utility class) cung cấp các phương thức factory (dùng để sinh đối tượng mới theo mẫu thiết kế Factory). Nó hỗ trợ cho interface java.util.concurrent.Executor và cung cấp các phương thức hữu ích để làm việc với các class ExecutorService, ScheduledExecutorService, ThreadFactory, và Callable thông qua nhiều phương thức factory khác nhau.

Để hiểu rõ cách hoạt động của Executors, chúng ta hãy cùng tạo một chương trình ví dụ đơn giản. Trước hết, ta cần một class Runnable tên là 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;
    }
}

Ví dụ ExecutorService

Dưới đây là class chương trình test ở file SimpleThreadPool.java, nơi chúng ta tạo một thread pool có kích thước cố định (fixed thread pool) bằng Executors framework.

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");
    }
}

Trong chương trình này, ta tạo một thread pool với kích thước cố định là 5 worker thread. Sau đó, ta gửi (submit) 10 job vào pool này. Vì pool chỉ có 5 thread, nó sẽ bắt đầu xử lý 5 job đầu tiên, trong khi các job còn lại sẽ ở trong hàng đợi. Ngay khi một job hoàn thành, một worker thread sẽ lấy job tiếp theo từ hàng đợi ra để thực thi.

Dưới đây là kết quả output của chương trình.

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

Kết quả trên xác thực rằng ta có năm thread trong pool (được đặt tên từ “pool-1-thread-1” đến “pool-1-thread-5”) và chúng chịu trách nhiệm thực thi các task đã được gửi vào.

Ví dụ ThreadPoolExecutor

Class Executors cung cấp một cách triển khai ExecutorService đơn giản bằng ThreadPoolExecutor, nhưng ThreadPoolExecutor thực chất cung cấp nhiều tính năng hơn thế. Ta có thể chỉ định số lượng thread sẽ hoạt động khi tạo thực thể ThreadPoolExecutor, giới hạn kích thước của thread pool, và tạo một triển khai RejectedExecutionHandler của riêng mình để xử lý các job không vừa với worker queue.

Dưới đây là một triển khai tuỳ chỉnh của chúng ta cho interface 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 cung cấp nhiều phương thức để chúng ta có thể theo dõi trạng thái hiện tại của executor, như kích thước pool, số thread đang hoạt động, và số lượng task. Vì vậy, ta có thể tạo một monitor thread (thread giám sát) để in ra thông tin của executor sau mỗi khoảng thời gian nhất định.

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();
                }
        }
            
    }
}

Dưới đây là ví dụ triển khai thread pool bằng 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 implementation
        RejectedExecutionHandlerImpl rejectionHandler = new RejectedExecutionHandlerImpl();
        //Get the ThreadFactory implementation to use
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        //creating the ThreadPoolExecutor
        ThreadPoolExecutor executorPool = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), threadFactory, rejectionHandler);
        //start the monitoring thread
        MyMonitorThread monitor = new MyMonitorThread(executorPool, 3);
        Thread monitorThread = new Thread(monitor);
        monitorThread.start();
        //submit work to the thread pool
        for(int i=0; i<10; i++){
            executorPool.execute(new WorkerThread("cmd"+i));
        }
        
        Thread.sleep(30000);
        //shut down the pool
        executorPool.shutdown();
        //shut down the monitor thread
        Thread.sleep(5000);
        monitor.shutdown();
        
    }
}

Lưu ý rằng khi khởi tạo ThreadPoolExecutor, chúng ta giữ kích thước pool ban đầu là 2, kích thước tối đa là 4, và kích thước work queue là 2. Vì vậy, nếu có 4 task đang chạy và có thêm task được gửi đến, work queue sẽ chỉ giữ 2 trong số đó, và các task còn lại sẽ được xử lý bởi RejectedExecutionHandler.

Kết quả output dưới đây xác nhận điều này.

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

Hãy chú ý sự thay đổi trong số lượng task đang hoạt động (active), đã hoàn thành (completed) và tổng số task đã hoàn thành của executor. Chúng ta có thể gọi phương thức shutdown() để hoàn tất việc thực thi tất cả các task đã gửi và chấm dứt thread pool. Nếu bạn muốn lập lịch cho một task chạy có độ trễ hoặc chạy định kỳ, bạn có thể sử dụng class ScheduledThreadPoolExecutor.

Tổng kết

Như bạn đã thấy, ThreadPoolExecutor giúp ta tạo thread pool trong Java và quản lý một pool các luồng một cách linh hoạt. Nó cho phép ta điều chỉnh số lượng luồng chạy đồng thời, quản lý công việc đang chờ, đồng thời xử lý các tác vụ không được nhận khi quá tải. Nếu bạn có thắc mắc gì hoặc cần hỗ trợ thêm, hãy để lại câu hỏi trong phần bình luận nhé!