Thread pool: tái sử dụng thread cho đúng cách

Bài trước Order Service đã dùng Executors.newFixedThreadPool(10) để gọi User, Inventory, Shipping song song. Giờ mình mổ xẻ cái pool đó: nó chạy thế nào, cấu hình ra sao cho production.

1. Bài toán

• Sau khi tạo đơn, Order Service phải gửi email xác nhận (~500ms mỗi email).
• Giờ sale, 1.000 đơn/phút. Không muốn request phải chờ gửi email xong.
• Cách đầu tiên ai cũng nghĩ: new Thread(() -> sendEmail(order)).start().

2. Vấn đề

Chạy được, nhưng khi tải tăng sẽ có hai vấn đề:

• Tốn chi phí tạo/hủy: mỗi thread là một OS thread, tạo ra rồi bỏ đi cho mỗi task là phí.
• Không giới hạn: 10.000 đơn đến cùng lúc là 10.000 thread, dễ hết bộ nhớ (OutOfMemoryError: unable to create native thread) hoặc nghẽn vì context switch.

new Thread mỗi task:
order 1 → tạo T1 → chạy → hủy
order 2 → tạo T2 → chạy → hủy
...
order 10000 → T10000   ← 10.000 stack, CPU chỉ 8 core

3. Giải pháp: Thread pool

Tạo sẵn một số thread cố định, task đến thì bỏ vào hàng đợi, thread rảnh lấy ra chạy, chạy xong quay lại lấy task tiếp. Task (việc) và thread (người làm) tách rời: 1000 task, chỉ 10 thread.

Mọi factory trong Executors đều tạo ra ThreadPoolExecutor với các tham số:

• corePoolSize: số thread giữ thường trực.
• maximumPoolSize: số thread tối đa khi quá tải.
• keepAliveTime: thread vượt core rảnh quá lâu thì bị hủy.
• workQueue: hàng đợi task.
• threadFactory: đặt tên thread.
• RejectedExecutionHandler: làm gì khi hết chỗ.

4. Code từng bước

Bước 1: bản đơn giản.

ExecutorService pool = Executors.newFixedThreadPool(10);
for (int i = 0; i < 1000; i++) {
    int id = i;
    pool.submit(() -> sendEmail(id));   // 1000 task, chỉ 10 thread
}

Bước 2: tự cấu hình đầy đủ.

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        4,                                   // corePoolSize
        8,                                   // maximumPoolSize
        60, TimeUnit.SECONDS,                // keepAliveTime cho thread vượt core
        new ArrayBlockingQueue<>(100),       // workQueue có giới hạn
        Executors.defaultThreadFactory(),    // threadFactory
        new ThreadPoolExecutor.CallerRunsPolicy()); // rejection handler

Bước 3: trong Spring Boot, khai báo bean có tên và giới hạn rõ ràng.

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

@Service
public class NotificationService {
    @Async("emailExecutor")
    public void sendOrderConfirm(String email) { ... }
}

Bước 4: tắt pool đúng cách (khi tự tạo, không phải bean Spring). Pool không tự tắt, thread của pool là non-daemon nên quên shutdown thì JVM có thể không thoát.

pool.shutdown();                         // không nhận task mới, chạy nốt task trong queue
try {
    if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
        pool.shutdownNow();              // quá hạn: interrupt thread đang chạy, trả task chưa chạy
    }
} catch (InterruptedException e) {
    pool.shutdownNow();
    Thread.currentThread().interrupt();
}

shutdownNow() chỉ gửi interrupt, không giết thread ngay; task phải tự kiểm tra interrupt (hoặc đang ở sleep/wait) thì mới dừng. Java 19+ ExecutorService là AutoCloseable, dùng được try-with-resources. Bean ThreadPoolTaskExecutor thì Spring tự shutdown khi context đóng.

Bước 5: test. Gọi API tạo đơn, xem log thread có tên email-1, email-2…; chụp thread dump bằng jcmd <pid> Thread.print để thấy pool.

5. Cách hoạt động

Thứ tự xử lý khi có task mới, đây là chỗ hay bị hỏi:

New Task
  ↓
worker < core? ── YES → tạo thread
  │ NO
queue còn chỗ? ── YES → vào queue
  │ NO
worker < max?  ── YES → tạo thêm thread
  │ NO
REJECT → RejectedExecutionHandler

Nhiều người tưởng đủ core thì tạo thêm thread ngay. Không phải: queue đầy rồi mới tăng lên max. Ví dụ core=5, max=10, queue=100, 200 task cùng vào (chưa task nào xong):

Task 1–5     → tạo 5 core thread, chạy ngay
Task 6–105   → vào queue (100/100)
Task 106–110 → queue đầy, workers 5 < max 10 → tạo thêm thread 6..10
Task 111–200 → queue đầy + đã max → REJECT (90 task)

Hệ quả: nếu queue quá lớn (ví dụ 10.000) thì max gần như không bao giờ được dùng, pool cứ chạy 5 thread và task chờ rất lâu. Khi traffic giảm, các thread vượt core rảnh quá keepAliveTime bị hủy, pool quay về 5.

6. So sánh

Rejection policy:

Vì sao nhiều team cấm Executors.newXxx():

Pool size bao nhiêu?

• CPU-bound (tính toán, nén, mã hóa): khoảng số core hoặc số core + 1. Nhiều hơn chỉ tốn context switch.
• I/O-bound (gọi DB, HTTP): thread phần lớn thời gian ngồi chờ, nên cần nhiều hơn: số core × (1 + thời gian chờ / thời gian tính). Ví dụ chờ 90ms, tính 10ms, 8 core → 8 × 10 = 80 thread.

Công thức chỉ là điểm bắt đầu, phải đo bằng load test. Nhớ giới hạn phía sau: pool 80 thread mà DB connection pool chỉ có 10 thì 70 thread vẫn ngồi chờ connection. Với I/O-bound trên Java 21, virtual thread thường là lựa chọn đơn giản hơn.

7. Khi nào nên dùng / không nên dùng

• Nên: mọi chỗ cần chạy nhiều task nền hoặc song song. Pool là static/bean dùng chung, không tạo mới mỗi lần gọi function.
• Tách pool theo loại việc: email riêng, báo cáo riêng, để một loại chậm không kéo chết loại khác.
• Pool không làm code thread-safe: 3 worker cùng counter++ vẫn race condition (bài 3.4).
• Cẩn thận ThreadLocal: thread trong pool không chết, giá trị cũ còn lại cho task sau. Luôn remove() trong finally. Chiều ngược lại, task submit sang pool không thấy ThreadLocal của thread request (SecurityContextHolder, MDC).

Request A → Thread-1 → currentUser = User A
request A xong, Thread-1 KHÔNG chết
Request B → Thread-1 được reuse → currentUser vẫn là User A ❌

CURRENT_USER.set(user);
try { handle(); } finally { CURRENT_USER.remove(); }

Đúng chủ đề rồi. Đi từng dòng:

1. newFixedThreadPool(2): tạo pool tối đa 2 worker + 1 queue. Worker được tạo dần khi có task.
2. Callable task = ...: chưa vào queue, chưa chạy. Chỉ là object mô tả việc.
3. executor.submit(task): đây mới là lúc bắt đầu. Bọc thành FutureTask, rồi theo đúng flow ở trên: còn dưới core thì tạo worker chạy ngay (không qua queue), đủ core và đều bận thì vào queue. Trả Future về gần như ngay lập tức.
4. future.get(): thread gọi chờ tới khi worker chạy xong call(). Nếu đã xong thì lấy luôn.
5. shutdown(): không nhận task mới, worker làm nốt rồi kết thúc.

main ── submit ──► [worker < core?] ─YES─► worker-1 chạy call() ngay
                        │ NO (2 worker đều bận)
                        ▼
                  Task Queue [task3, task4...] ──► worker rảnh lấy ra chạy
main ── get() ── WAITING ...................... ◄── worker xong, đánh thức main

Vậy "add vào queue" không phải lúc nào cũng xảy ra. Task chỉ ngồi queue khi các worker đang bận.

Đúng, đó là ExecutorService.invokeAll(). Rất hợp với bài doanh thu 10 cửa hàng ở bài trước: 10 Callable, pool 5 thread, phải chờ cả 10 xong.

record StoreRevenue(int storeId, double revenue) {}

public static double calculateTotalRevenue() throws Exception {
    List<Callable<StoreRevenue>> tasks = new ArrayList<>();
    for (int storeId = 1; storeId <= 10; storeId++) {
        int id = storeId;
        tasks.add(() -> new StoreRevenue(id, getRevenue(id)));
    }
    // submit tất cả và BLOCK tới khi cả 10 task xong
    List<Future<StoreRevenue>> futures = EXECUTOR.invokeAll(tasks);
    double total = 0;
    for (Future<StoreRevenue> f : futures) {   // cùng thứ tự với tasks
        total += f.get().revenue();              // không còn phải chờ
    }
    return total;
}

Còn có invokeAny(): trả kết quả của task xong đầu tiên, hủy phần còn lại.

Đúng, là hai pool khác nhau trong cùng một JVM. Chỉ chỉnh một chút: Tomcat không hẳn "bọc application". Trong Spring Boot, Tomcat là HTTP server embedded chạy cùng JVM; Tomcat worker gọi vào Spring MVC → Controller → Service. ExecutorService là pool riêng, optional, do mình tạo.

Clients
  │ HTTP
  ▼
┌──────────────────────── Order Service JVM ────────────────────────┐
│ Tomcat pool (http-nio-8080-exec-1..200)                           │
│      │                                                            │
│      ▼                                                            │
│ DispatcherServlet → OrderController → OrderService                │
│                                          │ submit / @Async        │
│                                          ▼                        │
│                       Application pool (email-1..8, pool-1-...)   │
└───────────────────────────────────────────────────────────────────┘

Lưu ý quan trọng: nếu Tomcat thread submit sang pool của mình rồi gọi future.get() ngay, Tomcat thread vẫn bị giữ, giờ thành 2 thread cùng dính vào một request. Pool riêng có ích khi chạy song song nhiều việc (như checkout ở bài 3.2) hoặc chạy nền không chờ (email). Còn để Tomcat thread được nhả ra thật sự thì cần async servlet / CompletableFuture trả từ controller, sẽ học ở bài 3.9.

Câu 1: Thread pool là gì, vì sao cần?

🇻🇳 Tiếng Việt:

Thread pool giữ sẵn một số thread để tái sử dụng cho nhiều task, tránh chi phí tạo/hủy thread và giới hạn số thread chạy cùng lúc. Task được đưa vào queue, worker rảnh lấy ra chạy. Pool không làm code thread-safe, và vì thread được reuse nên ThreadLocal phải được remove sau mỗi task.

🇬🇧 English:

A thread pool keeps a set of reusable threads for many tasks, avoiding thread creation cost and bounding concurrency. Tasks go into a queue and idle workers pick them up. A pool does not make code thread-safe, and since threads are reused, ThreadLocal values must be removed after each task.

Câu 2: ThreadPoolExecutor xử lý task mới thế nào?

🇻🇳 Tiếng Việt:

Có core, max, keepAlive, queue và rejection policy. Task mới: chưa đủ core thì tạo thread, đủ core thì vào queue, queue đầy mới tăng tới max, vẫn đầy thì reject. Nên queue quá lớn thì max gần như vô dụng.

🇬🇧 English:

It has core size, max size, keep-alive, a work queue and a rejection policy. A new task creates a thread if below core, otherwise goes to the queue; only when the queue is full does the pool grow to max, and beyond that the task is rejected. So with a huge queue, max is practically never reached.

Câu 3: Cấu hình pool thế nào trong production?

🇻🇳 Tiếng Việt:

Em tránh Executors.newFixedThreadPool và newCachedThreadPool vì queue hoặc số thread không giới hạn. Em tự cấu hình queue có giới hạn, đặt tên thread, chọn policy như CallerRunsPolicy để có backpressure, tách pool theo loại việc. Size: CPU-bound khoảng số core, I/O-bound nhiều hơn theo tỉ lệ chờ/tính, không vượt quá xa DB connection pool, rồi đo bằng load test. Khi tắt thì shutdown + awaitTermination.

🇬🇧 English:

I avoid Executors.newFixedThreadPool and newCachedThreadPool because of their unbounded queue or thread count. I configure a bounded queue, named threads, a policy like CallerRunsPolicy for backpressure, and separate pools per workload. Sizing: about the core count for CPU-bound work, more for I/O-bound based on wait/compute ratio, not far above the DB connection pool, then verified with load tests. On shutdown I use shutdown plus awaitTermination.

Cách nhớ ngắn gọn

• Pool = đội thợ cố định + hàng đợi việc.
• core → queue → max → reject (queue đầy mới tăng max).
• Queue phải có giới hạn; CallerRuns = tự giảm tốc.
• CPU-bound ≈ số core; I/O-bound nhiều hơn, nhưng nhìn DB pool.
• Tomcat pool ≠ pool của mình; thread reuse → nhớ remove ThreadLocal.