books

《Java 并发编程·线程与内存模型》1.2 Thread、Runnable、Callable 与任务抽象

把任务从执行器中拆出来：对比 Runnable 与 Callable 的签名差异，用完整 TaskProbe 演示 submit 取回 Optional 结果、Future.get 阻塞、ExecutionException 包装，追踪裸线程、execute 与 submit 三条异常路径，并给出可命名的 ThreadFactory。

Leeting Yan 8 分钟阅读 3758 字

承接的问题

上一节把线程本身看清楚了：它是操作系统调度的实体，创建有成本，数量有上限。但上一节的示例有一个隐含问题——任务被写死在 Thread 的构造参数里。

Thread t = new Thread(() -> repository.findByIsbn("978-7-111-00001"));
t.start();

这行代码把「做什么」和「用哪个线程做」绑在了一起。想换成线程池、想加超时、想批量提交、想拿到返回值，都得重写。真实系统里，任务的生命周期和线程的生命周期从来不是一回事：线程可以复用，任务是一次性的；线程属于执行器，任务属于业务。

这一节把两者拆开，并回答三个具体问题：Runnable 和 Callable 到底差在哪；Future.get() 阻塞时发生了什么；任务里抛出的异常，最后被谁吞了。

任务与执行分离

JDK 用两个函数式接口表达任务：

@FunctionalInterface
public interface Runnable {
    void run();
}

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;
}

差异不只是「有没有返回值」：

维度RunnableCallable<V>
方法签名void run()V call() throws Exception
返回值无有，类型由泛型参数决定
受检异常不能抛可以抛
能否直接传给 Thread能不能，Thread 构造器只接受 Runnable
能否提交给 ExecutorService能能
异常去向线程的未捕获异常处理器包进 Future，get() 时抛 ExecutionException

Callable 不能直接传给 Thread，这不是遗漏。Thread 的 run() 没有返回值，也没有声明受检异常，接口对不上。要在裸线程里跑 Callable，得包一层 FutureTask——它同时实现了 Runnable 和 Future，是这两个世界之间的桥，线程池内部提交 Callable 时用的也是它。

任务与执行分离之后，执行器可以决定用几个线程、要不要排队、满了怎么办、什么时候关闭；任务只负责描述计算。第 10 章会把执行器拆到七个参数，这一节先用最简单的单线程池把流程跑通。

提交一个 Callable 并取回结果

下面这个类完整演示了提交、阻塞取结果、异常包装和三种异常路径。它可以直接编译运行：

import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public final class TaskProbe {

    public enum ErrorCode {
        BOOK_NOT_FOUND, OUT_OF_STOCK, QUOTA_EXCEEDED, CONFLICT, INTERRUPTED
    }

    public static final class LoanException extends RuntimeException {
        private final ErrorCode code;

        public LoanException(ErrorCode code, String message) {
            super(message);
            this.code = code;
        }

        public ErrorCode code() {
            return code;
        }
    }

    public static void main(String[] args) throws Exception {
        ThreadFactory factory = Thread.ofPlatform().name("conc-core-", 1).factory();
        ExecutorService pool = Executors.newSingleThreadExecutor(factory);

        Callable<Optional<String>> reserve = () -> {
            Thread.sleep(50);
            System.out.println("running on " + Thread.currentThread().getName());
            return Optional.of("R-1001");
        };

        long begin = System.nanoTime();
        Future<Optional<String>> future = pool.submit(reserve);
        System.out.println("afterSubmit isDone=" + future.isDone());
        Optional<String> result = future.get();
        long elapsedMs = (System.nanoTime() - begin) / 1_000_000;
        System.out.println("result=" + result.orElseThrow());
        System.out.println("getBlockedMs=" + elapsedMs);

        Callable<String> failing = () -> {
            throw new LoanException(ErrorCode.OUT_OF_STOCK, "isbn=978-7-111-00002");
        };
        Future<String> failed = pool.submit(failing);
        try {
            failed.get();
        } catch (ExecutionException e) {
            System.out.println("wrapper=" + e.getClass().getName());
            System.out.println("cause=" + e.getCause().getClass().getName());
            System.out.println("causeMessage=" + e.getCause().getMessage());
            System.out.println("causeCode=" + ((LoanException) e.getCause()).code());
        }

        pool.execute(() -> {
            throw new LoanException(ErrorCode.CONFLICT, "execute-task");
        });
        Thread.sleep(300);

        Thread bare = new Thread(() -> {
            throw new LoanException(ErrorCode.BOOK_NOT_FOUND, "bare-thread");
        }, "conc-bare");
        bare.setUncaughtExceptionHandler((t, e) ->
                System.out.println("handler thread=" + t.getName()
                        + " exception=" + e.getClass().getSimpleName()
                        + " message=" + e.getMessage()));
        bare.start();
        bare.join();

        pool.shutdown();
        boolean terminated = pool.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("poolTerminated=" + terminated);
    }
}

编译运行：

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
$JAVA_HOME/bin/javac -d classes TaskProbe.java
$JAVA_HOME/bin/java -cp classes TaskProbe

本机输出：

afterSubmit isDone=false
running on conc-core-1
result=R-1001
getBlockedMs=55
wrapper=java.util.concurrent.ExecutionException
cause=TaskProbe$LoanException
causeMessage=isbn=978-7-111-00002
causeCode=OUT_OF_STOCK
Exception in thread "conc-core-1" TaskProbe$LoanException: execute-task
	at TaskProbe.lambda$main$2(TaskProbe.java:77)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1090)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:614)
	at java.base/java.lang.Thread.run(Thread.java:1474)
handler thread=conc-bare exception=LoanException message=bare-thread
poolTerminated=true

三个细节值得停一下。

第一，afterSubmit isDone=false。submit() 是异步的，返回时任务大概率还没开始跑。isDone() 只表示「完成、异常或取消」，不表示「成功」。

第二，getBlockedMs=55。任务里 sleep(50)，加上线程调度和提交开销，主线程在 get() 上等了 55 ms。这段时间里主线程什么也干不了——这就是同步等待的代价。

第三，异常输出出现了两次不同形态：execute() 提交的任务把异常打到了标准错误，线程名是 conc-core-1；submit() 提交的任务没有打印任何东西，异常被 Future 收走了，只有 get() 时才以 ExecutionException 的形式出现。

Future.get 的阻塞与异常包装

Future<V> 是一个跨线程的结果句柄，常用方法只有五个：

方法行为异常
get()阻塞到任务完成InterruptedException、ExecutionException、CancellationException
get(timeout, unit)最多等指定时间上面的异常，外加 TimeoutException
isDone()是否已结束（成功/失败/取消）无
cancel(mayInterrupt)尝试取消无
isCancelled()是否已取消无

get() 抛出的 ExecutionException 是一个包装，真实异常在 getCause() 里。常见的错误是 log.error("query failed", e) 直接打包装，日志栈顶没有业务信息。正确做法是先取 cause，再按业务异常处理：

try {
    future.get();
} catch (ExecutionException e) {
    Throwable cause = e.getCause();
    if (cause instanceof LoanException loan) {
        log.warn("reserve failed code={} isbn={}", loan.code(), isbn);
    } else {
        throw new IllegalStateException("unexpected task failure", cause);
    }
}

还有一个容易忽略的点：ExecutionException 包装的是 Throwable，不只是 Exception。任务里抛 Error（比如 AssertionError、OutOfMemoryError）也会被包进来。所以不要假设 cause 一定是 RuntimeException。

get(timeout, unit) 超时后任务并不会自动停止。它只是让调用方不再等，任务还在线程池里继续跑。要真正停下来，得配合 cancel(true) 和任务内部对中断的响应，这是第 6 章的内容。一个常见的错误是「超时了就当任务没发生」，然后后台任务继续写数据，造成重复扣减。

InterruptedException 的处理也有讲究。get() 被中断时会抛这个异常，同时清除中断标志。如果方法签名不允许抛出，catch 里至少要先 Thread.currentThread().interrupt() 恢复标志，再包装成业务异常。只打日志不恢复，会让上层失去取消信号，线程池关闭时可能一直等下去。

异常去哪了：三条路径

同样一个 throw new LoanException(...)，放在不同位置，去向完全不同：

提交方式异常捕获者表现是否静默
new Thread(runnable).start()线程的 UncaughtExceptionHandler默认打印 Exception in thread "name" ...，线程终止默认不静默，但只到 stderr
executor.execute(runnable)工作线程的 UncaughtExceptionHandler同上，线程池随后补一个新工作线程默认不静默，但业务日志里看不到
executor.submit(callable)FutureTask，存进结果get() 抛 ExecutionException是，不 get() 就永远不知道
executor.submit(runnable)同上get() 返回 null 或抛包装异常是

第三、四行是最危险的。很多代码 submit() 之后把返回的 Future 丢掉，异常就彻底消失了：

pool.submit(() -> repository.save(book));   // 编译通过，异常无人知晓

如果任务不需要返回值，但又不想吞异常，有两个选择：用 execute() 让异常走到未捕获处理器；或者保留 Future 并至少记录失败。更好的做法是在 ThreadFactory 或工作线程上装一个统一的 UncaughtExceptionHandler，把兜底日志打出来。但注意，submit() 的异常不会经过这个处理器，它被 FutureTask 截住了。

还有一个 lambda 重载的坑：

pool.submit(() -> list.add("x"));

List.add 返回 boolean，这个 lambda 同时兼容 Runnable 和 Callable<Boolean>，编译器会选择 Callable 重载，返回 Future<Boolean>。如果你的本意是「提交一个副作用」，这个 Future 通常会被丢掉，异常也就被吞了。先声明 Runnable task = () -> list.add("x"); 再提交，可以避免歧义。

ThreadFactory：命名是第一可观测性

Executors.defaultThreadFactory() 生成的线程名是 pool-N-thread-M。这个格式在只有一个池的小程序里够用，在多池、多下游的系统里几乎无法排查：你看到 pool-3-thread-7，不知道它属于借阅服务、通知服务还是对账任务。

自定义 ThreadFactory 的价值不在「能创建线程」，而在命名和兜底：

static final class NamedThreadFactory implements ThreadFactory {
    private final String prefix;
    private final AtomicInteger seq = new AtomicInteger();

    NamedThreadFactory(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, prefix + "-" + seq.incrementAndGet());
        t.setDaemon(false);
        t.setUncaughtExceptionHandler((thread, error) ->
                System.err.println("uncaught in " + thread.getName() + ": " + error));
        return t;
    }
}

三个约定：

  • 前缀要能定位到池的用途，比如 loan-query-、notify-mail-，而不是 thread-。
  • 序号用 AtomicInteger，ThreadFactory 会被多个线程并发调用。
  • 不要在工厂里返回 null，线程池会直接抛 NullPointerException，而且报错点离根因很远。

ThreadFactory 还可以设置 daemon、优先级、线程组。daemon 的语义在第 3 节讲，优先级在第 3 节证明它不可靠。这里只强调一点：工厂是统一设置这些属性的唯一入口，不要在任务体里临时改线程属性。

在 libraryd-conc 里落地

把借阅查询包装成 Callable，提交给带名字的线程池，并用超时保护调用方：

package com.plume.conc.core;

import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

public final class LoanQueryService implements AutoCloseable {

    private final BookRepository repository;
    private final ExecutorService pool;

    public LoanQueryService(BookRepository repository) {
        this.repository = repository;
        this.pool = Executors.newFixedThreadPool(4, new ThreadFactory() {
            private final AtomicInteger seq = new AtomicInteger();

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "loan-query-" + seq.incrementAndGet());
            }
        });
    }

    public Optional<Book> findWithTimeout(String isbn, long timeoutMs)
            throws InterruptedException, TimeoutException {
        Callable<Optional<Book>> task = () -> repository.findByIsbn(isbn);
        Future<Optional<Book>> future = pool.submit(task);
        try {
            return future.get(timeoutMs, TimeUnit.MILLISECONDS);
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            if (cause instanceof LoanException loan) {
                throw loan;
            }
            throw new IllegalStateException("book query failed: " + isbn, cause);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw e;
        }
    }

    @Override
    public void close() {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
                pool.shutdownNow();
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

这段代码里有三个可复用的决定：查询任务只依赖 BookRepository，不关心线程；超时后先 cancel(true) 再抛出，避免调用方以为任务已停；close() 走两段式关闭，先礼后兵。第 10 章会把固定线程池换成显式 ThreadPoolExecutor，解释为什么生产代码通常不该用 Executors 的快捷工厂。

注意 LoanException 是 RuntimeException，被 FutureTask 捕获后同样包进 ExecutionException。上面的代码把 cause 还原成业务异常再抛，调用方只需要处理 LoanException，不需要知道 Future 的存在。

常见错误与边界

错误后果改法
future.get() 不带超时下游卡住时调用线程永久阻塞用 get(timeout, unit)，并定义超时后的降级
submit() 后丢弃 Future任务异常彻底静默用 execute()，或保留 Future 并记录失败
捕获 ExecutionException 只打包装日志里看不到业务 cause先 getCause() 再分类处理
超时后不 cancel任务继续跑，可能重复写cancel(true) + 任务响应中断
在任务里改线程名/优先级线程池复用时污染下一个任务属性在 ThreadFactory 里设
用 submit(() -> list.add(x)) 表达副作用误选 Callable<Boolean>，异常被吞显式声明 Runnable 变量

什么时候不要用 Callable + Future：需要组合多个异步结果、需要回调链、需要超时降级时，手写 Future.get() 会退化成串行等待。第 13 章的 CompletableFuture 和第 15 章的结构化并发更适合。Future 的定位是「单个任务的取货凭证」，不是编排工具。

小结

  • Runnable 无返回值、不能抛受检异常；Callable<V> 有返回值、可以抛，但不能直接传给 Thread。
  • submit() 是异步的，isDone() 为 false 只说明没结束，不说明没开始。
  • Future.get() 会阻塞；ExecutionException 是包装，业务异常在 getCause() 里。
  • 异常去向分三条：裸线程和 execute() 走未捕获处理器，submit() 存进 Future，不 get() 就静默。
  • ThreadFactory 是统一命名和兜底的地方，线程名是最便宜的可观测性。

阅读导航：上一节 1.1 从进程到线程：JVM 线程模型 ｜ 下一节 1.3 生命周期、优先级与守护线程

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收