spring异步@Async原理和注意问题

1、@Async导致循环依赖失败,项目启动报错

@Service
public class UserServiceImpl implements UserService {
    @Autowired
    UserService userService;
    
     @Override
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
    public int save(UserDTO userDTO) {
        User user = new User();
        BeanCopyUtils.copy(userDTO, user);
        int insert = userMapper.insert(user);
        System.out.println("User 保存用户成功:" + user);
        userService.senMsg(user);
        userService.senEmail(user);
        return insert;
    }

    @Async
    @Override
    public Boolean senMsg(User user) {
        try {
            TimeUnit.SECONDS.sleep(2);
            System.out.println("发送短信中:.....");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "给用户id:" + user.getId() + ",手机号:" + user.getMobile() + "发送短信成功");
        return true;
    }

    @Async
    @Override
    public Boolean senEmail(User user) {
        try {
            TimeUnit.SECONDS.sleep(3);
            System.out.println("发送邮件中:.....");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + "给用户id:" + user.getId() + ",邮箱:" + user.getEmail() + "发送邮件成功");
        return true;
    }

(1)现象

  如上,在使用@Async的类中出现循环依赖,启动项目时会报UnsatisfiedDependencyException,has been injected into other beans [userServiceImpl] in its raw version as part of a circular reference。

(2)解决方式
  *  创建一个过渡类,将@Async异步逻辑放到过渡类中,再将过渡类注入到原类中
  *  将循环依赖的类使用懒加载
@Autowired
@Lazy
UserService userService;

(3)原理分析

  参考文章:https://segmentfault.com/a/1190000021217176

  原理简单的描述就是:发生循环依赖时会注入一个早期暴露的bean,而被标注@Async方法的类会生成代理对象,但是spring不允许发生循环依赖时创建新对象,所以会报错。allowRawInjectionDespiteWrapping 默认是false。

//当前Bean依赖其他Bean,并且当发生循环引用时不允许新创建实例对象
                else if (!this.allowRawInjectionDespiteWrapping && hasDependentBean(beanName)) {
                    String[] dependentBeans = getDependentBeans(beanName);
                    Set<String> actualDependentBeans = new LinkedHashSet<>(dependentBeans.length);
                    //获取当前Bean所依赖的其他Bean
                    for (String dependentBean : dependentBeans) {
                        //对依赖Bean进行类型检查
                        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) +
                                "] in its raw version as part of a circular reference, but has eventually been " +
                                "wrapped. This means that said other beans do not use the final version of the " +
                                "bean. This is often the result of over-eager type matching - consider using " +
                                "'getBeanNamesOfType' with the 'allowEagerInit' flag turned off, for example.");
                    }
                }

  来分析下两个对象的创建时机:

  循环依赖的注入的原始对象是在属性注入时,通过三级缓存中的ObjectFactory.getObject()工厂类创建的对象,然后放入二级缓存中,也就是在(1)中创建好注入的。

            //将Bean实例对象封装,并且Bean定义中配置的属性值赋值给实例对象
          (1)  populateBean(beanName, mbd, instanceWrapper);
                //初始化Bean对象
          (2)  exposedObject = initializeBean(beanName, exposedObject, mbd);

  @Async标注的代理对象的创建是在(2)中,初始化bean时通过后置处理器创建的。@EnableAsync开启时它会向容器内注入AsyncAnnotationBeanPostProcessor,它是一个BeanPostProcessor,实现了postProcessAfterInitialization方法。此处我们看代码,创建代理的动作在抽象父类AbstractAdvisingBeanPostProcessor上:所以会在不同时机对一个类创建两个对象,导致spring在执行自检时,发现循环依赖出现了两个对象,所以就会抛异常。

// @since 3.2   注意:@EnableAsync在Spring3.1后出现
// 继承自ProxyProcessorSupport,所以具有动态代理相关属性~ 方便创建代理对象
public abstract class AbstractAdvisingBeanPostProcessor extends ProxyProcessorSupport implements BeanPostProcessor {

// 这里会缓存所有被处理的Bean~~~  eligible:合适的
private final Map<Class<?>, Boolean> eligibleBeans = new ConcurrentHashMap<>(256);

//postProcessBeforeInitialization方法什么不做~
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) {
    return bean;
}

// 关键是这里。当Bean初始化完成后这里会执行,这里会决策看看要不要对此Bean创建代理对象再返回~~~
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {
    if (this.advisor == null || bean instanceof AopInfrastructureBean) {
        // Ignore AOP infrastructure such as scoped proxies.
        return bean;
    }

    // 如果此Bean已经被代理了(比如已经被事务那边给代理了~~)
    if (bean instanceof Advised) {
        Advised advised = (Advised) bean;
    
        // 此处拿的是AopUtils.getTargetClass(bean)目标对象,做最终的判断
        // isEligible()是否合适的判断方法  是本文最重要的一个方法,下文解释~
        // 此处还有个小细节:isFrozen为false也就是还没被冻结的时候,就只向里面添加一个切面接口   并不要自己再创建代理对象了  省事
        if (!advised.isFrozen() && isEligible(AopUtils.getTargetClass(bean))) {
            // Add our local Advisor to the existing proxy's Advisor chain...
            // beforeExistingAdvisors决定这该advisor最先执行还是最后执行
            // 此处的advisor为:AsyncAnnotationAdvisor  它切入Class和Method标注有@Aysnc注解的地方~~~
            if (this.beforeExistingAdvisors) {
                advised.addAdvisor(0, this.advisor);
            } else {
                advised.addAdvisor(this.advisor);
            }
            return bean;
        }
    }

    // 若不是代理对象,此处就要下手了~~~~isEligible() 这个方法特别重要
    if (isEligible(bean, beanName)) {
        // copy属性  proxyFactory.copyFrom(this); 生成一个新的ProxyFactory 
        ProxyFactory proxyFactory = prepareProxyFactory(bean, beanName);
        // 如果没有强制采用CGLIB 去探测它的接口~
        if (!proxyFactory.isProxyTargetClass()) {
            evaluateProxyInterfaces(bean.getClass(), proxyFactory);
        }
        // 添加进此切面~~ 最终为它创建一个getProxy 代理对象
        proxyFactory.addAdvisor(this.advisor);
        //customize交给子类复写(实际子类目前都没有复写~)
        customizeProxyFactory(proxyFactory);
        return proxyFactory.getProxy(getProxyClassLoader());
    }

    // No proxy needed.
    return bean;
}

// 我们发现BeanName最终其实是没有用到的~~~
// 但是子类AbstractBeanFactoryAwareAdvisingPostProcessor是用到了的  没有做什么 可以忽略~~~
protected boolean isEligible(Object bean, String beanName) {
    return isEligible(bean.getClass());
}
protected boolean isEligible(Class<?> targetClass) {
    // 首次进来eligible的值肯定为null~~~
    Boolean eligible = this.eligibleBeans.get(targetClass);
    if (eligible != null) {
        return eligible;
    }
    // 如果根本就没有配置advisor  也就不用看了~
    if (this.advisor == null) {
        return false;
    }
    
    // 最关键的就是canApply这个方法,如果AsyncAnnotationAdvisor  能切进它  那这里就是true
    // 本例中方法标注有@Aysnc注解,所以铁定是能被切入的  返回true继续上面方法体的内容
    eligible = AopUtils.canApply(this.advisor, targetClass);
    this.eligibleBeans.put(targetClass, eligible);
    return eligible;
}
...
}

2、异步失效问题

@Service
public class UserServiceImpl implements UserService {
    @Autowired
    UserMapper userMapper;
    @Autowired
    SendService sendService;
    
    @Override
    @Transactional()
    public int save(UserDTO userDTO) {
        User user = new User();
        BeanCopyUtils.copy(userDTO, user);
        int insert = userMapper.insert(user);
        System.out.println("User 保存用户成功:" + user);
        this.senMsg(user);
        this.senEmail(user);
        return insert;
    }


    @Async
    @Override
    public Boolean senMsg(User user) {
        System.out.println(Thread.currentThread().getName() + "给用户id:" + user.getId() + ",手机号:" + user.getMobile() + "发送短信成功");
        return true;
    }

    @Async
    @Override
    public Boolean senEmail(User user) {
        System.out.println(Thread.currentThread().getName() + "给用户id:" + user.getId() + ",邮箱:" + user.getEmail() + "发送邮件成功");
        return true;
    }

  在本类中调用异常失效,因为在本类中调用者是this,是当前对象本身,而不是使用的代理对象。

  要在本类中使用异步,需要使用当前类的代理对象:AopContext.currentProxy(),但是要引入相应的jar包,因为要配合切面织入才能使用。

3、spring的异步默认使用SimpleAsyncTaskExecutor

  每个任务会创建一个新线程去执行,源码如下:

protected void doExecute(Runnable task) {
        Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
        thread.start();
    }

  可以自定义线程池替换默认的线程池,两种方式:

(1)定义一个线程池,然后在@Async("myTaskAsyncPool")设置

@Bean
    public Executor myTaskAsyncPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
        executor.setCorePoolSize(config.getCorePoolSize());
        //最大线程数
        executor.setMaxPoolSize(config.getMaxPoolSize());
        //队列容量
        executor.setQueueCapacity(config.getQueueCapacity());
        //活跃时间
        executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
        //线程名字前缀
        executor.setThreadNamePrefix("MyExecutor-");

        // setRejectedExecutionHandler:当pool已经达到max size的时候,如何处理新任务
        // CallerRunsPolicy:不在新线程中执行任务,而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }

  线程池对拒绝任务的处理策略:这里采用了CallerRunsPolicy策略,当线程池没有处理能力的时候,该策略会直接在 execute 方法的调用线程中运行被拒绝的任务;如果执行程序已关闭,则会丢弃该任务

  说明:setWaitForTasksToCompleteOnShutdown(true)该方法就是这里的关键,用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean,这样这些异步任务的销毁就会先于Redis线程池的销毁。同时,这里还设置了setAwaitTerminationSeconds(60),该方法用来设置线程池中任务的等待时间,如果超过这个时候还没有销毁就强制销毁,以确保应用最后能够被关闭,而不是阻塞住。

(2)实现AsyncConfigurer接口

@Slf4j
@Configuration
public class NativeAsyncTaskExecutePool implements AsyncConfigurer {


    //注入配置类
    @Autowired
    TaskThreadPoolConfig config;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程池大小
        executor.setCorePoolSize(config.getCorePoolSize());
        //最大线程数
        executor.setMaxPoolSize(config.getMaxPoolSize());
        //队列容量
        executor.setQueueCapacity(config.getQueueCapacity());
        //活跃时间
        executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
        //线程名字前缀
        executor.setThreadNamePrefix("MyExecutor-");

        // setRejectedExecutionHandler:当pool已经达到max size的时候,如何处理新任务
        // CallerRunsPolicy:不在新线程中执行任务,而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }


    /**
     *  异步任务中异常处理
     * @return
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncUncaughtExceptionHandler() {

            @Override
            public void handleUncaughtException(Throwable arg0, Method arg1, Object... arg2) {
                log.error("=========================="+arg0.getMessage()+"=======================", arg0);
                log.error("exception method:"+arg1.getName());
            }
        };
    }
}

参考文章:https://juejin.im/post/5d47a80a6fb9a06ad3470f9a

原文地址:https://www.cnblogs.com/jing-yi/p/13140871.html