[Advanced Java] ExecutorService và Thread Pool trong Java

Thread pool tách hai thứ mà new Thread(task).start() gộp làm một: việc gì cần làm và thread nào làm việc đó. Khi hai thứ đó tách ra, số lượng thread không còn là hệ quả của số lượng task nữa mà thành một con số bạn tự chọn.

ExecutorService là interface cho chuyện đó. Gần như mọi implementation bạn gặp đều là một class duy nhất, ThreadPoolExecutor, và gần như mọi bug người ta dính với nó đều đến từ việc không biết các argument của constructor tác động lẫn nhau ra sao. Bài này đi từ constructor ra ngoài: một task vừa submit thực sự đi đâu, vì sao maximum pool size thường là config chết, vì sao submit có thể giấu một task hỏng mãi mãi, và shutdown thế nào để không bỏ lại một JVM vẫn đang chạy.

Mọi output bên dưới đều lấy từ việc compile và chạy code thật trên OpenJDK 21.0.6 (arm64). Chỗ nào kết quả thay đổi giữa các lần chạy thì bài có ghi rõ và đưa ra nhiều hơn một giá trị quan sát được.

Vì sao cần thread pool

Một platform thread là một thread của hệ điều hành. Tạo nó là một system call, nó xin sẵn một vùng stack, và scheduler phải quản lý nó suốt thời gian nó sống. Làm việc đó cho mỗi task thì ổn với mười task và thảm hoạ với mười nghìn task.

Cái giá về mặt cấu trúc thì nhìn thấy được mà không cần đo thời gian. ThreadMXBean.getTotalStartedThreadCount() đếm mọi thread mà JVM từng khởi động, nên nó trả lời thẳng câu hỏi:

import java.util.concurrent.*;
import java.lang.management.*;
 
public class WhyPool {
    static final int TASKS = 10_000;
 
    public static void main(String[] args) throws Exception {
        ThreadMXBean mx = ManagementFactory.getThreadMXBean();
 
        long before = mx.getTotalStartedThreadCount();
        for (int i = 0; i < TASKS; i++) {
            Thread t = new Thread(() -> Math.sqrt(42));
            t.start();
            t.join();
        }
        long threadPerTask = mx.getTotalStartedThreadCount() - before;
 
        before = mx.getTotalStartedThreadCount();
        ExecutorService pool = Executors.newFixedThreadPool(4);
        for (int i = 0; i < TASKS; i++) pool.execute(() -> Math.sqrt(42));
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.MINUTES);
        long pooled = mx.getTotalStartedThreadCount() - before;
 
        System.out.println("tasks run                       : " + TASKS);
        System.out.println("threads started, thread-per-task: " + threadPerTask);
        System.out.println("threads started, pool of 4      : " + pooled);
    }
}

tasks run                       : 10000
threads started, thread-per-task: 10000
threads started, pool of 4      : 4

Mười nghìn task, mười nghìn thread được tạo rồi huỷ, so với bốn thread được tạo một lần và dùng lại mười nghìn lần. Đó là toàn bộ lý lẽ ủng hộ pool, và nó đúng bất kể máy nhanh tới đâu.

Nửa còn lại của lý lẽ là admission control. Một chương trình thread-per-task không có giới hạn trên nào cho lượng việc đang chạy — mỗi request tới là lập tức thành một thread sống. Một pool có số worker cố định và một queue, nên một đợt dồn dập biến thành một queue dài hơn chứ không thành nhiều thread hơn tranh nhau cùng mấy cái core.

Bên trong ThreadPoolExecutor: bảy argument của constructor

ThreadPoolExecutor có một constructor đầy đủ. Mọi thứ khác trong java.util.concurrent sinh ra pool đều là một lời gọi tới nó với vài argument đã được chọn sẵn giúp bạn.

Điều thú vị không nằm ở ý nghĩa của từng argument, mà ở thứ tự execute hỏi tới chúng. Thứ tự đó cố định, và nó là nguồn gốc của hành vi làm nhiều người bất ngờ.

Vì sao queue không giới hạn khiến maximumPoolSize không bao giờ được dùng tới

Đọc bước 2 và bước 3 cùng nhau. Một thread vượt core size chỉ được tạo sau khi queue đã từ chối task. Queue không giới hạn thì không bao giờ từ chối gì, nên bước 3 không bao giờ chạy.

Câu đó dễ nói và cũng dễ nghi ngờ, nên đây là cái pool chứng minh nó: core 1, maximum 10, và một LinkedBlockingQueue khởi tạo không truyền capacity, tức capacity bằng Integer.MAX_VALUE.

import java.util.concurrent.*;
import java.util.*;
 
public class Anatomy1 {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1,                              // corePoolSize
                10,                             // maximumPoolSize
                60L, TimeUnit.SECONDS,          // keepAliveTime
                new LinkedBlockingQueue<>());   // UNBOUNDED queue
 
        Set<String> threads = Collections.synchronizedSet(new TreeSet<>());
        for (int i = 0; i < 50; i++) {
            pool.execute(() -> {
                threads.add(Thread.currentThread().getName());
                try { Thread.sleep(20); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }
        System.out.println("queued right after submitting 50: " + pool.getQueue().size());
        System.out.println("poolSize right after submitting 50: " + pool.getPoolSize());
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.MINUTES);
        System.out.println("largestPoolSize ever reached: " + pool.getLargestPoolSize());
        System.out.println("maximumPoolSize configured:   " + pool.getMaximumPoolSize());
        System.out.println("distinct worker threads used: " + threads);
        System.out.println("completed tasks: " + pool.getCompletedTaskCount());
    }
}

queued right after submitting 50: 49
poolSize right after submitting 50: 1
largestPoolSize ever reached: 1
maximumPoolSize configured:   10
distinct worker threads used: [pool-1-thread-1]
completed tasks: 50

Năm mươi task, một pool được phép có mười thread, và đúng một thread từng tồn tại. getLargestPoolSize() là mốc cao nhất mà chính pool ghi lại, và nó dừng ở 1 trong cả sáu lần chạy đã lấy mẫu. Con số 10 ở maximumPoolSize không bao giờ được hỏi tới, vì offer trên một LinkedBlockingQueue có capacity Integer.MAX_VALUE lúc nào cũng trả về true.

Bạn kiểm tra được capacity đó trực tiếp thay vì tin vào Javadoc:

ThreadPoolExecutor fixed = (ThreadPoolExecutor) Executors.newFixedThreadPool(4);
System.out.println("fixed pool queue remainingCapacity = " + fixed.getQueue().remainingCapacity());

fixed pool queue remainingCapacity = 2147483647

Đây là điều hữu ích nhất cần biết về ThreadPoolExecutor. Một pool khai báo "core 1, max 10" đọc lên như một pool nở ra khi tải tăng. Nó không nở. Nó là một pool một thread với backlog vô hạn, và triệu chứng duy nhất của quá tải là một cái queue phình ra tới khi hết heap.

Queue có giới hạn: pool lớn tới maximum rồi từ chối task

Đổi đúng một argument — cái queue — là mọi argument còn lại sống dậy. Vẫn pool đó nhưng với ArrayBlockingQueue capacity 5:

import java.util.concurrent.*;
import java.util.*;
 
public class Anatomy2 {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1, 10,
                60L, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(5));   // BOUNDED: capacity 5
 
        Set<String> threads = Collections.synchronizedSet(new TreeSet<>());
        int accepted = 0, rejected = 0;
        for (int i = 0; i < 50; i++) {
            final int id = i;
            try {
                pool.execute(() -> {
                    threads.add(Thread.currentThread().getName());
                    try { Thread.sleep(50); }
                    catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                });
                accepted++;
            } catch (RejectedExecutionException e) {
                rejected++;
                if (rejected == 1) {
                    System.out.println("first rejection at task " + id);
                    System.out.println(e);
                }
            }
        }
        System.out.println("accepted " + accepted + ", rejected " + rejected);
        System.out.println("poolSize now: " + pool.getPoolSize()
                + ", queued: " + pool.getQueue().size());
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.MINUTES);
        System.out.println("largestPoolSize: " + pool.getLargestPoolSize()
                + " / max " + pool.getMaximumPoolSize());
        System.out.println("distinct worker threads: " + threads.size());
    }
}

first rejection at task 15
java.util.concurrent.RejectedExecutionException: Task Anatomy2$$Lambda/0x00000070010009f8@60e53b93 rejected from java.util.concurrent.ThreadPoolExecutor@55f96302[Running, pool size = 10, active threads = 10, queued tasks = 5, completed tasks = 0]
accepted 15, rejected 35
poolSize now: 10, queued: 5
largestPoolSize: 10 / max 10
distinct worker threads: 10

Mười lăm task vừa chỗ: mười trong thread cộng năm trong queue. Task thứ mười sáu không còn chỗ nào và pool ném exception. Chỉ số task 15, accepted 15 và rejected 35 giống hệt nhau ở cả năm lần chạy đã lấy mẫu; riêng phần hex định danh lambda trong message (0x00000070010009f8 ở trên) đổi sau mỗi lần chạy, vì đó là tên do class loader sinh ra cộng một identity hash code.

RejectedExecutionException kế thừa RuntimeException, nên không có gì bắt bạn phải catch nó. Điểm này đáng nhớ: chuyển một pool từ queue không giới hạn sang queue có giới hạn là dời lỗi từ "rồi sẽ đầy heap" sang "lời gọi này ném ngay bây giờ", và cái thứ hai chỉ hữu ích nếu có ai đó xử lý nó.

Để ý phần chẩn đoán mà chính message tự mang theo — pool size = 10, active threads = 10, queued tasks = 5. Khi một RejectedExecutionException hiện lên trong log production, cái ngoặc vuông đó nói cho bạn biết pool đang bão hoà hay đã shutdown rồi, không cần thêm bất cứ instrumentation nào.

Rejected-execution handler và bốn policy của nó

Handler quyết định "hết chỗ" nghĩa là gì. JDK có sẵn bốn nested class trong ThreadPoolExecutor, và mặc định là AbortPolicy. Pool này có một thread, queue 2 chỗ, và nhận 8 task:

static void run(String label, RejectedExecutionHandler h) throws Exception {
    ThreadPoolExecutor p = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(2), h);
    Set<String> ranOn = Collections.synchronizedSet(new TreeSet<>());
    int rejected = 0;
    for (int i = 0; i < 8; i++) {
        try {
            p.execute(() -> {
                ranOn.add(Thread.currentThread().getName());
                try { Thread.sleep(60); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        } catch (RejectedExecutionException e) { rejected++; }
    }
    p.shutdown();
    p.awaitTermination(10, TimeUnit.SECONDS);
    System.out.printf("%-18s submitted=8 completed=%d rejectedToCaller=%d ranOn=%s%n",
            label, p.getCompletedTaskCount(), rejected, ranOn);
}

AbortPolicy        submitted=8 completed=3 rejectedToCaller=5 ranOn=[pool-1-thread-1]
CallerRunsPolicy   submitted=8 completed=5 rejectedToCaller=0 ranOn=[main, pool-2-thread-1]
DiscardPolicy      submitted=8 completed=3 rejectedToCaller=0 ranOn=[pool-3-thread-1]
DiscardOldestPolicy submitted=8 completed=3 rejectedToCaller=0 ranOn=[pool-4-thread-1]

Cột ranOn cho thấy CallerRunsPolicy thực sự làm gì: main xuất hiện trong tập các thread đã chạy task. Năm task hoàn thành trên worker và ba task chạy trên caller. Để ý getCompletedTaskCount() chỉ đếm việc do thread của pool làm, nên nó ra 5 chứ không phải 8.

CallerRunsPolicy là thứ nên chọn mặc định khi pool được nuôi bởi một bên sản xuất mà bạn kiểm soát. Nó biến tràn thành chậm lại thay vì thành lỗi, queue không phình và không mất việc. Nó cũng có nghĩa là thread sản xuất ngừng sản xuất trong lúc chạy task, và đó đúng là cái van tiết lưu bạn cần.

Keep-alive: thread rảnh rỗi thì sao

keepAliveTime áp dụng cho thread vượt core size. Khi một thread như vậy rảnh đủ lâu, nó thoát và pool co lại. allowCoreThreadTimeOut(true) mở rộng luật đó cho cả core thread, cho phép một pool nhàn rỗi tụt về không.

ThreadPoolExecutor p = new ThreadPoolExecutor(2, 6, 500, TimeUnit.MILLISECONDS,
        new SynchronousQueue<>());
for (int i = 0; i < 6; i++)
    p.execute(() -> { try { Thread.sleep(200); } catch (InterruptedException e) {} });
Thread.sleep(100);
System.out.println("under load        : poolSize=" + p.getPoolSize());
Thread.sleep(1500);
System.out.println("after idle 1.5s   : poolSize=" + p.getPoolSize()
        + " (core=" + p.getCorePoolSize() + ")");
p.allowCoreThreadTimeOut(true);
Thread.sleep(1500);
System.out.println("allowCoreThreadTimeOut(true), idle again: poolSize=" + p.getPoolSize());

under load        : poolSize=6
after idle 1.5s   : poolSize=2 (core=2)
allowCoreThreadTimeOut(true), idle again: poolSize=0

Để ý cái queue ở đây: SynchronousQueue có capacity bằng không, nên offer chỉ thành công khi đã có sẵn một thread đang chờ nhận task. Đó là loại queue làm pool nở ra ngay lập tức, và đúng là thứ newCachedThreadPool dùng.

Các factory method của Executors và default bị giấu đi

Executors là một bộ lối tắt. Mỗi lối tắt là một ThreadPoolExecutor với argument đã chọn sẵn, và reflection trên một instance thật cho biết chọn thế nào:

static void describe(String name, ExecutorService es) {
    if (es instanceof ThreadPoolExecutor p) {
        System.out.printf("%-32s core=%-4d max=%-11s keepAlive=%-4ds queue=%s%n",
                name, p.getCorePoolSize(),
                p.getMaximumPoolSize() == Integer.MAX_VALUE
                        ? "MAX_VALUE" : p.getMaximumPoolSize(),
                p.getKeepAliveTime(TimeUnit.SECONDS),
                p.getQueue().getClass().getSimpleName());
    } else {
        System.out.printf("%-32s not a ThreadPoolExecutor: %s%n", name, es.getClass().getName());
    }
    es.shutdown();
}

availableProcessors = 10
newFixedThreadPool(4)            core=4    max=4           keepAlive=0   s queue=LinkedBlockingQueue
newCachedThreadPool()            core=0    max=MAX_VALUE   keepAlive=60  s queue=SynchronousQueue
newSingleThreadExecutor()        not a ThreadPoolExecutor: java.util.concurrent.Executors$AutoShutdownDelegatedExecutorService
newScheduledThreadPool(2)        core=2    max=MAX_VALUE   keepAlive=0   s queue=DelayedWorkQueue
newWorkStealingPool()            not a ThreadPoolExecutor: java.util.concurrent.ForkJoinPool
newVirtualThreadPerTaskExecutor() not a ThreadPoolExecutor: java.util.concurrent.ThreadPerTaskExecutor

newFixedThreadPool và newSingleThreadExecutor

newFixedThreadPool(n) là core = max = n với một LinkedBlockingQueue không giới hạn. Core bằng maximum nên đường nở ra vốn đã vô nghĩa — nhưng cái queue không giới hạn mới là default thật mà bạn đang chấp nhận. Một fixed pool không bao giờ từ chối việc; nó tích việc lại. Khi quá tải kéo dài, kiểu hỏng là một queue phình ra và cuối cùng là OutOfMemoryError, trong một stack trace không hề nhắc tới pool.

newSingleThreadExecutor() là fixed pool một thread, được bọc lại để các method của ThreadPoolExecutor không với tới được. Trên JDK 21, class bọc đó là Executors$AutoShutdownDelegatedExecutorService; ép kiểu về ThreadPoolExecutor sẽ hỏng lúc runtime. Lớp bọc đó là cố ý — nó bảo đảm hợp đồng "đúng một thread, task chạy theo thứ tự submit" bằng cách không cho ai cấu hình lại pool. Nếu bạn cần bảo đảm đó thì dùng nó. Nếu bạn cần soi hoặc tinh chỉnh pool thì tự dựng một ThreadPoolExecutor core 1, max 1.

newCachedThreadPool

core = 0, max = Integer.MAX_VALUE, SynchronousQueue, keep-alive 60 giây. Vì queue có capacity bằng không, mỗi task tới lúc không còn thread rảnh đều tạo một thread mới — tới tối đa khoảng hai tỉ.

cached pool queue remainingCapacity = 0

Đó là ảnh phản chiếu của vấn đề ở fixed pool. Fixed pool có số thread bị chặn và queue vô hạn; cached pool có số thread vô hạn và không có queue nào cả. Gặp một đợt dồn dập, nó sẽ vui vẻ tạo mỗi task một thread, đúng cái tình huống mà pool sinh ra để ngăn chặn. Nó ổn với việc ngắn, khối lượng thấp, thưa thớt, và nguy hiểm với bất cứ thứ gì bị điều khiển bởi tải bên ngoài.

Hai cái default đó, cộng lại, là lý do nhiều team có luật cấm dùng Executors và dựng thẳng ThreadPoolExecutor. Constructor dài dòng, nhưng không giấu gì cả.

newScheduledThreadPool

ScheduledExecutorService thêm schedule, scheduleAtFixedRate và scheduleWithFixedDelay. Queue của nó là DelayedWorkQueue, vốn không giới hạn, nên — theo đúng lập luận ở trên — cái maximum MAX_VALUE là không thể chạm tới và pool không bao giờ vượt quá core size.

Nó cũng có một cái bẫy riêng cùng họ với submit:

ScheduledExecutorService s = Executors.newScheduledThreadPool(2);
AtomicInteger ticks = new AtomicInteger();
ScheduledFuture<?> repeat = s.scheduleAtFixedRate(() -> {
    int n = ticks.incrementAndGet();
    if (n == 3) throw new IllegalStateException("tick 3 failed");
}, 0, 100, TimeUnit.MILLISECONDS);
 
Thread.sleep(1200);
System.out.println("ticks after 1.2s of a 100ms fixed rate: " + ticks.get());
System.out.println("repeating future isDone=" + repeat.isDone()
        + " isCancelled=" + repeat.isCancelled());
try { repeat.get(); }
catch (ExecutionException e) { System.out.println("its get() -> " + e.getCause()); }

ticks after 1.2s of a 100ms fixed rate: 3
repeating future isDone=true isCancelled=false
its get() -> java.lang.IllegalStateException: tick 3 failed

Mười hai nhịp đáng lẽ phải chạy và chỉ ba nhịp xảy ra. Một task lặp lại mà ném exception thì không được retry và không được log — việc lặp đơn giản là dừng, vĩnh viễn, và dấu vết duy nhất là một Future lặng lẽ chuyển sang done. Thân của mọi periodic task nên được bọc trong try/catch của chính nó, bao trọn mọi thứ nó làm.

newVirtualThreadPerTaskExecutor

Java 21 hoàn thiện virtual thread, và Executors.newVirtualThreadPerTaskExecutor() trả về một ExecutorService vốn không phải pool — class là ThreadPerTaskExecutor và nó khởi động một virtual thread cho mỗi task, không giới hạn.

virtual tasks run          : 10000
platform threads started   : 11
distinct carrier threads   : 10
executor class             : java.util.concurrent.ThreadPerTaskExecutor
virtual thread isDaemon    : true

Mười nghìn task, mỗi task đều sleep, chạy trên mười một platform thread. Không có gì được pool lại: mười nghìn virtual thread được tạo rồi vứt đi, và mười carrier thread bên dưới khớp đúng với mười processor của máy này. Pool virtual thread là vô nghĩa vì tạo một cái rất rẻ; lý do giữ ThreadPoolExecutor trong một codebase Java 21 là thứ mà virtual-thread executor không cho bạn — một giới hạn cho lượng việc chạy cùng lúc. Giới hạn đó thường mới là mục đích, nhất là khi pool đứng trước một database hay một API có hạn mức connection.

submit khác execute, và exception biến mất

Executor.execute(Runnable) trả về void. ExecutorService.submit có ba overload và trả về một Future:

Future<?> r = es.submit(() -> System.out.println("runnable body"));
System.out.println("submit(Runnable).get() = " + r.get());
Future<String> t = es.submit(() -> System.out.println("with result"), "TOKEN");
System.out.println("submit(Runnable, T).get() = " + t.get());
Future<Integer> c = es.submit(() -> 6 * 7);
System.out.println("submit(Callable).get() = " + c.get());

runnable body
submit(Runnable).get() = null
with result
submit(Runnable, T).get() = TOKEN
submit(Callable).get() = 42

execute chỉ nhận Runnable, và đưa cho nó một Callable thì không compile được:

Bad.java:7: error: incompatible types: Callable<String> cannot be converted to Runnable
        es.execute(c);
                   ^

Đó là khác biệt nhìn thấy được. Khác biệt không nhìn thấy được mới quan trọng hơn nhiều.

Lỗi mà bạn không bao giờ nhìn thấy

Cùng một task, cùng một pool, hai method:

import java.util.concurrent.*;
 
public class SubmitVsExecute {
    static void boom() {
        throw new IllegalStateException("task blew up");
    }
 
    public static void main(String[] args) throws Exception {
        ExecutorService es = Executors.newFixedThreadPool(1);
 
        System.out.println("--- execute(Runnable) ---");
        es.execute(SubmitVsExecute::boom);
        Thread.sleep(300);
 
        System.out.println("--- submit(Runnable), Future ignored ---");
        Future<?> f = es.submit(SubmitVsExecute::boom);
        Thread.sleep(300);
        System.out.println("nothing printed above. isDone=" + f.isDone());
 
        System.out.println("--- the same Future, get() called ---");
        try {
            f.get();
        } catch (ExecutionException e) {
            System.out.println("caught: " + e);
            System.out.println("cause:  " + e.getCause());
            System.out.println("cause class: " + e.getCause().getClass().getName());
        }
        es.shutdown();
    }
}

--- execute(Runnable) ---
Exception in thread "pool-1-thread-1" java.lang.IllegalStateException: task blew up
	at SubmitVsExecute.boom(SubmitVsExecute.java:5)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
--- submit(Runnable), Future ignored ---
nothing printed above. isDone=true
--- the same Future, get() called ---
caught: java.util.concurrent.ExecutionException: java.lang.IllegalStateException: task blew up
cause:  java.lang.IllegalStateException: task blew up
cause class: java.lang.IllegalStateException

Với execute, throwable lan ra khỏi Runnable.run(), không có gì trong runWorker catch nó, worker thread chết và uncaught-exception handler của nó in stack trace. Pool sau đó lặng lẽ tạo một worker thay thế:

next task ran on: pool-1-thread-2

Với submit, task được bọc trong một FutureTask mà run() của nó catch Throwable và lưu lại làm kết quả của task. Worker thread không bao giờ thấy exception nào, nên không in gì và không chết. Lỗi chỉ tồn tại bên trong Future, và Future.get() là thứ duy nhất trả nó lại cho bạn — bọc trong một ExecutionException, với exception gốc nằm ở getCause().

⚠️ Một task submit mà Future bị vứt đi có thể hỏng ở mọi lần chạy, mãi mãi, mà không sinh ra một dòng output nào. Đây là cách phổ biến nhất khiến một background job âm thầm ngừng hoạt động.

Làm cho lỗi của task đã submit hiện ra trở lại

Đặt một UncaughtExceptionHandler trên ThreadFactory không giúp được gì, vì với submit thì chẳng có gì "uncaught" cả:

AtomicInteger n = new AtomicInteger();
ThreadFactory tf = r -> {
    Thread t = new Thread(r, "ingest-" + n.incrementAndGet());
    t.setUncaughtExceptionHandler((th, e) ->
            System.out.println("  UEH on " + th.getName() + ": " + e));
    return t;
};
ThreadPoolExecutor p = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(), tf);
 
System.out.println("execute:");
p.execute(() -> { throw new IllegalStateException("via execute"); });
Thread.sleep(300);
System.out.println("submit:");
p.submit(() -> { throw new IllegalStateException("via submit"); });
Thread.sleep(300);
System.out.println("(nothing printed for submit)");

execute:
  UEH on ingest-1: java.lang.IllegalStateException: via execute
submit:
(nothing printed for submit)

Có ba cách thực sự hiệu quả, xếp theo công sức tăng dần:

1. Catch ngay trong task. Bọc toàn bộ thân task trong try/catch (Throwable t) và log ở đó. Nhàm chán, tường minh, và đúng với cả hai method.
2. Giữ lại Future và gọi get(). Đúng ở mọi chỗ bạn có nơi để chờ, và tiện thể lấy luôn kết quả.
3. Override afterExecute. Hook này chạy trên worker thread sau mỗi task và moi được throwable ra khỏi Future:

ThreadPoolExecutor hooked = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>()) {
    @Override protected void afterExecute(Runnable r, Throwable t) {
        if (t == null && r instanceof Future<?> f && f.isDone()) {
            try { f.get(); }
            catch (CancellationException ce) { t = ce; }
            catch (ExecutionException ee) { t = ee.getCause(); }
            catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
        }
        if (t != null) System.out.println("  afterExecute saw: " + t);
    }
};
hooked.submit(() -> { throw new IllegalStateException("via submit, hooked"); });

  afterExecute saw: java.lang.IllegalStateException: via submit, hooked

Luật rút ra: dùng execute cho việc bắn-rồi-quên để lỗi kêu to, và dùng submit khi bạn thực sự sẽ nhìn vào Future. submit mà bỏ qua giá trị trả về là tệ nhất trong mọi lựa chọn.

Callable và Future

Runnable.run() không trả về gì và không được phép ném checked exception. Callable<V>.call() trả về một V và được ném bất cứ thứ gì. Khác biệt chỉ có vậy, và đó là lý do một lambda trả về giá trị hoặc ném checked exception sẽ là một Callable.

Future<V> là cái tay cầm trỏ tới một task đang chạy ở chỗ khác. Nó có năm method hữu dụng: get(), get(timeout, unit), cancel(boolean), isCancelled() và isDone().

get, get có timeout và TimeoutException

get() block tới khi task xong rồi trả về giá trị, ném ExecutionException nếu task đã ném, hoặc ném CancellationException nếu task bị cancel. Overload có timeout thêm một kết cục thứ tư:

Future<String> f2 = es.submit(slow("B", 2000));
try {
    f2.get(200, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
    System.out.println("get(timeout)     -> threw " + e.getClass().getName()
            + ", message=" + e.getMessage());
    System.out.println("  after timeout: isDone=" + f2.isDone()
            + " isCancelled=" + f2.isCancelled());
}
boolean c = f2.cancel(true);
System.out.println("  cancel(true)   -> " + c + ", isCancelled=" + f2.isCancelled());
try { f2.get(); }
catch (CancellationException e) { System.out.println("  get() after cancel -> "
        + e.getClass().getName()); }

get()            -> A done
get(timeout)     -> threw java.util.concurrent.TimeoutException, message=null
  after timeout: isDone=false isCancelled=false
  cancel(true)   -> true, isCancelled=true
  get() after cancel -> java.util.concurrent.CancellationException

Hai chi tiết đáng giữ. TimeoutException mang message null, nên log mỗi e.getMessage() thì chẳng được gì — hãy tự log tên của thao tác. Và timeout không cancel bất cứ thứ gì: sau khi nó bắn, isDone() vẫn là false và task vẫn đang chạy. Một get có timeout mà không kèm cancel ngay sau đó là rò rỉ cả một task đang chạy lẫn một thread của pool.

Một get() không timeout nằm bên trong một task của pool mà lại chờ một task khác trong cùng pool đó chính là cách một thread pool tự deadlock: kẻ đang chờ chiếm mất đúng cái thread lẽ ra phải chạy task mà nó chờ. Kiểu hỏng đó, và deadlock nói chung, là chủ đề của một bài sau trong loạt này.

cancel(true) khác cancel(false)

cancel đánh dấu Future là đã cancel trong cả hai trường hợp. Giá trị boolean chỉ quyết định thread đang chạy có bị interrupt hay không:

// A: đang block trong sleep, cancel với true
Future<?> a = es.submit(() -> {
    try {
        Thread.sleep(3000);
        System.out.println("A: ran to completion");
    } catch (InterruptedException e) {
        System.out.println("A: InterruptedException, interrupt flag now = "
                + Thread.currentThread().isInterrupted());
    }
});
// B: đang block trong sleep, cancel với false
Future<?> b = es.submit(() -> {
    try { Thread.sleep(600); System.out.println("B: ran to completion despite cancel(false)"); }
    catch (InterruptedException e) { System.out.println("B: InterruptedException"); }
});
// C: một vòng lặp CPU không bao giờ kiểm tra cờ interrupt
Future<?> c = es.submit(() -> {
    long end = System.nanoTime() + 1_500_000_000L;
    while (System.nanoTime() < end) spins.incrementAndGet();
    System.out.println("C: finished anyway, interrupt flag = "
            + Thread.currentThread().isInterrupted());
});
 
Thread.sleep(200);
System.out.println("cancel A (true)  = " + a.cancel(true));
System.out.println("cancel B (false) = " + b.cancel(false));
System.out.println("cancel C (true)  = " + c.cancel(true));

cancel A (true)  = true
A: InterruptedException, interrupt flag now = false
cancel B (false) = true
cancel C (true)  = true
B: ran to completion despite cancel(false)
C: finished anyway, interrupt flag = true
A isCancelled=true completed=false
B isCancelled=true completed=true
C isCancelled=true

Thứ tự đan xen giữa dòng cancel A và output của chính A thay đổi giữa các lần chạy — trong ba lần chạy đã lấy mẫu, A: InterruptedException in ra sau dòng cancel A hai lần và trước nó một lần, vì hai thread cùng ghi vào một stream. Mọi thứ còn lại giống hệt nhau ở cả ba lần.

Ba bài học, đều nhìn thấy ở trên:

• cancel(true) trên một thread đang đỗ trong sleep, wait, join hoặc một thao tác blocking queue làm lời gọi đó ném InterruptedException ngay lập tức, và JVM xoá cờ interrupt lúc ném — đó là lý do A in ra interrupt flag now = false.
• cancel(false) không làm gì được một task đang chạy. B bị đánh dấu cancel, vẫn chạy tới hết, và kết quả của nó bị vứt đi. cancel(false) chỉ hữu ích với task chưa khởi động: nó gỡ task khỏi queue để task đó không bao giờ chạy.
• Interrupt là hợp tác. C bị bật cờ và vẫn chạy hết vòng lặp, vì không có chỗ nào trong vòng lặp kiểm tra cờ. isCancelled() trả về true cho một task đã chạy tới dòng cuối. Không có cách nào ép một Java thread dừng lại.

invokeAll và invokeAny

invokeAll submit cả một collection và block tới khi mọi task xong, rồi trả về danh sách Future theo thứ tự của collection đầu vào — không phải theo thứ tự hoàn thành:

List<Callable<String>> tasks = List.of(
        task("slow-300", 300), task("fast-50", 50), task("mid-150", 150));
List<Future<String>> all = es.invokeAll(tasks);
System.out.println("invokeAll returned " + all.size() + " futures, all done? "
        + all.stream().allMatch(Future::isDone));
for (Future<String> f : all) System.out.println("  " + f.get());
String winner = es.invokeAny(tasks);
System.out.println("invokeAny -> " + winner);

invokeAll returned 3 futures, all done? true
  slow-300
  fast-50
  mid-150
invokeAny -> fast-50

Mọi Future trả về đều đã done, nên các lời gọi get() không block. Task hỏng vẫn có Future của nó; exception lộ ra ở get(), như thường lệ:

  mixed -> ok
  mixed -> ExecutionException, cause java.lang.IllegalArgumentException: bad input

invokeAny trả về kết quả của task đầu tiên thành công và cancel phần còn lại. Nó chỉ ném ExecutionException khi mọi task đều hỏng, và nó báo lỗi nào thì thay đổi: trong năm lần chạy đã lấy mẫu, cause là IllegalStateException: first ba lần và IllegalStateException: second hai lần.

Cả hai đều có overload kèm timeout. invokeAll(tasks, timeout, unit) cancel những gì chưa xong khi hết giờ và vẫn trả về một Future cho mỗi task, nên những task chưa hoàn thành là bị cancel chứ không phải biến mất.

Shutdown một thread pool

ExecutorService là một resource. Không ai dọn nó hộ bạn, và mặc định thì thread của nó giữ JVM sống.

Chương trình không chịu thoát

Executors.defaultThreadFactory() tạo ra thread non-daemon:

ExecutorService es = Executors.newFixedThreadPool(1);
Future<Boolean> f = es.submit(() -> Thread.currentThread().isDaemon());
System.out.println("pool thread isDaemon = " + f.get());

pool thread isDaemon = false

JVM chỉ thoát khi thread non-daemon cuối cùng kết thúc, nên một pool nhàn rỗi không được shutdown sẽ giữ nó mở vô thời hạn. Hai chương trình, giống hệt nhau trừ một dòng:

public class NoShutdown {
    public static void main(String[] args) throws Exception {
        ExecutorService es = Executors.newFixedThreadPool(2);
        es.submit(() -> System.out.println("task ran on " + Thread.currentThread().getName()));
        Thread.sleep(200);
        System.out.println("main() is returning now");
        // không shutdown
    }
}

task ran on pool-1-thread-1
main() is returning now
STILL ALIVE after 5s -> JVM did not exit

Chương trình in hết những gì cần in, main đã return, và process vẫn đang chạy khi bị kill sau năm giây. Thêm es.shutdown(); trước khi main return là toàn bộ cách sửa — chính chương trình đó sau đó thoát ngay sau dòng cuối cùng.

Đây là loại bug không bao giờ lộ ra trong test và luôn lộ ra ở một job CI hay một công cụ CLI tự dưng không bao giờ kết thúc.

shutdown, shutdownNow và awaitTermination

Hai method shutdown làm hai việc khác nhau và không method nào block:

=== shutdown() ===
  started task1
  submit after shutdown: RejectedExecutionException
  isShutdown=true isTerminated=false
  finished task1
  started task2
  finished task2
  started task3
  finished task3
  awaitTermination(2s) = true
  isTerminated=true
=== shutdownNow() ===
  started task1
  interrupted task1
  shutdownNow returned 2 never-run tasks
  awaitTermination(2s) = true

isShutdown() thành true ngay khoảnh khắc bạn gọi một trong hai. isTerminated() chỉ thành true khi mọi task đã thực sự xong, và đó là lý do awaitTermination tồn tại — nó là method duy nhất ở đây có block.

shutdownNow interrupt, và interrupt là hợp tác, y hệt phần cancel ở trên. Một task catch InterruptedException rồi chạy tiếp sẽ không dừng, và awaitTermination sẽ hết giờ.

Trình tự đầy đủ, đáng giữ lại làm một hàm tiện ích:

static void shutdownGracefully(ExecutorService es, long timeoutSeconds) {
    es.shutdown();                                  // ngừng nhận việc, chạy nốt queue
    try {
        if (!es.awaitTermination(timeoutSeconds, TimeUnit.SECONDS)) {
            List<Runnable> dropped = es.shutdownNow();  // bỏ cuộc, interrupt tất cả
            System.out.println("  forced: " + dropped.size() + " tasks never started");
            if (!es.awaitTermination(timeoutSeconds, TimeUnit.SECONDS))
                System.out.println("  pool did not terminate");
        }
    } catch (InterruptedException e) {
        es.shutdownNow();
        Thread.currentThread().interrupt();          // trả cờ lại cho người gọi
    }
}

well-behaved pool:
  isTerminated=true
pool with a task that outlives the timeout:
  forced: 3 tasks never started
  isTerminated=true

Khối catch quan trọng không kém phần còn lại. Nếu chính thread gọi shutdownGracefully bị interrupt trong lúc chờ, nuốt mất cái interrupt đó là giấu đi một tín hiệu shutdown khỏi mọi thứ ở phía trên stack.

close() trên ExecutorService từ Java 19

Java 19 cho ExecutorService extends AutoCloseable và thêm một default close(). Trên JDK 21:

System.out.println("java.version = " + System.getProperty("java.version"));
System.out.println("ExecutorService instanceof AutoCloseable? "
        + AutoCloseable.class.isAssignableFrom(ExecutorService.class));
Method m = ExecutorService.class.getMethod("close");
System.out.println("declared by: " + m.getDeclaringClass().getName()
        + ", default? " + m.isDefault());
 
try (ExecutorService es = Executors.newFixedThreadPool(2)) {
    es.submit(() -> { Thread.sleep(400); System.out.println("  long task finished"); return null; });
    System.out.println("  leaving the try block now");
}
System.out.println("  after close(): the block waited for the task");

java.version = 21.0.6
ExecutorService instanceof AutoCloseable? true
declared by: java.util.concurrent.ExecutorService, default? true
  leaving the try block now
  long task finished
  after close(): the block waited for the task

Vậy là nó có thật, nó là một method default ngay trên interface, và try-with-resources chạy được. Việc nó làm là shutdown() rồi chờ terminate không giới hạn thời gian — khối try không thoát ra cho tới khi mọi task đã submit đều xong. Không có tham số timeout và không có cách nào truyền vào.

Lối thoát của nó là interrupt. Gọi close trên một pool có task sleep một phút sẽ block; interrupt thread đang close làm close() chuyển sang shutdownNow():

  closer thread still waiting? true
  hung task interrupted
  after interrupting the closing thread, closer alive = false, executor isShutdown=true isTerminated=true

close() là lựa chọn mặc định đúng cho một pool có vòng đời gói gọn trong một method. Nó là công cụ sai cho một pool sống lâu của cả ứng dụng, nơi bạn muốn một timeout tường minh và một quyết định về việc phải làm gì khi hết giờ — tức là trình tự shutdownGracefully ở trên.

Một pool nên có bao nhiêu thread?

Một thread chỉ chiếm core trong lúc nó đang tính toán. Ngay khi nó block trên một socket, một lock hay một lần đọc đĩa, scheduler giao core cho việc khác. Đúng một sự thật đó là toàn bộ chuyện chọn kích thước pool.

Bắt đầu từ số processor mà JVM nhìn thấy:

System.out.println("availableProcessors = " + Runtime.getRuntime().availableProcessors());

availableProcessors = 10

Giá trị này thuộc về máy chứ không thuộc về lần chạy — nó là 10 ở mọi lần chạy ở đây — và trong container nó phản ánh giới hạn CPU của cgroup chứ không phải số core của host.

Với việc CPU-bound, khoảng một thread cho mỗi processor. Thêm nữa không tăng throughput vì không còn CPU nào để mà thêm; nó chỉ thêm context switch và áp lực lên cache. Các thread state nói rõ điều đó — 40 thread cùng RUNNABLE trên một máy 10 processor nghĩa là tại mỗi thời điểm có 30 thread đang xếp hàng trong scheduler của hệ điều hành:

CPU workload, pool of 40:
  worker thread states: {RUNNABLE=40}
  runnable threads=40 but only 10 cores to run them

Với việc blocking, một thread cho mỗi processor để máy ngồi không. Mười thread mà mỗi thread sleep 100 ms là mười thread hoàn toàn không làm gì:

blocking workload, pool of 10:
  active=10 queued=180 poolSize=10
  worker thread states: {TIMED_WAITING=10}

Mười worker, tất cả ở TIMED_WAITING, 180 task trong queue, và mười processor chẳng làm gì. Độ sâu queue ở đây bản chất là nhạy với thời điểm lấy mẫu; nó đọc ra queued=180 ở cả bốn lần chạy đã lấy mẫu, và bản đồ thread state là {TIMED_WAITING=10} mọi lần.

Công thức tiêu chuẩn, lấy từ Java Concurrency in Practice của Brian Goetz, chính là phép tính số học của bức tranh đó:

threads ≈ cores × targetUtilisation × (1 + waitTime / computeTime)

waitTime / computeTime là tỉ lệ quyết định tất cả. Nó bằng 0 với việc tính toán thuần tuý, cho ra một thread mỗi core. Một task dành 90% thời gian để block có tỉ lệ bằng 9, cho ra mười thread mỗi core. Một task block 99% thời gian cho ra chín mươi chín.

Hãy trung thực về bản chất của công thức: nó là một điểm khởi đầu ở mức bậc độ lớn, tính từ hai con số mà bạn thường không biết chính xác, cho một workload có thành phần thay đổi theo input. Nó không tính được chuyện pool đang chia sẻ máy với các pool khác, chuyện database phía sau có hạn mức connection riêng, hay chuyện tỉ lệ block của task thay đổi theo từng request.

Cách duy nhất để có một con số thật là đo chính workload của bạn — và đo cho đàng hoàng nghĩa là dùng một benchmark harness như JMH, có warm-up, có fork JVM riêng và có thống kê trên nhiều vòng lặp. Một vòng lặp bọc quanh System.nanoTime() sẽ đo cả JIT warm-up, garbage collection và bất cứ thứ gì máy đang tiện tay làm, rồi đưa cho bạn một con số đủ tự tin để gây hiểu lầm. Hai nguyên tắc kinh nghiệm đáng giá hơn một phép đo tồi: chặn pool ở mức mà tài nguyên phía sau chịu được, và đừng bao giờ để một pool blocking và một pool CPU là cùng một pool.

FAQ

Vì sao thread pool của tôi không bao giờ vượt quá corePoolSize?

Vì queue không bao giờ từ chối task. ThreadPoolExecutor chỉ tạo thread vượt core size sau khi workQueue.offer(task) đã trả về false, mà queue không giới hạn thì luôn trả về true. Với core 1, maximum 10, một LinkedBlockingQueue và 50 task, getLargestPoolSize() dừng ở 1 — maximum không bao giờ được hỏi tới. newFixedThreadPool mặc định dùng queue không giới hạn. Nếu bạn muốn pool nở ra, hãy cho nó một queue có giới hạn như ArrayBlockingQueue, hoặc SynchronousQueue nếu muốn nó nở ra ngay lập tức.

submit và execute trong Java khác nhau ở đâu?

execute nhận một Runnable và trả về void; submit nhận Runnable hoặc Callable và trả về một Future. Hệ quả quan trọng nằm ở chuyện xử lý exception. Exception từ execute lan ra khỏi method run của worker, giết thread đó và tới được uncaught-exception handler của nó, nên nó được in ra và pool tạo một thread thay thế. Exception từ submit bị lớp bọc FutureTask catch và lưu lại, nên không in gì và worker vẫn sống — bạn chỉ thấy nó khi có ai gọi Future.get(), lúc đó nó ném ExecutionException với exception gốc làm cause. Dùng execute cho việc bắn-rồi-quên, và chỉ dùng submit khi bạn thực sự sẽ dùng tới Future.

Một thread pool nên có bao nhiêu thread?

Với việc CPU-bound, khoảng Runtime.getRuntime().availableProcessors(). Với việc blocking thì nhiều hơn, tỉ lệ với phần thời gian mỗi task nằm chờ: threads ≈ cores × targetUtilisation × (1 + waitTime / computeTime). Một task block 90% thời gian cần khoảng mười thread cho mỗi core. Đó là điểm khởi đầu chứ không phải câu trả lời — con số thật phụ thuộc vào thành phần thực tế của workload và vào tài nguyên phía sau mà bạn đang gọi tới, và muốn có nó thì cần một benchmark harness thật như JMH chứ không phải một vòng lặp đo giờ. Ngoài ra hãy tách việc CPU-bound và việc blocking ra hai pool khác nhau, vì không có một kích thước nào đúng cho cả hai.

Vì sao chương trình Java của tôi treo sau khi main return?

Gần như chắc chắn là một ExecutorService chưa bao giờ được shutdown. Executors.defaultThreadFactory() tạo thread non-daemon, và JVM chỉ thoát khi thread non-daemon cuối cùng kết thúc. Một pool có worker đang rảnh và đang chờ trên queue cũng tính. Một chương trình thử nghiệm chạy đúng một task và in xong dòng cuối vẫn còn sống sau năm giây và phải bị kill; thêm es.shutdown() là nó thoát ngay. Hoặc gọi shutdown(), hoặc dùng try-with-resources để close() được gọi, hoặc cấp một ThreadFactory đánh dấu thread là daemon — cách cuối chỉ khi mất việc đang chạy lúc thoát là chấp nhận được.

newCachedThreadPool có dùng được trong production không?

Chỉ với việc ngắn, khối lượng thấp, thưa thớt. Cấu hình của nó là core = 0, max = Integer.MAX_VALUE và một SynchronousQueue capacity bằng không, nên mọi task tới lúc không còn thread rảnh đều tạo ra một thread mới. Không có trần hiệu quả nào cho số thread. Dưới tải bị điều khiển bởi thứ gì đó bên ngoài — tốc độ request, một hàng đợi message tồn đọng — nó sẽ cố tạo mỗi task một thread, đúng cái kiểu hỏng mà pool sinh ra để ngăn. Hãy dùng một ThreadPoolExecutor tường minh với queue có giới hạn và một rejection policy do chính bạn chọn.

Future.cancel(true) có thực sự dừng được task đang chạy không?

Nó interrupt thread; còn việc đó có dừng được gì hay không thì tuỳ vào task. Một task đang block trong Thread.sleep, wait, join hoặc một thao tác blocking queue sẽ ném InterruptedException ngay lập tức, nên nó dừng. Một task nằm trong vòng lặp tính toán mà không bao giờ kiểm tra Thread.currentThread().isInterrupted() sẽ chạy tới hết với cờ đã bật — một task như vậy in ra interrupt flag = true ở chính dòng cuối của nó trong khi isCancelled() đã báo true. cancel(false) không interrupt gì cả: một task đang sleep bị cancel kiểu đó vẫn chạy tới hết và kết quả bị vứt đi. cancel(false) chỉ thực sự hữu ích với task chưa khởi động, khi nó gỡ task khỏi queue.

Vì sao task scheduleAtFixedRate của tôi ngừng chạy?

Vì nó đã ném exception. Một task lặp lại được lên lịch bằng scheduleAtFixedRate hoặc scheduleWithFixedDelay sẽ dừng vĩnh viễn ngay lần đầu nó ném exception, và không có gì được in ra. Trong một thử nghiệm mà mười hai nhịp đáng lẽ phải chạy ở tốc độ 100 ms và nhịp thứ ba ném exception, bộ đếm dừng ở 3 và Future lặp lại lặng lẽ chuyển sang done; exception chỉ lấy lại được qua repeat.get(). Hãy bọc toàn bộ thân của mọi periodic task trong try/catch (Throwable t) của chính nó và log ở đó.

Java 21 đã có virtual thread rồi thì còn nên dùng thread pool không?

Có, khi bạn cần một giới hạn. Executors.newVirtualThreadPerTaskExecutor() không phải pool: nó khởi động một virtual thread cho mỗi task và không có trần. Mười nghìn task đang sleep chạy trên mười một platform thread với mười carrier, nên pool virtual thread là vô nghĩa — tạo một cái rất rẻ. Nhưng một executor không giới hạn thì không cho bạn admission control, mà giới hạn thường mới là lý do pool có mặt ở đó: một database 20 connection, một API bị rate limit, một hạn mức bộ nhớ. Việc blocking với mức song song không giới hạn thì hợp với virtual-thread executor; việc CPU-bound và mọi thứ đứng trước một tài nguyên có hạn vẫn cần một ThreadPoolExecutor đã chọn kích thước.

Kết luận

Phần lớn những gì hỏng với ExecutorService đến từ ba thứ vô hình trong code như nó được viết ra. Cái queue quyết định maximum pool size có ý nghĩa gì không, và default không giới hạn của newFixedThreadPool khiến nó thường là không — core 1, max 10 và 50 task cho ra đúng một thread. submit nuốt mất exception mà execute lẽ ra đã in, nên một background job hỏng có thể chạy im lặng mãi mãi. Và một pool không shutdown giữ một thread non-daemon sống, kéo theo cả JVM, rất lâu sau khi chương trình đã làm xong mọi việc nó định làm.

Không có thứ nào trong đó là khó, một khi bạn nắm được thứ tự execute kiểm tra mọi thứ: core thread, rồi queue, rồi thread phụ, rồi handler. Hãy dựng pool bằng constructor tường minh để không có gì bị giấu, chọn queue có giới hạn và rejection policy một cách có chủ đích, ưu tiên CallerRunsPolicy khi bên sản xuất là của bạn và có thể làm nó chậm lại, và kết thúc vòng đời mọi pool bằng shutdown rồi awaitTermination rồi shutdownNow.

Future là một tay cầm mà bạn phải giữ và phải block lên nó. Bài tiếp theo chuyển sang CompletableFuture, thứ biến tay cầm đó thành một thứ bạn có thể chain, kết hợp và hoàn tất từ nơi khác, để một chuỗi bước bất đồng bộ thôi là một chuỗi lời gọi get().