Spring | Async注解

//启动类
@SpringBootApplication
@EnableAspectJAutoProxy
@EnableAsync
public class SpringBootWithDockerStarter {
    public static void main(String[] args) {
        SpringApplication.run(SpringBootWithDockerStarter.class, args);
    }
}
//需要异步执行的方法
@Slf4j
@Service
public class BeanE {
    @Async
    public void asyncMethod(){
        log.info("BeanE + asyncMethod");
    }
}

@RestController
@Slf4j
public class BeanController {

    @Autowired
    private BeanE beanE;

    @RequestMapping("/test")
    public void test() throws InterruptedException, ExecutionException {
        beanE.asyncMethod();
    }
}

//第一种
@Slf4j
@Service
public class BeanE {
    @Async("myExecutor") //前提是加一个myExecutor的线程池到Spring容器中
    public void asyncMethod(){
        log.info("BeanE + asyncMethod");
    }
}

//第二种
@Configuration
@Slf4j
public class AsyncConfig implements AsyncConfigurer {

    @Bean
    public Executor asyncExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setKeepAliveSeconds(30);
        executor.setQueueCapacity(10);
        executor.setThreadNamePrefix("my-async-");
        return executor;
    }

    @Override
    public Executor getAsyncExecutor() {
        return asyncExecutor();
    }
}

@Configuration
public class AsyncConfig implements AsyncConfigurer {
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        log.info("getAsyncUncaughtExceptionHandler");
        return  (Throwable ex, Method method, Object... params) -> {
            log.info("customer handle when handle async");
            log.info("Method: {}", method);
            throw new NullPointerException();
        } ;
    }
}

    public void testInternalClassCall(){
        log.info("BeanE + testInternalClassCall");
        asyncMethod();
    }

    @Async
    public void asyncMethod(){
        log.info("BeanE + asyncMethod");
    }

@Service
public class BeanE {
    @Autowired
    @Lazy
    private BeanE beanE;

    public void testInternalClassCall(){
        log.info("BeanE + testInternalClassCall");
        beanE.asyncMethod();
    }

    @Async
    public void asyncMethod(){
        log.info("BeanE + asyncMethod");
    }
}

//将返回值的类型写入CompletableFuture
@Async
public CompletableFuture<String> asyncMethodByCompletableFuture(){
    log.info("BeanF + asyncMethodByCompletableFuture");
    try {
        Thread.sleep(3000L);
    }catch (Exception e){

    }
    //将返回的类型进行包装
    return CompletableFuture.completedFuture("Hello CompletableFuture");
}

@Async
public Future<String> asyncMethod() throws InterruptedException {
    log.info("BeanF + asyncMethod");
    return new AsyncResult<>("Hello asyncMethod");
}

//在调用的时候需要调用get方法，该方法会阻塞主线程，等异步任务执行完才会继续后续的执行
public void test() throws InterruptedException, ExecutionException {
    Future<String> result = beanF.asyncMethod();
    log.info("test .....");
    String value = result.get();

    CompletableFuture<String> stringFuture = beanF.asyncMethodByCompletableFuture();
    String stringValue = stringFuture.get();
}

public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
		Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
		AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
		//此处可用于配置Async注解的线程池和异常处理
        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;
	}

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");
      //实例化这个对象时执行setBeforeExistingAdvisors(true) 这让该切面会加在所有的切面前面
      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;
   }

}

//在实例化该后处理器的时候执行了set方法
public AsyncAnnotationBeanPostProcessor() {
		setBeforeExistingAdvisors(true);
}


//该后处理器的初始化后的逻辑
public Object postProcessAfterInitialization(Object bean, String beanName) {
		if (this.advisor == null || bean instanceof AopInfrastructureBean) {
			// Ignore AOP infrastructure such as scoped proxies.
			return bean;
		}
		if (bean instanceof Advised) {
			Advised advised = (Advised) bean;
			if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {
				// Add our local Advisor to the existing proxy's Advisor chain...
				//此处就用到了上述构造函数中的set方法的变量，将Async注解切面加在已经存在的切面前面
                if (this.beforeExistingAdvisors) {
					advised.addAdvisor(0, this.advisor);
				}
				else {
					advised.addAdvisor(this.advisor);
				}
				return bean;
			}
		}
    //此处省略后续逻辑
}

    //在发生循环依赖时，为Before After等注解提前创建代理对象时会往earlyProxyReferences塞入非代理Bean
	public Object getEarlyBeanReference(Object bean, String beanName) {
		Object cacheKey = getCacheKey(bean.getClass(), beanName);
		this.earlyProxyReferences.put(cacheKey, bean);
		return wrapIfNecessary(bean, beanName, cacheKey);
	}

    //初始化后后处理逻辑为Before After等注解生成代理
	public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {
		if (bean != null) {
			Object cacheKey = getCacheKey(bean.getClass(), beanName);
            //会对之前的生成过的Bean进行判断，如果之前生成了就直接返回bean(非代理)
			if (this.earlyProxyReferences.remove(cacheKey) != bean) {
				return wrapIfNecessary(bean, beanName, cacheKey);
			}
		}
		return bean;
	}

//允许循环依赖
if (earlySingletonExposure) {
    //先从第一级缓存获取，没有再从第二级缓存获取，不涉及第三级缓存
	Object earlySingletonReference = getSingleton(beanName, false);
		if (earlySingletonReference != null) {
            //从第一级或第二级缓存中拿到Bean对象说明之前已经创建了，这是需要替换(主要是为了替换成代理对象)
			if (exposedObject == bean) {
					exposedObject = earlySingletonReference;
				}
				else if (!this.allowRawInjectionDespiteWrapping && hasDependentBean(beanName)) {
					String[] dependentBeans = getDependentBeans(beanName);
					Set<String> actualDependentBeans = new LinkedHashSet<>(dependentBeans.length);
					for (String dependentBean : dependentBeans) {
						if (!removeSingletonIfCreatedForTypeCheckOnly(dependentBean)) {
							actualDependentBeans.add(dependentBean);
						}
					}
					if (!actualDependentBeans.isEmpty()) {
						throw new BeanCurrentlyInCreationException(beanName,
								"Bean with name '" + beanName + "' has been injected into other beans [" +
								StringUtils.collectionToCommaDelimitedString(actualDependentBeans) + ...");
					}
				}
			}
		}