Executor Framework: Thread pool và thuật toán nhận task

TL;DR: Ngừng new Thread() — đóng gói công việc thành task (Runnable/Callable) rồi giao cho ExecutorService lo thread pool và vòng đời. Bên dưới gần như mọi pool là ThreadPoolExecutor với thuật toán nhận task phản trực giác: tạo core thread → xếp queue → tạo thêm thread tới maximumPoolSize → từ chối. Hệ quả cốt lõi: queue unbounded làm maximumPoolSize vô nghĩa, nên newFixedThreadPool và newCachedThreadPool đều thiếu trần ở một chiều — production cần giới hạn hữu hạn ở mọi chiều. Còn số thread bao nhiêu thì tùy bản chất task: CPU-bound xấp xỉ số core, I/O-bound cần nhiều hơn hẳn theo tỉ số thời gian chờ trên thời gian tính.

1. Vì sao nên ngừng new Thread()?

Suốt phần vừa rồi của series, mọi câu hỏi đều xoay quanh state: nhiều thread cùng chạm một dữ liệu mà không phá vỡ invariant. Hai bài synchronizer (16, 16b) khép lại phần ấy, nhưng chưa nói ai chạy công việc, trên thread nào, và bao nhiêu thread — trục mới của series: task execution.

Luận điểm trung tâm: ngừng tự tạo thread. Bài Process và Thread đã cho thấy platform thread không miễn phí — tốn stack, tốn context switching, áp lực lên scheduler. Một server đặt vé cứ mỗi request lại new Thread(task).start() sẽ sụp dưới tải thật: mười nghìn request thành mười nghìn thread tranh nhau vài chục core, CPU trôi vào chuyển ngữ cảnh, JVM ném OutOfMemoryError vì không cấp nổi stack. Nó còn trộn hai thứ vốn nên tách rời — mô tả công việc và chính sách chạy công việc.

Executor Framework (Java 5, gói java.util.concurrent) chính là cú tách đôi đó: một bên là task — Runnable hoặc Callable; bên kia là chính sách thực thi đóng gói trong một Executor — interface nhỏ đến bất ngờ:

public interface Executor {
    void execute(Runnable command);
}

Chỉ một method, không nói gì về thread, pool hay hàng đợi. Sự nghèo nàn ấy cố ý: Executor tách việc gửi task khỏi việc chạy task — người gửi không cần biết task chạy trên thread mới, thread của pool, hay ngay thread gọi.

Executor đủ cho "bắn rồi quên", nhưng đời thực còn cần biết task xong chưa, lấy kết quả, hủy nó, tắt hệ thống cho gọn. ExecutorService thêm submit (nhận cả Callable lẫn Runnable, trả về một Future — tay cầm tới kết quả tương lai), invokeAll/invokeAny cho cả mẻ task, và bộ vòng đời shutdown/shutdownNow/awaitTermination.

ExecutorService pool = Executors.newFixedThreadPool(4);
pool.execute(() -> log.info("fire-and-forget"));          // Runnable, khong ket qua
Future<Booking> handle = pool.submit(() ->                // Callable, co ket qua
        bookingService.book("concert-01", "user-42"));

execute và submit khác nhau không chỉ ở giá trị trả về mà ở một bẫy exception. execute để exception của task nổi lên UncaughtExceptionHandler — bạn thấy stack trace trong log. submit nuốt exception vào Future, chỉ phát lại dưới dạng ExecutionException khi gọi future.get(). submit rồi không bao giờ get thì một lỗi nghiêm trọng biến mất không dấu vết — cách phổ biến khiến lỗi concurrency ẩn mình, và bài Future & CompletableFuture sẽ mổ tiếp.

Từ Java 19, ExecutorService là AutoCloseable nên dùng được trong try-with-resources: close tự gọi shutdown rồi chờ task chạy nốt. Pool sống suốt vòng đời ứng dụng vẫn cần shutdown thủ công — chuyện vận hành ấy để dành cho bài 17b — Vận hành pool production.

2. ThreadPoolExecutor: cỗ máy bên dưới

Hầu hết ExecutorService ta dùng hằng ngày, dù tạo qua factory nào, đều là một ThreadPoolExecutor. Hiểu nó là hiểu gần như mọi thread pool trong JVM; cách tốt nhất là nhìn constructor đầy đủ — mỗi tham số là một nút vặn của chính sách thực thi:

new ThreadPoolExecutor(
        int corePoolSize,                  // so thread giu thuong truc
        int maximumPoolSize,               // tran so thread
        long keepAliveTime, TimeUnit unit, // thoi gian song cua thread vuot core khi ranh
        BlockingQueue<Runnable> workQueue, // noi task cho khi khong co thread ranh
        ThreadFactory threadFactory,       // cach tao thread
        RejectedExecutionHandler handler); // lam gi khi khong nhan noi task

workQueue chính là BlockingQueue mà bài 15 — Blocking queues đã mổ: pool là một hệ producer–consumer, người submit là producer, các worker thread là consumer, và queue là vùng đệm giữa hai bên.

Bảy tham số ấy quy về một thuật toán quyết định duy nhất, chạy mỗi lần có task mới. Pool không hỏi "có thread nào rảnh không": nó tạo core thread trước — ngay cả khi đang có thread rảnh, miễn số thread còn dưới corePoolSize; đủ core rồi thì xếp task vào workQueue; queue đầy mới tạo thêm thread tới maximumPoolSize; cả hai kịch trần thì task bị từ chối, RejectedExecutionHandler vào cuộc.

Thứ tự "tạo core thread → xếp hàng → tạo thêm thread → từ chối" phản trực giác đến mức là nguồn của vô số hiểu nhầm: nhiều người tưởng pool bung tới maximumPoolSize rồi mới xếp hàng, thực tế queue được ưu tiên. Hệ quả trực tiếp: queue unbounded làm maximumPoolSize vô nghĩa — queue không bao giờ đầy nên pool không bao giờ tạo quá corePoolSize thread.

keepAliveTime lo phần co lại: thread vượt corePoolSize mà ngồi không quá keepAliveTime sẽ tự kết thúc, đưa pool về kích thước core khi tải hạ. Hình dung quán cà phê: corePoolSize là nhân viên ca cố định, workQueue là hàng khách chờ, maximumPoolSize là trần khi gọi thêm người thời vụ giờ cao điểm, keepAliveTime là khoảng vắng đủ lâu để họ về.

Booking pool của TicketFlow v3 dựng thẳng từ constructor để chọn từng nút vặn có chủ đích:

new ThreadPoolExecutor(
        workerCount,                          // core = max: pool kich thuoc co dinh
        workerCount,
        0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(queueCapacity),       // queue BOUNDED — chu y
        new NamedThreadFactory("booking-worker"),      // ten thread cho thread dump
        new ThreadPoolExecutor.CallerRunsPolicy());    // backpressure khi qua tai

Hai nút cuối — ThreadFactory (đặt tên thread để đọc được thread dump) và RejectedExecutionHandler (điều xảy ra khi pool đầy) — là chuyện vận hành, ta để dành cho bài 17b. Ở đây, hãy chú ý điều quan trọng nhất: ArrayBlockingQueue là queue bounded. Vì sao lại cố tình giới hạn nó, là câu hỏi của mục kế.

3. Khi nào KHÔNG nên dùng Executors factory?

Executors cung cấp factory tạo sẵn các cấu hình ThreadPoolExecutor thông dụng — gọn, và chính sự tiện lợi đó khiến chúng bị lạm dụng.

3.1 Ba factory thông dụng

newFixedThreadPool(n) tạo pool đúng n thread thường trực, queue unbounded: task vượt n thread đang bận xếp hàng vô tận. newSingleThreadExecutor là trường hợp n = 1 — một thread xử lý mọi task theo thứ tự gửi vào, cho serial execution mà vẫn hưởng task/Future (một dạng confinement). newCachedThreadPool ngược cực: corePoolSize 0, maximumPoolSize Integer.MAX_VALUE, SynchronousQueue không chứa nổi phần tử nào — nên task không gặp thread rảnh làm pool tạo thread mới ngay, thread rảnh quá 60 giây thì bị thu hồi.

3.2 Vì sao tránh newCachedThreadPool và unbounded queue trong production

Hai factory này hỏng theo hai cách đối xứng, cùng một gốc — không có trần. newCachedThreadPool không giới hạn số thread: khi request đến nhanh hơn tốc độ xử lý (kịch bản tải đỉnh), pool cứ tạo thêm thread, mỗi thread một stack, tới khi JVM kiệt bộ nhớ. Một cơn tăng tải thành OutOfMemoryError.

newFixedThreadPool hỏng âm thầm hơn nên nguy hiểm hơn: thread chặn cứng nhưng queue unbounded. Quá tải thì task không bị từ chối mà chất đống vô hạn — bộ nhớ phình theo từng task chờ, latency task cuối hàng dài vô nghĩa, hệ thống không báo gì cho tới khi đổ.

Bài học: production cần giới hạn hữu hạn ở mọi chiều — chặn số thread, chặn độ dài queue, rồi quyết định có ý thức điều gì xảy ra khi chạm trần. TicketFlow vì thế dùng ArrayBlockingQueue bounded: cả thread lẫn queue đầy thì pool từ chối task thay vì âm thầm chất đống. Chính sách từ chối cụ thể là RejectedExecutionHandler — bài 17b mổ bốn lựa chọn.

3.3 Sizing pool: CPU-bound khác I/O-bound

n nên là bao nhiêu? Không có con số thần kỳ — nguyên lý phụ thuộc bản chất task.

Task CPU-bound — mã hóa, nén, tính toán thuần — chiếm trọn một core trong lúc chạy; thêm thread vượt số core chỉ tăng chi phí context switching. Điểm xuất phát: số thread xấp xỉ số core, thường là Runtime.getRuntime().availableProcessors() cộng một (thêm một để lấp core trong lúc một thread lỡ page fault).

Task I/O-bound — gọi database, remote API, đọc file — phần lớn thời gian chỉ ngồi chờ, không dùng CPU; chạy số thread bằng số core thì core rảnh rỗi trong lúc mọi thread treo trên lời gọi mạng. Cần nhiều thread hơn số core, tỉ lệ thuận với phần thời gian chờ. Công thức của Brian Goetz (Java Concurrency in Practice §8.2): N = Ncpu × Ucpu × (1 + W/C) — trong đó Ncpu là số core, Ucpu là mức tận dụng CPU mục tiêu (từ 0 tới 1), và W/C là tỉ số thời gian chờ trên thời gian tính. Tỉ số W/C càng lớn cần càng nhiều thread: task chờ I/O 90% thời gian có W/C = 9, nên 1 + W/C = 10 — cần gấp mười lần số core mới bão hòa CPU.

Căng thẳng "I/O-bound cần rất nhiều thread mà platform thread thì đắt" là động lực dẫn tới virtual thread — Executors.newVirtualThreadPerTaskExecutor() cho một thread mỗi task không trả giá đắt (bài Virtual Threads). Hệ quả thực dụng: đừng nhét task blocking I/O và task CPU-bound vào cùng một pool — một mẻ mạng chậm sẽ chiếm hết thread, bỏ đói tính toán ngắn; tách pool theo workload (TicketFlow tách booking pool khỏi notify pool).

4. 📚 Deep Dive Oracle

5. Liên hệ các bài khác

• Bài 15 — Blocking queues & producer–consumer — workQueue của pool chính là một BlockingQueue; hiểu bài đó là hiểu vì sao pool là một hệ producer–consumer và vì sao chọn ArrayBlockingQueue bounded.
• Bài 17b — Vận hành pool production — phần tiếp nối trực tiếp: lập lịch, ThreadFactory, RejectedExecutionHandler, và shutdown graceful — những nút vặn ta cố ý gác lại ở đây.
• Bài 20 — Virtual Threads — lời giải cho căng thẳng "I/O-bound cần rất nhiều thread mà platform thread thì đắt" nêu ở mục 3.3.

6. Tóm tắt

• Executor Framework tách chính sách thực thi khỏi nghiệp vụ: gói việc thành task (Runnable/Callable), giao ExecutorService, thôi tự new Thread(). execute để exception nổi lên UncaughtExceptionHandler; submit nuốt vào Future, chỉ phát lại khi get — submit-rồi-quên là nguồn lỗi ẩn.
• Dưới mọi factory là ThreadPoolExecutor, thuật toán "core → queue → thêm thread → từ chối"; queue được ưu tiên trước khi bung thread, nên queue unbounded làm maximumPoolSize vô nghĩa.
• newCachedThreadPool không chặn số thread, newFixedThreadPool không chặn queue — production cần giới hạn hữu hạn ở mọi chiều, dùng bounded queue để pool từ chối thay vì âm thầm chất đống.
• Sizing tách CPU-bound (xấp xỉ số core, Ncpu + 1) với I/O-bound (N = Ncpu × Ucpu × (1 + W/C)); đừng trộn hai workload vào một pool.

Bounded queue nghĩa là pool có lúc từ chối task — nhưng từ chối thế nào? Và task định kỳ, thread không tên trong thread dump, pool không chịu tắt lúc deploy — những chuyện phân biệt code demo với code production. Đó là nội dung bài 17b.

7. Tự kiểm tra

Bài tiếp theo: Vận hành pool production: lập lịch, rejection & shutdown