Vận hành pool production: lập lịch, rejection & shutdown

TL;DR: Dựng pool thì dễ, vận hành nó cho sạch mới là chuyện production. Task định kỳ dùng ScheduledExecutorService thay Timer, và phải bắt mọi exception ngay trong thân task kẻo cả lịch chết lặng lẽ. ThreadFactory đặt tên thread để đọc được thread dump lúc sự cố; RejectedExecutionHandler quyết định điều xảy ra khi pool đầy, trong đó CallerRunsPolicy tạo backpressure tự nhiên. Hủy task là hợp tác qua interruption — đừng bao giờ nuốt InterruptedException. Và shutdown theo pattern graceful-trước-forceful: shutdown → awaitTermination → shutdownNow, đóng nguồn sinh task trước nơi tiêu thụ sau, đăng ký vào một JVM shutdown hook.

1. Từ dựng pool sang vận hành pool

Bài 17 dừng ở chỗ dựng một ThreadPoolExecutor đúng đắn: thuật toán nhận task, bounded queue, sizing theo workload. Nhưng một pool chạy được trong bài test và một pool sống sót qua sự cố production lúc 3 giờ sáng là hai chuyện khác nhau.

Hình dung TicketFlow v3 đã lên production: sweeper định kỳ dọn chỗ giữ vé hết hạn ngừng chạy mà không ai hay, tới khi hàng nghìn vé quá hạn chất đống; thread dump đầy pool-3-thread-7 vô nghĩa; và mỗi lần deploy, pool cũ không chịu tắt, task đang dở bị cắt ngang. Bốn triệu chứng — lịch chết lặng, thread vô danh, quá tải không kiểm soát, shutdown bẩn — là bốn mục của bài này.

2. Lập lịch định kỳ: vì sao Timer không đủ?

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 — đúng triệu chứng "lịch chết lặng" của mục 1, 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 và bỏ qua maximumPoolSize — chỉ co giãn theo corePoolSize; đừng trông cậy nó tự bung thread khi task dồn, giữ task lập lịch ngắn gọn.

3. ThreadFactory và RejectedExecutionHandler

Hai tham số cuối của constructor ThreadPoolExecutor 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 — pool gọi newThread từ nhiều thread khi bung worker đồng thời, nên chính class quản lý thread cũng phải thread-safe (đúng bài học sold++ không nguyên tử ở bài 04 — Atomicity).

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 (bài 17 giải thích vì sao) thì phải quyết định từ chối nghĩa là gì. JDK cung cấp bốn chính sách:

CallerRunsPolicy đáng chú ý nhất vì nó 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.

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

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. Nền tảng của mọi chuyện dừng là interruption — cơ chế hủy hợp tác của Java mà bài 02b đã mổ kỹ (Java không ép thread dừng ngay; ta yêu cầu, thread chủ động kiểm tra rồi thu xếp). Ở đây ta xem executor dùng nó ra sao.

future.cancel(true) khiến executor interrupt() thread đang chạy task: bật cờ interrupt và làm các method blocking (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(); 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(); 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) để lần sweep đang chạy hoàn tất thay vì bị cắt ngang).

5. 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. 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. Trước khi xem lời giải, hãy tự dựng nó.

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)) {
            /* TODO: het han grace ma chua dung — lam gi? */
        }
    } catch (InterruptedException e) {
        /* TODO: thread chay shutdown bi interrupt — lam gi (2 viec)? */
    }
}

Đáp án — LifecycleManager của TicketFlow đóng gói pattern này:

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
    }
}

Hết hạn grace thì gọi shutdownNow() để ép, rồi awaitTermination lần hai cho task ngoan cố cơ hội cuối phản ứng sau interrupt. Trong nhánh catch, hai việc là: ép dừng (shutdownNow()) và khôi phục cờ interrupt (Thread.currentThread().interrupt()) — đúng luật mục 4, không nuốt InterruptedException. Toàn bộ đường đi:

flowchart TD
    A["shutdown(): ngung nhan task moi"] --> B["awaitTermination(grace)"]
    B --> C{"pool da dung han?"}
    C -- "roi" --> D["dong sach"]
    C -- "het han, chua dung" --> E["shutdownNow(): interrupt thread ngoan co"]
    E --> F["awaitTermination(force)"]
    F --> G{"phan hoi interrupt?"}
    G -- "roi" --> D
    G -- "khong" --> H["log: pool khong phan hoi"]
    style D fill:#6EE7B7
    style H fill:#FCA5A5

Còn một chi tiết dễ xem nhẹ: thứ tự đóng nhiều pool — đó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. Liên hệ các bài khác

• Bài 17 — Executor Framework & thuật toán nhận task — phần đầu trực tiếp: dựng pool, bounded queue, sizing; bài này là nửa "vận hành" còn lại.
• Bài 02b — Cooperative cancellation — cơ chế interrupt + InterruptedException mà Future.cancel và shutdownNow dựa lên; đọc để hiểu vì sao "không nuốt InterruptedException".
• Bài 18 — Future & CompletableFuture — đi tiếp từ Future (kết quả blocking) sang pipeline async không chặn thread.

9. Tóm tắt

• ScheduledExecutorService thay Timer cho task định kỳ; chọn fixedRate (tần suất đều) hay fixedDelay (khoảng nghỉ đều, tránh dồn), và bắt mọi exception trong thân task kẻo cả lịch chết lặng lẽ.
• ThreadFactory tự viết đặt tên thread có nghĩa để thread dump đọc được; bộ đếm phải AtomicInteger vì newThread bị gọi đồng thời.
• RejectedExecutionHandler quyết định điều xảy ra khi pool đầy: AbortPolicy ném lỗi, Discard* vứt task, CallerRunsPolicy biến giới hạn thành backpressure trung thực.
• Hủy là hợp tác qua interruption; cancel(true) interrupt task đang chạy, cancel(false) chỉ chặn task chưa khởi động — và không bao giờ nuốt InterruptedException.
• Shutdown graceful-trước-forceful: shutdown() → awaitTermination → shutdownNow() → awaitTermination lần hai; đóng nguồn sinh task trước nơi tiêu thụ, đăng ký vào JVM shutdown hook.

Pool đã dựng đúng và vận hành sạch. Nhưng ta vẫn né một câu: 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.

10. Tự kiểm tra

Bài tiếp theo: Future, FutureTask & CompletionService — nhận kết quả từ task async