Giới thiệu

Thread Pool là thành phần quan trọng nhất trong bộ công cụ xử lý đồng thời của Java. Hầu hết các ứng dụng yêu cầu thực thi tác vụ bất đồng bộ hoặc song song đều sử dụng Thread Pool. Việc sử dụng Thread Pool mang lại ba lợi ích chính:

• Giảm thiểu tài nguyên: Tái sử dụng các thread đã được tạo ra, tránh chi phí cho việc tạo và hủy thread
• Tăng tốc độ phản hồi: Không cần chờ đợi quá trình khởi tạo thread mới để thực thi tác vụ
• Quản lý tập trung: Thread là tài nguyên giới hạn, việc tạo không kiểm soát sẽ ảnh hưởng đến hiệu suất hệ thống

Cơ chế hoạt động của Thread Pool

2.1. Các thành phần cơ bản

Trước khi tìm hiểu chi tiết, chúng ta cần nắm vững ba thành phần chính của Thread Pool:

Core Pool (Hồ bơi lõi): Đây là số lượng thread tối thiểu được duy trì trong Thread Pool. Khi khởi tạo, Thread Pool sẽ tạo các thread này theo nhu cầu. Có thể gọi phương thức prestartAllCoreThreads() để khởi tạo trước tất cả các thread lõi.

Task Queue (Hàng đợi tác vụ): Khi tất cả các thread lõi đang bận, các tác vụ mới sẽ được đưa vào hàng đợi để chờ xử lý.

Saturation Policy (Chính sách bão hòa): Áp dụng khi Thread Pool không thể tiếp nhận thêm tác vụ mới.

2.2. Quy trình xử lý tác vụ

Khi một tác vụ mới được gửi đến Thread Pool, quy trình xử lý diễn ra theo ba bước sau:

Bước 1: Kiểm tra xem Core Pool đã đầy chưa. Nếu chưa đầy, tạo một thread mới để thực thi tác vụ (ngay cả khi có thread rảnh rỗi). Chỉ khi số lượng thread đạt mức tối đa của Core Pool mới chuyển sang bước tiếp theo.

Bước 2: Kiểm tra hàng đợi tác vụ đã đầy chưa. Nếu chưa đầy, đưa tác vụ vào hàng đợi và chờ đợi. Các thread trong Core Pool sẽ lấy tác vụ từ hàng đợi sau khi hoàn thành tác vụ hiện tại.

Bước 3: Nếu hàng đợi đã đầy, kiểm tra xem số lượng thread đang hoạt động có đạt mức tối đa không. Nếu chưa đạt, tạo thread mới để thực thi tác vụ. Nếu đã đạt mức tối đa, áp dụng saturation policy.

2.3. Cấu hình Thread Pool

Thread Pool được tạo thông qua lớp ThreadPoolExecutor với bảy tham số cấu hình:

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

Các tham số bao gồm:

• corePoolSize: Số lượng thread tối thiểu trong Pool
• maximumPoolSize: Số lượng thread tối đa được phép tạo
• keepAliveTime: Thời gian giữ thread hoạt động khi không có tác vụ
• unit: Đơn vị thời gian cho keepAliveTime
• workQueue: Hàng đợi lưu trữ tác vụ chờ xử lý
• threadFactory: Factory tạo thread mới
• handler: Xử lý khi Thread Pool bị bão hòa

2.4. Phương thức gửi tác vụ

Có hai phương thức chính để gửi tác vụ đến Thread Pool:

execute(Runnable command): Dùng cho tác vụ không cần trả về kết quả. Không thể kiểm tra trạng thái thực thi.

submit(Callable<T> task): Dùng cho tác vụ cần trả về kết quả. Trả về đối tượng Future, cho phép kiểm tra trạng thái và lấy kết quả. Phương thức get() sẽ chặn luồng hiện tại cho đến khi tác vụ hoàn thành.

2.5. Cách dừng Thread Pool

Có hai phương thức để đóng Thread Pool:

shutdown(): Đặt trạng thái Thread Pool thành SHUTDOWN, ngừng tiếp nhận tác vụ mới và dừng các thread đang chờ.

shutdownNow(): Đặt trạng thái thành STOP, cố gắng dừng tất cả các thread đang thực thi và trả về danh sách tác vụ chưa được xử lý.

Để đảm bảo tất cả tác vụ được hoàn thành, nên sử dụng phương thức shutdown().

2.6. Giám sát Thread Pool

Thread Pool cung cấp các phương thức để theo dõi trạng thái:

• getTaskCount(): Tổng số tác vụ cần thực thi
• getCompletedTaskCount(): Số tác vụ đã hoàn thành
• getLargestPoolSize(): Số lượng thread lớn nhất từng được tạo
• getPoolSize(): Số lượng thread hiện tại
• getActiveCount(): Số thread đang hoạt động

Hàng đợi tác vụ và Chính sách bão hòa

3.1. Các loại hàng đợi

BlockingQueue được sử dụng làm hàng đợi trong Thread Pool, có bốn loại chính:

• ArrayBlockingQueue: Hàng đợi có giới hạn dựa trên mảng, sắp xếp theo nguyên tắc FIFO
• LinkedBlockingQueue: Hàng đợi không giới hạn dựa trên LinkedList, hiệu suất cao hơn ArrayBlockingQueue
• SynchronousQueue: Hàng đợi không lưu trữ phần tử, mỗi thao tác chèn phải đợi thao tác lấy
• PriorityBlockingQueue: Hàng đợi ưu tiên không giới hạn

3.2. Các loại saturation policy

Khi Thread Pool không thể xử lý thêm tác vụ, các chính sách sau được áp dụng:

• AbortPolicy: Ném exception RejectedExecutionException (chính sách mặc định)
• CallerRunsPolicy: Thực thi tác vụ bằng thread của người gọi
• DiscardOldestPolicy: Bỏ qua tác vụ cũ nhất trong hàng đợi và thực thi tác vụ hiện tại
• DiscardPolicy: Bỏ qua tác vụ mà không có thông báo

Ví dụ thực hành

Dưới đây là ví dụ minh họa cơ chế hoạt động của Thread Pool:

public class WorkerTask implements Runnable {
    private int taskId;
    
    public WorkerTask(int id) {
        this.taskId = id;
    }
    
    @Override
    public void run() {
        try {
            Thread.sleep(2000);
            System.out.println("Thread " + Thread.currentThread().getName() + 
                             " xu ly tac vu: " + taskId);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Ví dụ cấu hình Thread Pool và thực thi tác vụ:

public class ThreadPoolDemo {
    public static void main(String[] args) {
        // Cau hinh: Core Pool = 2, Max Pool = 4, Queue = 6
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 
            4, 
            10, 
            TimeUnit.MILLISECONDS, 
            new ArrayBlockingQueue<>(6)
        );
        
        // Gui 10 tac vu
        for (int i = 0; i < 10; i++) {
            executor.execute(new WorkerTask(i));
        }
        
        executor.shutdown();
    }
}

Kết quả thực thi với 10 tác vụ: hai thread lõi xử lý hai tác vụ đầu, sáu tác vụ tiếp theo vào hàng đợi, và hai thread còn lại xử lý hai tác vụ cuối.

Khi tăng số tác vụ lên 12, saturation policy sẽ được kích hoạt cho hai tác vụ vượt quá dung lượng.