异步执行：代理调度、上下文与取消责任
调用返回Future并不证明业务已完成；下一任务日志带着上一租户，也不说明任务提交时带错参数。本篇固定Framework v6.1.12 / Java 17 API，围绕提交、执行、结果和清理四条时间线解释这些现象。MVC请求状态见Web分发，认证过滤链见安全正文，线程绑定事务以事务正文为准。
三种异步不能混为一谈
@Async是代理把调用封装为Executor任务；Servlet Callable/DeferredResult把请求拆成初始调度与结果重调度；持久任务系统把可恢复状态存储下来。三者分别回答“谁执行”“谁写响应”“崩溃后谁恢复”。前两种内存状态都不会自动承诺重启后继续。
Callable生产者由MVC配置池管理；DeferredResult不意味着框架替应用运行生产者。@Async也不替原调用方开启跨线程共同事务。HTTP超时、Future终态、任务退出和数据库提交必须分别观察。HTTP/TLS协议只回查网络正文。
代理入口与执行器隔离
常见入口是AsyncAnnotationBeanPostProcessor登记Advisor，调用经AnnotationAsyncExecutionInterceptor选择具名或默认执行器，再把invocation.proceed封装为Callable交给doSubmit。模式为PROXY时，this自调用仍在目标对象上执行，不能因为方法上有注解就宣称切线程。
保留原文原码：AsyncExecutionInterceptor.invoke，v6.1.12，111–129行。仅此连续范围，不拼接提交实现。
Callable<Object> task = () -> {
try {
Object result = invocation.proceed();
if (result instanceof Future<?> future) {
return future.get();
}
}
catch (ExecutionException ex) {
handleError(ex.getCause(), userMethod, invocation.getArguments());
}
catch (Throwable ex) {
handleError(ex, userMethod, invocation.getArguments());
}
return null;
};
return doSubmit(task, executor, invocation.getMethod().getReturnType());
}proceed在任务里执行；若目标返回Future，工作任务还会future.get。目标返回Future的异常与当前提交线程直接遭遇拒绝不是同一路径。任务返回值也不是一个自动传播任意ThreadLocal的载体。
实验显式使用ioPool/cpuPool，各1线程、队列1：外部调用确实进入对应线程，self调用与调用方线程相同。这样可证明所选池与代理边界，不靠“输出顺序看起来异步”。不同工作类型分池能限制互相挤占，但总线程、连接和下游并发仍有共同预算；拆池不是无限容量。
结果、拒绝与异常通道
原码：AsyncExecutionAspectSupport.java，v6.1.12，连续284–304行。固定来源。
@Nullable
@SuppressWarnings("deprecation")
protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {
if (CompletableFuture.class.isAssignableFrom(returnType)) {
return executor.submitCompletable(task);
}
else if (org.springframework.util.concurrent.ListenableFuture.class.isAssignableFrom(returnType)) {
return ((org.springframework.core.task.AsyncListenableTaskExecutor) executor).submitListenable(task);
}
else if (Future.class.isAssignableFrom(returnType)) {
return executor.submit(task);
}
else if (void.class == returnType || "kotlin.Unit".equals(returnType.getName())) {
executor.submit(task);
return null;
}
else {
throw new IllegalArgumentException(
"Invalid return type for async method (only Future and void supported): " + returnType);
}
}提交按返回类型分派：CompletableFuture、Future、void/Unit各走自己的提交入口，非法返回类型同步拒绝。提交点在任务体之外，因此队列饱和的TaskRejectedException可以直接到调用方，不会先进入用户任务的catch或void handler。
原码：AsyncExecutionAspectSupport.java，v6.1.12，连续306–332行。固定来源。
/**
* Handles a fatal error thrown while asynchronously invoking the specified
* {@link Method}.
* <p>If the return type of the method is a {@link Future} object, the original
* exception can be propagated by just throwing it at the higher level. However,
* for all other cases, the exception will not be transmitted back to the client.
* In that later case, the current {@link AsyncUncaughtExceptionHandler} will be
* used to manage such exception.
* @param ex the exception to handle
* @param method the method that was invoked
* @param params the parameters used to invoke the method
*/
protected void handleError(Throwable ex, Method method, Object... params) throws Exception {
if (Future.class.isAssignableFrom(method.getReturnType())) {
ReflectionUtils.rethrowException(ex);
}
else {
// Could not transmit the exception to the caller with default executor
try {
this.exceptionHandler.obtain().handleUncaughtException(ex, method, params);
}
catch (Throwable ex2) {
logger.warn("Exception handler for async method '" + method.toGenericString() +
"' threw unexpected exception itself", ex2);
}
}
}Future方法的目标异常交还Future，必须有get/join/回调消费；void方法交AsyncUncaughtExceptionHandler，该handler本身抛错会被记录。它不是所有提交失败或所有后台错误的全局收集器。实验精确核对Future的IllegalStateException、void的IllegalArgumentException、非法返回值和拒绝根因，不把任意异常算通过。
上下文为何会丢失或串租户
租户、traceId、locale、认证身份和取消标记不能与事务连接一并复制。只捕获最小不可变必要信息，禁止共享可变SecurityContext、已关闭请求包装和活JDBC连接。InheritableThreadLocal在线程创建时继承，不按线程池每次任务更新。
安全包装的步骤是提交时捕获，执行前保存工作线程previous并安装快照，finally恢复previous。只有remove会破坏已有外层上下文；只set不恢复会串租户。空快照也须明确清理，不能把null解释成“保留上一任务”。
原码：ThreadPoolTaskExecutor.java，v6.1.12，连续249–316行。固定来源。
/**
* Specify a custom {@link TaskDecorator} to be applied to any {@link Runnable}
* about to be executed.
* <p>Note that such a decorator is not necessarily being applied to the
* user-supplied {@code Runnable}/{@code Callable} but rather to the actual
* execution callback (which may be a wrapper around the user-supplied task).
* <p>The primary use case is to set some execution context around the task's
* invocation, or to provide some monitoring/statistics for task execution.
* <p><b>NOTE:</b> Exception handling in {@code TaskDecorator} implementations
* is limited to plain {@code Runnable} execution via {@code execute} calls.
* In case of {@code #submit} calls, the exposed {@code Runnable} will be a
* {@code FutureTask} which does not propagate any exceptions; you might
* have to cast it and call {@code Future#get} to evaluate exceptions.
* See the {@code ThreadPoolExecutor#afterExecute} javadoc for an example
* of how to access exceptions in such a {@code Future} case.
* @since 4.3
*/
public void setTaskDecorator(TaskDecorator taskDecorator) {
this.taskDecorator = taskDecorator;
}
/**
* Note: This method exposes an {@link ExecutorService} to its base class
* but stores the actual {@link ThreadPoolExecutor} handle internally.
* Do not override this method for replacing the executor, rather just for
* decorating its {@code ExecutorService} handle or storing custom state.
*/
@Override
protected ExecutorService initializeExecutor(
ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
ThreadPoolExecutor executor = new ThreadPoolExecutor(
this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
queue, threadFactory, rejectedExecutionHandler) {
@Override
public void execute(Runnable command) {
Runnable decorated = command;
if (taskDecorator != null) {
decorated = taskDecorator.decorate(command);
if (decorated != command) {
decoratedTaskMap.put(decorated, command);
}
}
super.execute(decorated);
}
@Override
protected void beforeExecute(Thread thread, Runnable task) {
ThreadPoolTaskExecutor.this.beforeExecute(thread, task);
}
@Override
protected void afterExecute(Runnable task, Throwable ex) {
ThreadPoolTaskExecutor.this.afterExecute(task, ex);
}
};
if (this.allowCoreThreadTimeOut) {
executor.allowCoreThreadTimeOut(true);
}
if (this.prestartAllCoreThreads) {
executor.prestartAllCoreThreads();
}
this.threadPoolExecutor = executor;
return executor;
}TaskDecorator接收的可能是执行回调/FutureTask，而非用户原始Runnable。FutureTask可把异常记录进Future，外层装饰器catch看不到业务抛错；finally仍要恢复上下文。装饰也不证明未经过这个池的CompletableFuture回调或第三方线程具有相同上下文。无显式执行器的公共池任务须单独检查。
实验先在原始单线程池写A，再用下一任务看到A，作为泄漏反例；然后在带快照装饰器的池依次执行A、空、B、抛错任务。线程工厂预置worker-base，后续每次执行前记录的previous都必须仍为它。另用明确内联包装模型验证inner结束后恢复outer；该模型不冒称默认CallerRuns策略。
取消不是回滚
取消Future成功只说明取消竞争获胜，不能保证任何业务副作用未发生。interrupt是协作信号；吞掉中断、不可取消的驱动调用或已经提交的外部请求都可能继续产生效果。超时响应尤其不等于SQL回滚。
cancel模式使用真实@Async Future任务：闩锁确认worker已进入，再cancel(true)，get必须CancellationException，合作任务捕获中断后有界退出并执行finally。这验证的是本输入的合作取消；没有证明CompletableFuture.cancel会中断任意后台工作或真实JDBC/HTTP会停止。WebAsyncManager对Callable的超时取消与应用自行持有DeferredResult生产任务也不同。
异步写入应进入自己的事务代理并带幂等身份。主事务提交前触发查询时，异步连接可能看不到未提交状态；T14已验证线程不继承事务。本批不再伪造或重复计入T14实验，也未运行所有MySQL隔离变式。必须可靠交付时，提交回调仍可能因进程退出丢失，使用同事务Outbox和恢复投递；只在允许丢失的范围使用内存异步通知。
容量、隔离与实验
给定平均200ms、峰值1000任务/秒，近似稳态需要200在途；32个阻塞worker理论服务率仅160/秒，无界队列即使拒绝为0也会压垮deadline和内存。记录队列年龄、接纳/拒绝、运行与终结，不只看线程数。按租户和工作类型限并发，执行前再次检查过期预算。
从仓库根执行 JAVA_HOME=$(/usr/libexec/java_home -v17) bash labs/spring-senior/run.sh async。本批proxy/context/failures/cancel四模式已实跑；全部Future/闩锁/轮询/池关闭等待上限3秒，finally释放闩锁、关闭Context/池并清调用线程状态。
| 输入 | 实际断言 |
|---|---|
| proxy | 外调io/cpu具名池隔离；self不切线程 |
| context | 原始复用线程泄漏反例；A/空/B快照，异常也恢复worker-base；内联恢复outer |
| failures | Future、void、非法返回值各有错误通道；1运行+1排队后第三次真实拒绝，目标不因拒绝而执行 |
| cancel | 指定Future取消终态，合作任务退出并经过finally |
package study.spring;
import java.util.*;
import java.util.concurrent.*;
import org.springframework.context.annotation.*;
import org.springframework.scheduling.annotation.*;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
public class AsyncLab {
static final ThreadLocal<String> tenant=new ThreadLocal<>();
static final BlockingQueue<Throwable> voidErrors=new ArrayBlockingQueue<>(2);
static final CountDownLatch entered=new CountDownLatch(1),release=new CountDownLatch(1),exited=new CountDownLatch(1);
@Configuration(proxyBeanMethods=false) @EnableAsync(proxyTargetClass=true)
static class Config implements AsyncConfigurer {
@Bean(name="ioPool") ThreadPoolTaskExecutor io(){return pool("io-");}
@Bean(name="cpuPool") ThreadPoolTaskExecutor cpu(){return pool("cpu-");}
@Bean Service service(){return new Service();}
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler(){return (error,method,args)->voidErrors.add(error);}
}
static ThreadPoolTaskExecutor pool(String prefix){
var pool=new ThreadPoolTaskExecutor();pool.setThreadNamePrefix(prefix);pool.setCorePoolSize(1);pool.setMaxPoolSize(1);pool.setQueueCapacity(1);
return pool;
}
public static class Service {
@Async("ioPool") public CompletableFuture<String> io(){return CompletableFuture.completedFuture(Thread.currentThread().getName());}
public CompletableFuture<String> self(){return io();}
@Async("cpuPool") public CompletableFuture<String> cpu(){return CompletableFuture.completedFuture(Thread.currentThread().getName());}
@Async("ioPool") public CompletableFuture<String> failed(){throw new IllegalStateException("future failure");}
@Async("ioPool") public void failVoid(){throw new IllegalArgumentException("void failure");}
@Async("ioPool") public String invalid(){return "invalid";}
@Async("ioPool") public Future<String> block(){
entered.countDown();
try {if(!release.await(3,TimeUnit.SECONDS))throw new IllegalStateException("release timeout");}
catch(InterruptedException ex){Thread.currentThread().interrupt();}
finally {exited.countDown();}
return CompletableFuture.completedFuture("finished");
}
}
static void require(boolean condition,String message){if(!condition)throw new AssertionError(message);}
public static void main(String[] args)throws Exception {
String mode=args.length==0?"proxy":args[0];
if(!Set.of("proxy","failures","cancel","context").contains(mode))throw new IllegalArgumentException(mode);
if(mode.equals("context")){context();return;}
try(var c=new AnnotationConfigApplicationContext(Config.class)) {
var service=c.getBean(Service.class);
if(mode.equals("proxy")){
require(service.io().get(3,TimeUnit.SECONDS).startsWith("io-"),"external did not switch to ioPool");
require(service.cpu().get(3,TimeUnit.SECONDS).startsWith("cpu-"),"wrong executor");
require(service.self().get(3,TimeUnit.SECONDS).equals(Thread.currentThread().getName()),"self call unexpectedly switched");
} else if(mode.equals("failures")) {
try {service.failed().get(3,TimeUnit.SECONDS);throw new AssertionError("missing Future failure");}
catch(ExecutionException ex){require(ex.getCause() instanceof IllegalStateException&&"future failure".equals(ex.getCause().getMessage()),"wrong Future error");}
service.failVoid();Throwable error=voidErrors.poll(3,TimeUnit.SECONDS);
require(error instanceof IllegalArgumentException&&"void failure".equals(error.getMessage()),"void handler");
try {service.invalid();throw new AssertionError("accepted invalid return type");}
catch(IllegalArgumentException ex){require(ex.getMessage().contains("Invalid return type"),"wrong return error");}
Future<String> first=service.block();require(entered.await(3,TimeUnit.SECONDS),"worker did not enter");
try {
Future<String> second=service.io();
try {service.io();throw new AssertionError("accepted over capacity");}
catch(org.springframework.core.task.TaskRejectedException expected){require(expected.getCause() instanceof RejectedExecutionException,"wrong rejection");}
release.countDown();first.get(3,TimeUnit.SECONDS);second.get(3,TimeUnit.SECONDS);
require(voidErrors.isEmpty(),"rejection went to void handler");
} finally {release.countDown();}
} else {
Future<String> task=service.block();require(entered.await(3,TimeUnit.SECONDS),"worker did not enter");
try {
require(task.cancel(true),"cancel did not win");
try {task.get(3,TimeUnit.SECONDS);throw new AssertionError("cancelled task returned");}
catch(CancellationException expected){}
require(exited.await(3,TimeUnit.SECONDS),"cooperative worker did not exit");
} finally {release.countDown();}
}
System.out.println("PASS Async "+mode);
} finally {release.countDown();tenant.remove();}
}
static void context() throws Exception {
var pool=pool("context-");pool.initialize();
try {
// Negative control: a raw reusable worker keeps A for the following task.
pool.submit(()->tenant.set("A")).get(3,TimeUnit.SECONDS);
require("A".equals(pool.submit(tenant::get).get(3,TimeUnit.SECONDS)),"raw leakage control");
} finally {pool.shutdown();require(pool.getThreadPoolExecutor().awaitTermination(3,TimeUnit.SECONDS),"pool leaked");}
var decorated=pool("decorated-");
var previousValues=new LinkedBlockingQueue<String>();
decorated.setThreadFactory(task->new Thread(()->{install("worker-base");try{task.run();}finally{tenant.remove();}},"decorated-worker"));
decorated.setTaskDecorator(task->{
String captured=tenant.get();
return ()->{String previous=tenant.get();previousValues.add(String.valueOf(previous));install(captured);try{task.run();}finally{install(previous);}};
});
decorated.initialize();
try {
// The thread factory sets a baseline; each subsequent task observes it before install.
for(String id:List.of("A","none","B")){
install(id.equals("none")?null:id);
require(Objects.equals(tenant.get(),decorated.submit(tenant::get).get(3,TimeUnit.SECONDS)),"snapshot not propagated");
require("worker-base".equals(previousValues.poll(3,TimeUnit.SECONDS)),"worker baseline not restored");
}
try {decorated.submit(()->{throw new IllegalStateException("decorated failure");}).get(3,TimeUnit.SECONDS);throw new AssertionError("failure disappeared");}
catch(ExecutionException ex){require(ex.getCause() instanceof IllegalStateException&&"decorated failure".equals(ex.getCause().getMessage()),"wrong decorated failure");}
decorated.submit(tenant::get).get(3,TimeUnit.SECONDS);
require("worker-base".equals(previousValues.poll(3,TimeUnit.SECONDS))&&"worker-base".equals(previousValues.poll(3,TimeUnit.SECONDS)),"exception leaked context");
// Also verify nested/inline use restores an existing caller context, not merely remove().
org.springframework.core.task.TaskDecorator inline=task->{String captured=tenant.get();return ()->{String previous=tenant.get();install(captured);try{task.run();}finally{install(previous);}};};
install("inner");Runnable task=inline.decorate(()->require("inner".equals(tenant.get()),"inline snapshot"));
install("outer");task.run();require("outer".equals(tenant.get()),"inline restoration");
System.out.println("PASS Async context raw-leak/snapshot/exception/previous-restore");
} finally {tenant.remove();decorated.shutdown();require(decorated.getThreadPoolExecutor().awaitTermination(3,TimeUnit.SECONDS),"decorated pool leaked");}
}
static void install(String value){if(value==null)tenant.remove();else tenant.set(value);}
}架构师进阶：权限、恢复与成本
延时任务面临权限撤销、请求取消和重试同时发生，必须选择提交时授权还是执行时重新授权，并保存可信委托主体与审计。不能相信客户端自报tenant；资源访问仍由服务层/数据条件验证。跨进程只传必要身份和业务键，不传原始令牌、可变认证对象或数据库连接；后台再次认证/授权协议见安全正文。
候选一有限内存池，成本低且延迟小，但进程崩溃会丢任务，拒绝/取消需调用方处理。候选二持久作业/Outbox，多出状态存储、幂等、恢复任务与运维成本，却可在重启后追查。若通知从可丢失升级为资金动作，选择条件已变化，不能仅把队列开大。验证包括租户隔离、撤销后执行、故障重启、重复键、队列deadline与资源退出；未运行的跨进程恢复不冒充本地池结果。
连续追问：为什么代理存在仍未切线程（先查self入口）；任务为什么有结果但无当前租户（查提交捕获与实际池）；为什么取消成功仍有扣款（查资源与提交事实）；若必须恢复该如何选择持久协议。分层解释比只背@Async注解参数更重要。
闭卷验收
本页 8 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。