ExecutorService và thread pool: chọn đúng loại cho từng bài toán

1. Tổng quan

Tạo new Thread() cho mỗi tác vụ thì dễ viết, nhưng khó kiểm soát khi tải tăng. ExecutorService giải quyết việc này bằng cách tái sử dụng thread và xếp hàng tác vụ. Lớp Executors cung cấp nhiều loại pool khác nhau, và chọn sai loại có thể gây OutOfMemoryError hoặc làm treo ứng dụng.

Bài này đi qua từng loại pool, cách ThreadPoolExecutor xử lý tác vụ bên trong, cách chọn kích thước pool và cách cấu hình trong Spring Boot 3.x. Đọc xong, bạn sẽ biết nên dùng pool nào cho tác vụ CPU-bound, I/O-bound hay tác vụ định kỳ.

2. ExecutorService cơ bản

Ví dụ sau dùng một pool đơn giản để chạy một tác vụ có trả về kết quả:

import java.util.concurrent.*;

public class OrderReportApp {

    public static void main(String[] args) throws Exception {
        try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
            Future<Integer> total = executor.submit(() -> countOrders("2026-10"));
            System.out.println("Tổng đơn: " + total.get());
        }
    }

    static int countOrders(String month) {
        return 128; // giả lập truy vấn database
    }
}

Tổng đơn: 128

Đoạn code hoạt động như sau:

• submit() nhận một Callable và trả về Future<Integer> ngay lập tức. Tác vụ chạy trên một thread của pool.
• total.get() chặn thread hiện tại cho tới khi có kết quả.
• Từ Java 19, ExecutorService implement AutoCloseable. Khi ra khỏi khối try, close() gọi shutdown() rồi chờ mọi tác vụ chạy xong.

Lưu ý: Nếu bạn không đóng executor, các thread không phải daemon sẽ giữ JVM sống. Chương trình main sẽ không thoát.

3. Các loại pool trong Executors

3.1. newFixedThreadPool(n)

Pool này có đúng n thread. Tác vụ dư được đưa vào một LinkedBlockingQueue không giới hạn.

ExecutorService invoiceExecutor = Executors.newFixedThreadPool(8);

Loại này phù hợp khi bạn muốn giới hạn số tác vụ chạy đồng thời, ví dụ tác vụ CPU-bound. Rủi ro nằm ở hàng đợi. Nếu producer đẩy tác vụ nhanh hơn tốc độ xử lý, queue phình mãi cho tới khi heap cạn.

3.2. newCachedThreadPool()

Pool này không có thread cố định. Nó dùng SynchronousQueue, tức là không có hàng đợi thật. Mỗi tác vụ mới được giao cho một thread rảnh nếu có, nếu không thì pool tạo thread mới. Thread rảnh quá 60 giây sẽ bị huỷ.

ExecutorService notificationExecutor = Executors.newCachedThreadPool();

Pool này hợp với nhiều tác vụ ngắn và tải thấp, không đều. Khi tải tăng đột biến, nó có thể tạo hàng nghìn platform thread. Mỗi thread chiếm khoảng 1 MB stack theo mặc định, nên ứng dụng dễ gặp OutOfMemoryError: unable to create native thread.

3.3. newSingleThreadExecutor()

Pool này chỉ có một thread, và tác vụ chạy tuần tự theo đúng thứ tự gửi vào.

ExecutorService auditLogExecutor = Executors.newSingleThreadExecutor();
auditLogExecutor.submit(() -> System.out.println("Order #1 created"));
auditLogExecutor.submit(() -> System.out.println("Order #1 paid"));

Order #1 created
Order #1 paid

Hai dòng log luôn in theo đúng thứ tự trên. Loại này hữu ích khi bạn cần ghi log, ghi file hoặc cập nhật trạng thái theo thứ tự mà không muốn dùng lock. Nếu thread chết vì exception, pool sẽ tạo thread mới thay thế. Hàng đợi của nó cũng không giới hạn.

3.4. newScheduledThreadPool(n)

Loại này dùng cho tác vụ hẹn giờ hoặc lặp lại:

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

scheduler.scheduleAtFixedRate(
        () -> System.out.println("Đồng bộ tồn kho lúc " + java.time.LocalTime.now()),
        0, 5, TimeUnit.SECONDS);

scheduleAtFixedRate chạy tác vụ mỗi 5 giây, tính từ thời điểm bắt đầu của lần chạy trước. Ngược lại, scheduleWithFixedDelay tính 5 giây từ lúc lần trước kết thúc. Bạn nên chọn scheduleWithFixedDelay khi thời gian chạy của tác vụ không ổn định.

Lưu ý: Nếu tác vụ định kỳ ném exception, mọi lần chạy sau sẽ bị huỷ âm thầm, không có log nào. Hãy luôn bọc thân tác vụ trong try/catch.

3.5. newWorkStealingPool()

Phương thức này trả về một ForkJoinPool với số thread bằng số CPU. Mỗi thread có hàng đợi riêng. Thread rảnh sẽ “ăn trộm” việc từ hàng đợi của thread khác.

ExecutorService pricingExecutor = Executors.newWorkStealingPool();

Loại này hợp với tác vụ CPU-bound có thể chia nhỏ đệ quy, ví dụ tính toán song song trên tập dữ liệu lớn. Thread của pool là daemon, và pool không đảm bảo thứ tự thực thi.

3.6. newVirtualThreadPerTaskExecutor() (Java 21)

Với loại này, mỗi tác vụ chạy trên một virtual thread mới. Virtual thread rất rẻ, nên bạn có thể tạo hàng trăm nghìn cái.

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        int customerId = i;
        executor.submit(() -> {
            Thread.sleep(100); // giả lập gọi HTTP tới CustomerService
            return customerId;
        });
    }
}

Đoạn code trên chạy 10.000 tác vụ chặn 100 ms và hoàn tất trong khoảng vài trăm ms. Nếu dùng fixed pool 10 thread, cùng khối lượng việc đó mất khoảng 100 giây.

Đây là lựa chọn mặc định tốt cho tác vụ I/O-bound như gọi HTTP, query database hay đọc file. Bạn cần nhớ ba điều:

• Đừng pool virtual thread. Chúng sinh ra để dùng một lần rồi bỏ.
• Muốn giới hạn tải lên tài nguyên ngoài thì dùng Semaphore. Ví dụ, khi database chỉ chịu 20 kết nối, đừng dựa vào kích thước pool để giới hạn.
• Ở Java 21, khối synchronized có thao tác chặn sẽ “pin” virtual thread vào carrier thread. Khi đó lợi ích của virtual thread mất đi. Hãy ưu tiên ReentrantLock ở những đoạn như vậy.

Ví dụ giới hạn số truy vấn đồng thời bằng Semaphore:

Semaphore dbPermits = new Semaphore(20);

executor.submit(() -> {
    dbPermits.acquire();
    try {
        return orderRepository.findByCustomerId(customerId);
    } finally {
        dbPermits.release();
    }
});

4. Tự cấu hình ThreadPoolExecutor

Các factory method trong Executors thường giấu đi một tham số nguy hiểm: queue không giới hạn hoặc số thread không giới hạn. Với hệ thống production, bạn nên tự khởi tạo ThreadPoolExecutor để kiểm soát mọi thứ:

ThreadFactory factory = Thread.ofPlatform().name("payment-", 0).factory();

ThreadPoolExecutor paymentExecutor = new ThreadPoolExecutor(
        4,                                   // corePoolSize
        8,                                   // maximumPoolSize
        60, TimeUnit.SECONDS,                // keepAlive cho thread vượt core
        new ArrayBlockingQueue<>(100),       // queue có giới hạn
        factory,                             // đặt tên: payment-0, payment-1...
        new ThreadPoolExecutor.CallerRunsPolicy());

Khi một tác vụ mới tới, ThreadPoolExecutor xử lý theo đúng thứ tự sau:

1. Nếu số thread nhỏ hơn corePoolSize, pool tạo thread mới.
2. Nếu không, pool đưa tác vụ vào queue.
3. Nếu queue đầy và số thread nhỏ hơn maximumPoolSize, pool tạo thêm thread.
4. Nếu đã chạm cả hai giới hạn, pool gọi RejectedExecutionHandler.

Hệ quả là maximumPoolSize chỉ có tác dụng khi queue đầy. Nếu bạn dùng queue không giới hạn, pool sẽ không bao giờ vượt quá corePoolSize. Đây là nguồn nhầm lẫn rất phổ biến.

Java cung cấp sẵn bốn rejection policy:

Policy	Hành vi
AbortPolicy (mặc định)	Ném RejectedExecutionException
CallerRunsPolicy	Thread gọi execute() tự chạy tác vụ, giúp làm chậm producer (back-pressure)
DiscardPolicy	Bỏ tác vụ, không báo gì
DiscardOldestPolicy	Bỏ tác vụ cũ nhất trong queue rồi thử lại

Đặt tên thread qua ThreadFactory giúp thread dump và log dễ đọc hơn nhiều. So sánh payment-3 với pool-7-thread-3 là thấy ngay sự khác biệt.

5. Chọn đúng loại

5.1. Kích thước pool

Với tác vụ CPU-bound, số thread nên xấp xỉ số core:

int cores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuExecutor = Executors.newFixedThreadPool(cores);

Nhiều thread hơn số core chỉ làm tăng chi phí context switch.

Với tác vụ I/O-bound chạy trên platform thread, bạn có thể dùng công thức gần đúng:

số thread ≈ số core × (1 + thời gian chờ / thời gian tính toán)

Ví dụ: máy có 8 core, mỗi request chờ I/O 90 ms và tính toán 10 ms. Kết quả là 8 × (1 + 9) = 80 thread. Đây chỉ là điểm xuất phát. Bạn cần đo đạc thực tế bằng load test.

5.2. Bảng quyết định

Bài toán	Lựa chọn
Tính toán nặng CPU	newFixedThreadPool(cores) hoặc ForkJoinPool
Gọi API, query DB số lượng lớn	newVirtualThreadPerTaskExecutor() + Semaphore
Cần thứ tự tuần tự	newSingleThreadExecutor()
Tác vụ định kỳ	newScheduledThreadPool(n)
Production, cần back-pressure	ThreadPoolExecutor với queue có giới hạn
Tải đột biến, không giới hạn	Tránh newCachedThreadPool()

6. Thread pool trong Spring Boot 3.x

Đầu tiên, bạn cần dependency Maven:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

Sau đó, khai báo một executor riêng cho việc gửi email:

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "emailExecutor")
    public ThreadPoolTaskExecutor emailExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("email-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        return executor;
    }
}

@Service
public class EmailService {

    @Async("emailExecutor")
    public void sendOrderConfirmation(String customerEmail) {
        System.out.println(Thread.currentThread().getName() + " gửi mail tới " + customerEmail);
    }
}

email-1 gửi mail tới an.nguyen@example.com

Vài điểm cần chú ý:

• queueCapacity mặc định của ThreadPoolTaskExecutor là Integer.MAX_VALUE. Nếu bạn không đặt giá trị này, maxPoolSize sẽ vô nghĩa, như đã giải thích ở mục 4.
• setWaitForTasksToCompleteOnShutdown(true) giúp email đang gửi không bị cắt ngang khi ứng dụng tắt.
• Nếu muốn dùng virtual thread cho @Async, Tomcat và @Scheduled, bạn chỉ cần thêm một dòng cấu hình:

spring.threads.virtual.enabled=true

7. Kiểm thử hành vi của pool

Test sau xác nhận đúng thứ tự xử lý đã mô tả ở mục 4: core thread trước, rồi tới queue, rồi tới thread bổ sung, cuối cùng là reject.

import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
import static org.assertj.core.api.Assertions.*;

class PaymentExecutorTest {

    @Test
    void shouldRejectWhenThreadsAndQueueAreFull() throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 2, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.AbortPolicy());

        Runnable blockingPayment = () -> {
            try {
                release.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        try {
            executor.execute(blockingPayment); // 1: tạo core thread
            executor.execute(blockingPayment); // 2: vào queue
            executor.execute(blockingPayment); // 3: queue đầy -> tạo thread thứ 2

            assertThat(executor.getPoolSize()).isEqualTo(2);
            assertThat(executor.getQueue()).hasSize(1);
            assertThatThrownBy(() -> executor.execute(blockingPayment))
                    .isInstanceOf(RejectedExecutionException.class);
        } finally {
            release.countDown();
            executor.shutdown();
            assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
        }
    }
}

CountDownLatch giữ mọi tác vụ ở trạng thái chặn, nhờ vậy trạng thái pool đứng yên khi bạn kiểm tra. Tác vụ thứ tư không còn chỗ, nên AbortPolicy ném exception. Khối finally thả latch ra và đóng pool, nên test không để lại thread rác.

8. Lỗi hay gặp

• Dùng Executors.newFixedThreadPool cho producer không kiểm soát. Queue không giới hạn sẽ phình ra rồi gây OOM. Hãy dùng ArrayBlockingQueue có giới hạn.
• Tạo pool mới trong mỗi request. Pool phải được dùng chung, ví dụ khai báo là bean hoặc field static final. Tạo rồi quên đóng sẽ làm rò rỉ thread.
• Nuốt exception. Với submit(), exception được giữ trong Future. Nếu bạn không gọi get(), bạn sẽ không bao giờ thấy lỗi. Hãy gọi get() hoặc bắt lỗi ngay trong tác vụ.
• Deadlock do tác vụ chờ tác vụ con trên cùng một pool nhỏ. Ví dụ: pool 2 thread, cả hai tác vụ cha đều gọi get() trên tác vụ con đang nằm trong queue. Không còn thread nào để chạy tác vụ con, nên cả hệ thống đứng im.
• Đặt ThreadLocal nặng vào virtual thread. Bạn có thể có hàng triệu virtual thread, mỗi cái giữ một bản sao, nên bộ nhớ tăng rất nhanh.
• Gọi shutdownNow() mà tác vụ không xử lý interrupt. Khi đó tác vụ vẫn chạy tiếp. Tác vụ dài nên kiểm tra Thread.currentThread().isInterrupted().

Lưu ý: Không nên dùng ExecutorService cho luồng xử lý phức tạp nhiều bước phụ thuộc nhau. Trường hợp đó nên dùng CompletableFuture, vốn vẫn có thể chạy trên executor bạn tự cấu hình.

9. Kết luận

Mỗi loại thread pool giải quyết một bài toán riêng:

• Fixed pool hoặc ForkJoinPool cho tác vụ CPU-bound.
• Virtual thread kèm Semaphore cho tác vụ I/O-bound trên Java 21.
• Single thread executor khi cần thứ tự.
• Scheduled pool cho tác vụ định kỳ.

Trong production, hãy tự cấu hình ThreadPoolExecutor hoặc ThreadPoolTaskExecutor với queue có giới hạn, rejection policy rõ ràng và tên thread dễ nhận biết. Cuối cùng, bạn nên kiểm chứng kích thước pool bằng load test thay vì chỉ dựa vào công thức.