Executor Framework: Thread pool, lập lịch & shutdown

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, lập lịch 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ừ chối. Hệ quả: queue unbounded làm maximumPoolSize vô nghĩa. Production cần giới hạn hữu hạn ở mọi chiều — bounded queue cộng RejectedExecutionHandler, trong đó CallerRunsPolicy là backpressure tự nhiên. Task định kỳ dùng ScheduledExecutorService, nhớ bắt mọi exception trong thân task kẻo lịch chết lặng lẽ. Và shutdown theo pattern graceful-trước-forceful-sau, không bao giờ nuốt InterruptedException.

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. Bài Synchronizers khép lại bốn chiến lược ấ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 — quấn vào nhau thì đổi từ "mỗi task một thread" sang "pool cố định" phải sửa khắp nơi.

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 (mục 6).

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

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.

flowchart TD
    A["submit(task)"] --> B{"so thread dang co < corePoolSize?"}
    B -- "chua du core" --> C["tao CORE thread moi chay task<br/>(ke ca khi co thread ranh)"]
    B -- "du core roi" --> D{"workQueue con cho?"}
    D -- "con cho" --> E["xep task vao queue, cho thread ranh"]
    D -- "queue day" --> F{"so thread < maximumPoolSize?"}
    F -- "chua cham tran" --> G["tao them thread (non-core) chay task"]
    F -- "da kich tran" --> H["TU CHOI task<br/>RejectedExecutionHandler vao cuoc"]
    style C fill:#6EE7B7
    style G fill:#6EE7B7
    style E fill:#FCD34D
    style H fill:#FCA5A5

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

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.

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.

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) — tỉ số chờ trên tính (W/C) càng lớn cần càng nhiều thread. Task chờ I/O 90% thời gian có thể 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. Lập lịch với ScheduledExecutorService

Tới giờ mọi task đều chạy ngay khi có thể. Nhiều hệ thống cần chạy sau một khoảng trễ hoặc lặp đều: dọn cache hết hạn, gửi heartbeat, hết hạn chỗ giữ vé. Phản xạ cũ Timer/TimerTask có hai khuyết tật chí mạng: chỉ một thread nên một task chạy lâu làm trễ mọi task sau, và một exception không bắt trong TimerTask giết luôn thread Timer, khiến mọi lịch còn lại im lặng ngừng chạy. ScheduledExecutorService thay thế nó hoàn toàn.

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
scheduler.schedule(() -> sendReminder(), 10, TimeUnit.MINUTES);  // chay MOT lan sau 10 phut
scheduler.scheduleAtFixedRate(() -> emitHeartbeat(),
        0, 30, TimeUnit.SECONDS);                    // moi 30s tinh tu luc BAT DAU lan truoc
scheduler.scheduleWithFixedDelay(() -> sweepExpired(),
        1, 1, TimeUnit.MINUTES);                     // 1 phut SAU KHI lan truoc KET THUC

Hai biến thể lặp dễ chọn nhầm. scheduleAtFixedRate đo nhịp từ thời điểm bắt đầu mỗi lần chạy: lần thứ k hẹn vào mốc initialDelay + k*period, bất kể lần trước chạy bao lâu. scheduleWithFixedDelay đo từ thời điểm kết thúc: chờ đúng delay sau khi lần trước xong mới khởi động lần kế.

Khác biệt ấy vô hại khi task nhanh hơn period, thành tai họa khi task chậm hơn. Với fixedRate, sweep mất 90 giây mà period 30 giây thì các lần chạy dồn cục: scheduler cố giữ nhịp nên xếp các lần trễ nối đuôi không nghỉ, đè lên pool. fixedDelay luôn có đúng delay giây nghỉ giữa hai lần nên không bao giờ dồn. Quy tắc: cần tần suất đều (lấy mẫu metric) dùng fixedRate; cần khoảng nghỉ đều tránh chồng lấn (dọn dẹp định kỳ) dùng fixedDelay.

TicketFlow v3 chọn fixedDelay cho sweeper hết hạn chỗ giữ, đúng vì lý do tránh dồn:

handle = scheduler.scheduleWithFixedDelay(this::sweepOnce, ms, ms, TimeUnit.MILLISECONDS);

private void sweepOnce() {
    try {
        int expired = limiter.expireOldHolds();
        // ...
    } catch (Exception e) {            // BAT MOI exception ngay trong task body
        sweepErrors.incrementAndGet();
    }
}

Cái try/catch ấy không thừa. ScheduledExecutorService thừa hưởng hành vi khắc nghiệt: task định kỳ ném exception mà không bị bắt thì không chỉ lần đó thất bại — toàn bộ lịch của task ấy bị hủy vĩnh viễn, lặng lẽ, không log. Một NullPointerException từ expireOldHolds đủ khiến sweeper ngừng mãi mãi, phát hiện nhiều ngày sau khi vé hết hạn chất đống. Luật cứng với mọi periodic task: bắt mọi exception ngay trong thân task.

Lưu ý cài đặt: ScheduledThreadPoolExecutor luôn dùng hàng đợi unbounded theo thứ tự thời gian và bỏ qua maximumPoolSize — chỉ co giãn theo corePoolSize. Hợp lý cho task định kỳ thưa, nhưng đừng trông cậy nó tự bung thread khi task dồn; giữ task lập lịch ngắn gọn.

5. ThreadFactory và RejectedExecutionHandler

Hai tham số cuối của constructor thường bị bỏ mặc mặc định, nhưng trong production chúng phân định một pool dễ vận hành với một pool mù mịt khi sự cố.

ThreadFactory quyết định cách pool tạo thread. Mặc định thread mang tên kiểu pool-3-thread-1 — vô nghĩa khi bạn nhìn thread dump lúc 3 giờ sáng, cố hiểu thread nào đang treo. ThreadFactory tự viết cho phép đặt tên có ý nghĩa, daemon flag, gắn UncaughtExceptionHandler, hay set priority. Lợi ích rẻ nhất là cái tên:

public class NamedThreadFactory implements ThreadFactory {
    private final AtomicInteger counter = new AtomicInteger(0);   // newThread co the bi goi dong thoi
    @Override public Thread newThread(Runnable r) {
        Thread t = new Thread(r, namePrefix + "-" + counter.incrementAndGet());
        t.setDaemon(false);
        return t;
    }
}

Với prefix "booking-worker", thread dump hiện ra booking-worker-1, booking-worker-2, và đọc một sự cố trở nên khả thi: đống thread đang chờ database là của booking pool, không phải report pool. Bộ đếm phải là AtomicInteger — chính class quản lý thread cũng phải thread-safe.

RejectedExecutionHandler lo điều xảy ra khi pool không nhận nổi task nữa — khi cả maxPool lẫn bounded queue đều đầy, hoặc pool đã shutdown. Đã chọn bounded queue thì phải quyết định từ chối nghĩa là gì. JDK cung cấp bốn chính sách: AbortPolicy (mặc định) ném RejectedExecutionException để phía gọi tự xử; DiscardPolicy lặng lẽ vứt task mới, nguy hiểm vì mất việc không dấu vết; DiscardOldestPolicy vứt task chờ lâu nhất để nhường chỗ; và CallerRunsPolicy — đáng chú ý nhất — bắt chính thread gọi submit chạy luôn task đó.

CallerRunsPolicy là một cơ chế backpressure thanh lịch. Thread gọi (thường là thread nhận request) bị giữ lại tự chạy task, nên lúc đó không thể nhận thêm request mới. Áp lực truyền ngược lên acceptor rồi tới client, làm tốc độ đổ task vào đúng bằng tốc độ pool tiêu thụ, thay vì tích nợ hay ném lỗi tới tấp. TicketFlow chọn nó cho booking pool chính vì lẽ đó: dưới tải đỉnh, hệ thống chậm lại có kiểm soát thay vì sụp hay đánh rơi vé. Cần từ chối tường minh hơn thì viết handler riêng ném exception miền như RejectedException.

6. Cancellation và shutdown

Tạo pool thì dễ; dừng nó cho sạch mới phân biệt code demo với code production. Một pool tắt cẩu thả đánh rơi task đang dở, treo JVM vì thread không chịu chết, hoặc hỏng dữ liệu vì cắt ngang thao tác giữa chừng.

6.1 Hủy một task: interruption và Future.cancel

Java không có cách an toàn nào ép một thread dừng ngay — Thread.stop bị khai tử vì có thể bỏ object ở trạng thái hỏng. Cơ chế hủy của Java là hợp tác: ta yêu cầu thread dừng, thread phải chủ động kiểm tra rồi thu xếp. Phương tiện là interruption.

future.cancel(true) khiến executor interrupt() thread đang chạy task. Interrupt không tự dừng task; nó bật cờ trên thread và làm các method blocking như sleep, wait, take ném InterruptedException. Task tôn trọng hay không tùy cách viết: task tính toán dài cần định kỳ hỏi Thread.currentThread().isInterrupted() rồi thoát khi cờ bật; task chặn I/O bắt InterruptedException rồi dừng gọn.

Một luật mà vi phạm sẽ phá hỏng khả năng hủy của cả hệ thống: đừng bao giờ nuốt InterruptedException. Bắt mà không truyền tiếp được thì phải khôi phục cờ bằng Thread.currentThread().interrupt() để tầng trên biết đã có yêu cầu hủy. Nuốt im lặng là xóa sổ một yêu cầu dừng, pool có thể không bao giờ tắt.

future.cancel(false) nhẹ hơn: ngăn task chưa chạy khởi động, nhưng không interrupt task đang chạy. Dùng khi muốn task đang dở chạy nốt — sweeper của TicketFlow gọi handle.cancel(false) khi đóng để lần sweep đang chạy hoàn tất thay vì bị cắt ngang.

6.2 shutdown và shutdownNow

shutdown() là cuộc chia tay lịch sự: pool ngừng nhận task mới (mọi submit sau đó bị từ chối) nhưng mọi task đã nhận, kể cả trong queue, vẫn chạy cho xong.

shutdownNow() là cuộc chia tay gấp gáp: interrupt() mọi thread đang chạy, bỏ qua task còn chờ trong queue, và trả về danh sách task chưa kịp khởi động để bạn tự định đoạt. Vì dựa trên interruption, nó chỉ dừng được task biết tôn trọng interrupt. Cả hai trả về ngay; muốn biết pool dừng hẳn chưa phải hỏi bằng awaitTermination.

6.3 Pattern shutdown chuẩn: graceful trước, forceful sau

Cách tắt đúng là phối hợp cả hai: xin lịch sự trước, hết kiên nhẫn thì ép — LifecycleManager của TicketFlow đóng gói pattern đó:

static void shutdownAndAwait(String name, ExecutorService executor,
                             Duration grace, Duration force) {
    executor.shutdown();                                   // 1. ngung nhan, de task chay not
    try {
        if (!executor.awaitTermination(grace.toMillis(), TimeUnit.MILLISECONDS)) {
            executor.shutdownNow();                        // 2. qua han an hue → ep dung
            if (!executor.awaitTermination(force.toMillis(), TimeUnit.MILLISECONDS)) {
                // log: pool khong phan hoi interrupt
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();                // 3. khoi phuc co interrupt
    }
}

awaitTermination thứ hai cho task ngoan cố cơ hội cuối phản ứng sau shutdownNow(). Thread chạy shutdown mà bị interrupt giữa chừng thì ép dừng rồi khôi phục cờ — đúng luật mục 6.1.

Còn một chi tiết dễ xem nhẹ: thứ tự đóng nhiều pool. LifecycleManager đóng theo đúng thứ tự danh sách, có chủ đích: đóng nguồn sinh task trước, nơi tiêu thụ sau. Với TicketFlow là dừng sweeper trước rồi mới đóng booking pool; đảo ngược sẽ có cảnh một pool đã đóng còn bị pool khác cố submit việc vào, sinh RejectedExecutionException giữa lúc đang tắt. Đăng ký cả chuỗi vào một JVM shutdown hook để hệ thống tự dọn khi nhận tín hiệu dừng.

7. 📚 Deep Dive Oracle

8. 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 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 cộng RejectedExecutionHandler, và CallerRunsPolicy biến giới hạn thành backpressure trung thực.
• Sizing tách CPU-bound (xấp xỉ số core) với I/O-bound (N = Ncpu × Ucpu × (1 + W/C)); đừng trộn hai workload vào một pool.
• ScheduledExecutorService thay Timer; chọn đúng fixedRate hay fixedDelay, bắt mọi exception trong thân task kẻo cả lịch chết lặng lẽ.
• ThreadFactory đặt tên thread để vận hành được; shutdown graceful-trước-forceful-sau, dựa trên hủy hợp tác qua interruption mà ta không được nuốt.

Còn một câu hỏi bài này liên tục né: submit một Callable trả về một Future — rồi sao nữa? Làm sao lấy kết quả mà không chặn cứng một thread ngồi chờ, và ghép những bước phụ thuộc nhau thành chuỗi mà không lồng get vào get? Future.get là blocking và không compose được — giới hạn mà bài kế tiếp gỡ bằng CompletableFuture.

9. Tự kiểm tra

Bài tiếp theo: Future & CompletableFuture: Từ kết quả blocking đến pipeline async