Sửa lỗi java.util.concurrent.RejectedExecutionException Khi ThreadPool Đầy hoặc Đã Shutdown

TL;DR

Bạn đang submit task vào một ThreadPoolExecutor đã bị shutdown hoặc queue đã đầy mà không còn thread nào rảnh. Cách xử lý phụ thuộc vào nguyên nhân thực sự:

• Executor bị shutdown: dừng submit sau khi gọi shutdown() hoặc shutdownNow().
• Queue đầy: tăng dung lượng queue, thêm max threads, hoặc chuyển sang rejection handler CallerRunsPolicy.

Nguyên Nhân Gây Ra Exception

Stack trace đầy đủ trông như sau:

java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask rejected from java.util.concurrent.ThreadPoolExecutor[Running, pool size = 10, active threads = 10, queued tasks = 100, completed tasks = 2048]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365)

Có hai tình huống gây ra lỗi này:

• Nguyên nhân A — executor đã bị shutdown: Ai đó đã gọi executor.shutdown() hoặc executor.shutdownNow(), nhưng task vẫn tiếp tục được gửi vào sau đó. Điều này thường xảy ra với các singleton executor trong shutdown hook của ứng dụng — hook kích hoạt trong khi các producer vẫn đang chạy.
• Nguyên nhân B — queue bão hòa: Mọi thread đều đang bận VÀ blocking queue đã đạt giới hạn. Với AbortPolicy mặc định, task tiếp theo được đưa vào sẽ ngay lập tức ném ra RejectedExecutionException. Không retry, không chờ đợi.

Xác Định Bạn Đang Gặp Trường Hợp Nào

Đọc kỹ thông báo exception. Nếu trường state hiển thị Shutting down hoặc Terminated thì là Nguyên nhân A. Nếu là Running với queued tasks đã đạt giới hạn thì là Nguyên nhân B.

Vẫn chưa chắc? Thêm đoạn này trước lệnh submit:

// Ghi log trạng thái executor trước khi submit
ThreadPoolExecutor tpe = (ThreadPoolExecutor) executor;
System.out.println("Active: " + tpe.getActiveCount());
System.out.println("Queue size: " + tpe.getQueue().size());
System.out.println("Is shutdown: " + tpe.isShutdown());

Cách Xử Lý A: Executor Đã Bị Shutdown

Bảo vệ mọi lần submit bằng cách kiểm tra isShutdown(), hoặc tái cấu trúc trình tự shutdown để các producer dừng trước.

// Sai — submit sau khi shutdown
executor.shutdown();
executor.submit(someTask); // ném RejectedExecutionException

// Đúng — kiểm tra trước khi submit
if (!executor.isShutdown()) {
    executor.submit(someTask);
}

Các singleton dùng chung (ví dụ: executor @Bean của Spring) cần xử lý cẩn thận hơn. Hook @PreDestroy phải chạy sau khi tất cả producer đã dừng, không được chạy đồng thời. Khoảng thời gian chờ 30 giây là điểm khởi đầu hợp lý:

@PreDestroy
public void destroy() {
    executor.shutdown();
    try {
        if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
            executor.shutdownNow();
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

Cách Xử Lý B: Queue Đầy — Chọn Rejection Policy Phù Hợp

Java cung cấp sẵn bốn implementation của RejectedExecutionHandler. Chọn dựa trên hành vi mong muốn của ứng dụng khi pool không xử lý kịp:

Lựa chọn 1: CallerRunsPolicy (cách xử lý phổ biến nhất)

Thread đã submit task sẽ tự chạy task đó thay vì chuyển giao. Điều này tạo ra backpressure tự nhiên — producer sẽ chậm lại khi worker bị quá tải. Đây là lựa chọn mặc định tốt cho batch job và processing pipeline.

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4,                          // core threads
    10,                         // max threads
    60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(500),
    new ThreadPoolExecutor.CallerRunsPolicy()  // (500),
    new ThreadPoolExecutor.DiscardOldestPolicy()
);

Lựa chọn 3: DiscardPolicy

Bỏ qua task đến mà không thông báo. Chỉ dùng khi việc mất dữ liệu thực sự có thể chấp nhận được — ví dụ như các sự kiện analytics kiểu fire-and-forget.

new ThreadPoolExecutor(
    4, 10, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(500),
    new ThreadPoolExecutor.DiscardPolicy()
);

Lựa chọn 4: Custom policy có ghi log

Với bất kỳ hệ thống nào xử lý traffic thực tế, hãy ghi log rejection trước khi quyết định xử lý tiếp theo. Việc bỏ qua task âm thầm là cơn ác mộng debug khi có sự cố lúc 3 giờ sáng.

RejectedExecutionHandler loggingPolicy = (task, executor) -> {
    log.warn("Task bị từ chối: {} | Pool: active={}, queue={}",
        task,
        executor.getActiveCount(),
        executor.getQueue().size());
    // Tùy chọn: retry hoặc đẩy vào dead-letter queue
};

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4, 10, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(500),
    loggingPolicy
);

Cách Xử Lý C: Tăng Kích Thước Pool hoặc Queue

Đôi khi cách xử lý thực sự là cho pool nhiều tài nguyên hơn. Nếu profiling cho thấy workload thực sự cần nhiều concurrency hơn, hãy tăng các tham số:

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    8,                              // nhiều core threads hơn
    32,                             // max threads cao hơn
    120L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(2000) // queue lớn hơn
);

// Thay đổi kích thước động lúc runtime — không cần khởi động lại
executor.setCorePoolSize(16);
executor.setMaximumPoolSize(64);

Điều cần tránh: queue không giới hạn (new LinkedBlockingQueue<>() không có tham số capacity). Chúng hấp thụ áp lực âm thầm cho đến khi JVM hết heap — lúc đó bạn sẽ nhận được OutOfMemoryError thay vì RejectedExecutionException. OOM khó trace hơn nhiều so với một rejection rõ ràng. Hãy đặt giới hạn và xử lý rejection một cách tường minh.

Cấu Hình Spring Boot

ThreadPoolTaskExecutor của Spring bọc ThreadPoolExecutor bên trong và hoạt động tương tự. Cấu hình trong bean definition:

@Bean
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(8);
    executor.setMaxPoolSize(32);
    executor.setQueueCapacity(500);
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    executor.setThreadNamePrefix("async-");
    executor.initialize();
    return executor;
}

Hoặc điều chỉnh các giá trị mặc định qua application.properties mà không cần chỉnh sửa code Java:

spring.task.execution.pool.core-size=8
spring.task.execution.pool.max-size=32
spring.task.execution.pool.queue-capacity=500

Kiểm Tra Lại Sau Khi Xử Lý

Đừng chỉ khởi động lại và hy vọng. Hãy thử nghiệm executor dưới tải cao và quan sát kết quả:

• Submit task ở tốc độ cao điểm — không được có RejectedExecutionException nào xuất hiện trong log.
• Expose metrics qua JMX hoặc Micrometer và theo dõi executor.queueSize cùng executor.activeCount theo thời gian thực.
• Với CallerRunsPolicy, bạn sẽ thấy thread đang submit bị chậm lại khi có áp lực. Đó là backpressure hoạt động đúng như thiết kế — không phải lỗi.

Đây là bài stress test nhanh với 100 task trên pool nhỏ (2 core, 4 max, queue 10 slot). Tất cả 100 task đều phải hoàn thành thành công:

ExecutorService executor = new ThreadPoolExecutor(
    2, 4, 30L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(10),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

for (int i = 0; i  {
        Thread.sleep(100);
        System.out.println("Done: " + taskId + " on " + Thread.currentThread().getName());
        return null;
    });
}
executor.shutdown();

Nếu cả 100 task đều hoàn thành mà không có exception, cách xử lý đã hoạt động tốt.

Tổng Kết

• Kiểm tra isShutdown() trước — submit vào executor đã dừng luôn luôn ném exception.
• Với pool bão hòa, CallerRunsPolicy là lựa chọn mặc định phù hợp trong hầu hết trường hợp. Nó làm chậm producer một cách tự nhiên thay vì bỏ mất công việc.
• Luôn dùng queue có giới hạn. Queue không giới hạn đánh đổi một RejectedExecutionException rõ ràng lấy một crash OOM âm thầm.
• Ghi log mọi rejection trong môi trường production bằng RejectedExecutionHandler tùy chỉnh — bạn sẽ tự cảm ơn bản thân sau này.