Spring异步执行@Async与原理剖析

谁践踏了优雅 2022-11-15 11:17 236阅读 0赞

在实际的开发过程中，有些业务逻辑使用异步的方式处理更为合理。比如在某个业务逻辑中，需要把一些数据存入到redis缓存中，这个操作只是一个辅助的功能，成功或者失败对主业务并不会产生根本影响，这个过程可以通过异步的方法去进行。

Spring中通过在方法上设置`@Async`注解，可使得方法被异步调用。也就是说该方法会在调用时立即返回，而这个方法的实际执行交给Spring的TaskExecutor去完成。

# 异步执行的使用 #

## 配置类 ##

package com.morris.spring.config;
    
    import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    
    @Configuration
    @EnableAsync // 开启Async
    public class AsyncConfig implements AsyncConfigurer { 
    
    	@Override
    	public Executor getAsyncExecutor() { 
    		// 自定义线程池
    		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    		executor.setCorePoolSize(2);
    		executor.setMaxPoolSize(4);
    		executor.setQueueCapacity(10);
    		executor.setThreadNamePrefix("MyExecutor-");
    		executor.initialize();
    		return executor;
    	}
    
    }

## service层的使用 ##

package com.morris.spring.service;
    
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.scheduling.annotation.AsyncResult;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    
    @Component
    public class AsyncService { 
    
    	@Async
    	public void noResult() { 
    		System.out.println(Thread.currentThread().getName() + " execute noResult");
    	}
    
    	@Async
    	public Future<String> hasResult() throws InterruptedException { 
    		System.out.println(Thread.currentThread().getName() + " execute hasResult");
    		TimeUnit.SECONDS.sleep(5);
    		return new AsyncResult<>("success");
    	}
    
    	@Async
    	public CompletableFuture<String> completableFuture() throws InterruptedException { 
    		System.out.println(Thread.currentThread().getName() + " execute hasResult");
    		TimeUnit.SECONDS.sleep(5);
    		return CompletableFuture.completedFuture("success");
    	}
    }

## 测试类 ##

package com.morris.spring.demo.annotation;
    
    import com.morris.spring.config.AsyncConfig;
    import com.morris.spring.service.AsyncService;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    public class AsyncDemo { 
    	public static void main(String[] args) throws ExecutionException, InterruptedException { 
    		AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
    		applicationContext.register(AsyncService.class);
    		applicationContext.register(AsyncConfig.class);
    		applicationContext.refresh();
    
    		AsyncService asyncService = applicationContext.getBean(AsyncService.class);
    		asyncService.noResult(); // 无结果
    
    		Future<String> future = asyncService.hasResult();
    		System.out.println(future.get()); // 有结果
    
    		CompletableFuture<String> completableFuture = asyncService.completableFuture();
    		completableFuture.thenAccept(System.out::println);// 异步回调
    		System.out.println("-------------------");
    	}
    }

# 源码分析 #

## @EnableAsync ##

@EnableAsync导入了AsyncConfigurationSelector。

@Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Import(AsyncConfigurationSelector.class)
    public @interface EnableAsync {

## AsyncConfigurationSelector ##

AsyncConfigurationSelector的主要方法当然是selectImports()，注意这里会先调用父类的selectImports()

org.springframework.context.annotation.AdviceModeImportSelector\#selectImports(org.springframework.core.type.AnnotationMetadata)

public final String[] selectImports(AnnotationMetadata importingClassMetadata) { 
    	Class<?> annType = GenericTypeResolver.resolveTypeArgument(getClass(), AdviceModeImportSelector.class);
    	Assert.state(annType != null, "Unresolvable type argument for AdviceModeImportSelector");
    
    	AnnotationAttributes attributes = AnnotationConfigUtils.attributesFor(importingClassMetadata, annType);
    	if (attributes == null) { 
    		throw new IllegalArgumentException(String.format(
    				"@%s is not present on importing class '%s' as expected",
    				annType.getSimpleName(), importingClassMetadata.getClassName()));
    	}
    
    	AdviceMode adviceMode = attributes.getEnum(getAdviceModeAttributeName());
    	// 模板方法模式，回调子类的selectImports
    	String[] imports = selectImports(adviceMode);
    	if (imports == null) { 
    		throw new IllegalArgumentException("Unknown AdviceMode: " + adviceMode);
    	}
    	return imports;
    }

org.springframework.scheduling.annotation.AsyncConfigurationSelector\#selectImports

public String[] selectImports(AdviceMode adviceMode) { 
    	switch (adviceMode) { 
    		case PROXY:
    			// 返回ProxyAsyncConfiguration
    			return new String[] { ProxyAsyncConfiguration.class.getName()};
    		case ASPECTJ:
    			return new String[] { ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
    		default:
    			return null;
    	}
    }

AsyncConfigurationSelector导入了配置类ProxyAsyncConfiguration。

## ProxyAsyncConfiguration ##

@Configuration
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration { 
    
    	@Bean(name = TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)
    	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    	public AsyncAnnotationBeanPostProcessor asyncAdvisor() { 
    		Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
    		AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
    		bpp.configure(this.executor, this.exceptionHandler);
    		Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
    		if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) { 
    			bpp.setAsyncAnnotationType(customAsyncAnnotation);
    		}
    		bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
    		bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));
    		return bpp;
    	}
    
    }

ProxyAsyncConfiguration向容器中注入了一个AsyncAnnotationBeanPostProcessor。

疑问：这里为啥是BeanPostProcessor，不应该向事务切面或者缓存切面一样，注入一个Advisor和XxxxInterceptor（Advice）吗？

## AsyncAnnotationBeanPostProcessor ##

org.springframework.scheduling.annotation.AsyncAnnotationBeanPostProcessor\#setBeanFactory

public void setBeanFactory(BeanFactory beanFactory) { 
    	super.setBeanFactory(beanFactory);
    
    	// 注入Advisor
    	AsyncAnnotationAdvisor advisor = new AsyncAnnotationAdvisor(this.executor, this.exceptionHandler);
    	if (this.asyncAnnotationType != null) { 
    		advisor.setAsyncAnnotationType(this.asyncAnnotationType);
    	}
    	advisor.setBeanFactory(beanFactory);
    	this.advisor = advisor;
    }

注入了一个切面AsyncAnnotationAdvisor 。

## AsyncAnnotationAdvisor ##

切面AsyncAnnotationAdvisor包括通知AnnotationAsyncExecutionInterceptor  
和切点ComposablePointcut  
。

public AsyncAnnotationAdvisor(
    		@Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { 
    
    	Set<Class<? extends Annotation>> asyncAnnotationTypes = new LinkedHashSet<>(2);
    	asyncAnnotationTypes.add(Async.class);
    	try { 
    		asyncAnnotationTypes.add((Class<? extends Annotation>)
    				ClassUtils.forName("javax.ejb.Asynchronous", AsyncAnnotationAdvisor.class.getClassLoader()));
    	}
    	catch (ClassNotFoundException ex) { 
    		// If EJB 3.1 API not present, simply ignore.
    	}
    	this.advice = buildAdvice(executor, exceptionHandler); // 创建AnnotationAsyncExecutionInterceptor
    	this.pointcut = buildPointcut(asyncAnnotationTypes); // 创建ComposablePointcut
    }
    
    protected Advice buildAdvice(
    @Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) { 
    	AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
    	interceptor.configure(executor, exceptionHandler);
    	return interceptor;
    }
    
    protected Pointcut buildPointcut(Set<Class<? extends Annotation>> asyncAnnotationTypes) { 
    	ComposablePointcut result = null;
    	for (Class<? extends Annotation> asyncAnnotationType : asyncAnnotationTypes) { 
    		Pointcut cpc = new AnnotationMatchingPointcut(asyncAnnotationType, true); // 类
    		Pointcut mpc = new AnnotationMatchingPointcut(null, asyncAnnotationType, true); // 方法
    		if (result == null) { 
    			result = new ComposablePointcut(cpc);
    		}
    		else { 
    			result.union(cpc); // 类和方法的组合切点
    		}
    		result = result.union(mpc);
    	}
    	return (result != null ? result : Pointcut.TRUE);
    }

AnnotationMatchingPointcut其实就是查看类或者方法上面有没有@Async注解。

## AnnotationAsyncExecutionInterceptor ##

org.springframework.aop.interceptor.AsyncExecutionInterceptor\#invoke

public Object invoke(final MethodInvocation invocation) throws Throwable { 
    	Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);
    	Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
    	final Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
    
    	// 获得线程池
    	AsyncTaskExecutor executor = determineAsyncExecutor(userDeclaredMethod);
    	if (executor == null) { 
    		throw new IllegalStateException(
    				"No executor specified and no default executor set on AsyncExecutionInterceptor either");
    	}
    
    	// 将目标方法的执行封装为Callable，方便提交到线程池
    	Callable<Object> task = () -> { 
    		try { 
    			// 执行目标方法
    			Object result = invocation.proceed();
    			if (result instanceof Future) { 
    				return ((Future<?>) result).get();
    			}
    		}
    		catch (ExecutionException ex) { 
    			handleError(ex.getCause(), userDeclaredMethod, invocation.getArguments());
    		}
    		catch (Throwable ex) { 
    			handleError(ex, userDeclaredMethod, invocation.getArguments());
    		}
    		return null;
    	};
    
    	// 提交任务
    	return doSubmit(task, executor, invocation.getMethod().getReturnType());
    }
    
    protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) { 
    	// 执行任务
    	if (CompletableFuture.class.isAssignableFrom(returnType)) { 
    		return CompletableFuture.supplyAsync(() -> { 
    			try { 
    				return task.call();
    			}
    			catch (Throwable ex) { 
    				throw new CompletionException(ex);
    			}
    		}, executor);
    	}
    	else if (ListenableFuture.class.isAssignableFrom(returnType)) { 
    		return ((AsyncListenableTaskExecutor) executor).submitListenable(task);
    	}
    	else if (Future.class.isAssignableFrom(returnType)) { 
    		return executor.submit(task);
    	}
    	else { 
    		executor.submit(task);
    		return null;
    	}
    }