Spring—@Async注解执行异步任务

引言

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

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

代码示例

Spring项目中使用@Async

对于普通的Spring的项目，异步调用配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:task="http://www.springframework.org/schema/task"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/task
       http://www.springframework.org/schema/task/spring-task.xsd">

    <!-- 包扫描 -->
    <context:component-scan base-package="com.lzumetal.ssm"/>

    <!-- 执行异步任务的线程池TaskExecutor -->
    <task:executor id="myexecutor" pool-size="5"  />
    <task:annotation-driven executor="myexecutor"/>

</beans>

两个Service类：

package com.lzumetal.ssm.anotation.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 业务Service
 */
@Service
public class BusinessService {

    private static final Logger log = LoggerFactory.getLogger(BusinessService.class);

    @Autowired
    private CacheService cacheService;


    public void doBusiness() {
        log.error("start to deal with our business");
        cacheService.cacheData();
        log.error("comlete service operation");
    }

    /**
     * 获取异步方法执行的返回值
     */
    public void doBusinessWithAsyncReturn() throws ExecutionException, InterruptedException {
        log.error("start to deal with our business");
        Future<String> future = cacheService.cacheDataWithReturn();
        log.error(future.get()); //future.get()方法是会阻塞的
        log.error("comlete service operation");
    }
}

package com.lzumetal.ssm.anotation.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 缓存服务
 */
@Service
public class CacheService {

    private static final Logger log = LoggerFactory.getLogger(CacheService.class);


    @Async(value = "myexecutor")    //指定执行任务的TaskExecutor
    public void cacheData() {
        try {
            TimeUnit.SECONDS.sleep(3L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.error("success store the result to cache");
    }


    @Async
    public Future<String> cacheDataWithReturn() {
        try {
            TimeUnit.SECONDS.sleep(3L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.error("success store the result to cache");
        //返回的结果需要通过AsyncResult这个类包装
        return new AsyncResult<>("Async operation success");
    }
}

测试类：

package com.lzumetal.ssm.anotation.test;

import com.lzumetal.ssm.anotation.service.BusinessService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.util.concurrent.TimeUnit;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:spring-context.xml"})
public class MainTest {


    @Autowired
    private BusinessService businessService;


    @Test
    public void test() throws InterruptedException {
        businessService.doBusiness();
        //不让主线程过早结束，否则控制台看不到异步方法中的输出内容
        TimeUnit.SECONDS.sleep(5L);     
    }

    @Test
    public void testAsyncReturn() throws Exception {
        businessService.doBusinessWithAsyncReturn();
        TimeUnit.SECONDS.sleep(5L);
    }

}

执行test()方法的结果：

22:20:33,207  INFO main support.DefaultTestContextBootstrapper:260 - Loaded default TestExecutionListener class names from location [META-INF/spring.factories]: [org.springframework.test.context.web.ServletTestExecutionListener, org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener, org.springframework.test.context.support.DependencyInjectionTestExecutionListener, org.springframework.test.context.support.DirtiesContextTestExecutionListener, org.springframework.test.context.transaction.TransactionalTestExecutionListener, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener]
22:20:33,226  INFO main support.DefaultTestContextBootstrapper:209 - Could not instantiate TestExecutionListener [org.springframework.test.context.web.ServletTestExecutionListener]. Specify custom listener classes or make the default listener classes (and their required dependencies) available. Offending class: [javax/servlet/ServletContext]
22:20:33,227  INFO main support.DefaultTestContextBootstrapper:187 - Using TestExecutionListeners: [org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener@100fc185, org.springframework.test.context.support.DependencyInjectionTestExecutionListener@643b1d11, org.springframework.test.context.support.DirtiesContextTestExecutionListener@2ef5e5e3, org.springframework.test.context.transaction.TransactionalTestExecutionListener@36d4b5c, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener@6d00a15d]22:20:33,324  INFO main xml.XmlBeanDefinitionReader:317 - Loading XML bean definitions from class path resource [spring-context.xml]
22:20:33,585  INFO main support.GenericApplicationContext:583 - Refreshing org.springframework.context.support.GenericApplicationContext@4f7d0008: startup date [Wed May 30 22:20:33 CST 2018]; root of context hierarchy
22:20:33,763  INFO main concurrent.ThreadPoolTaskExecutor:165 - Initializing ExecutorService 
22:20:33,766  INFO main support.PostProcessorRegistrationDelegate$BeanPostProcessorChecker:325 - Bean 'myexecutor' of type [org.springframework.scheduling.config.TaskExecutorFactoryBean] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
22:20:33,767  INFO main support.PostProcessorRegistrationDelegate$BeanPostProcessorChecker:325 - Bean 'myexecutor' of type [org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
22:20:34,107 ERROR main service.BusinessService:24 - start to deal with our business
22:20:34,113 ERROR main service.BusinessService:26 - comlete service operation
22:20:37,166 ERROR myexecutor-1 service.CacheService:28 - success store the result to cache
22:20:39,117  INFO Thread-0 support.GenericApplicationContext:984 - Closing org.springframework.context.support.GenericApplicationContext@4f7d0008: startup date [Wed May 30 22:20:33 CST 2018]; root of context hierarchy
22:20:39,118  INFO Thread-0 concurrent.ThreadPoolTaskExecutor:203 - Shutting down ExecutorService

执行testAsyncReturn()方法的结果：

21:38:16,908  INFO main support.DefaultTestContextBootstrapper:260 - Loaded default TestExecutionListener class names from location [META-INF/spring.factories]: [org.springframework.test.context.web.ServletTestExecutionListener, org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener, org.springframework.test.context.support.DependencyInjectionTestExecutionListener, org.springframework.test.context.support.DirtiesContextTestExecutionListener, org.springframework.test.context.transaction.TransactionalTestExecutionListener, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener]
21:38:16,926  INFO main support.DefaultTestContextBootstrapper:209 - Could not instantiate TestExecutionListener [org.springframework.test.context.web.ServletTestExecutionListener]. Specify custom listener classes or make the default listener classes (and their required dependencies) available. Offending class: [javax/servlet/ServletContext]
21:38:16,927  INFO main support.DefaultTestContextBootstrapper:187 - Using TestExecutionListeners: [org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener@100fc185, org.springframework.test.context.support.DependencyInjectionTestExecutionListener@643b1d11, org.springframework.test.context.support.DirtiesContextTestExecutionListener@2ef5e5e3, org.springframework.test.context.transaction.TransactionalTestExecutionListener@36d4b5c, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener@6d00a15d]21:38:17,025  INFO main xml.XmlBeanDefinitionReader:317 - Loading XML bean definitions from class path resource [spring-context.xml]
21:38:17,263  INFO main support.GenericApplicationContext:583 - Refreshing org.springframework.context.support.GenericApplicationContext@4f7d0008: startup date [Wed May 30 21:38:17 CST 2018]; root of context hierarchy
21:38:17,405  INFO main concurrent.ThreadPoolTaskExecutor:165 - Initializing ExecutorService 
21:38:17,407  INFO main support.PostProcessorRegistrationDelegate$BeanPostProcessorChecker:325 - Bean 'myexecutor' of type [org.springframework.scheduling.config.TaskExecutorFactoryBean] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
21:38:17,407  INFO main support.PostProcessorRegistrationDelegate$BeanPostProcessorChecker:325 - Bean 'myexecutor' of type [org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
21:38:17,692 ERROR main service.BusinessService:35 - start to deal with our business
21:38:20,833 ERROR myexecutor-1 service.CacheService:39 - success store the result to cache
21:38:20,834 ERROR main service.BusinessService:37 - Async operation success
21:38:20,835 ERROR main service.BusinessService:38 - comlete service operation
21:38:25,838  INFO Thread-0 support.GenericApplicationContext:984 - Closing org.springframework.context.support.GenericApplicationContext@4f7d0008: startup date [Wed May 30 21:38:17 CST 2018]; root of context hierarchy
21:38:25,839  INFO Thread-0 concurrent.ThreadPoolTaskExecutor:203 - Shutting down ExecutorService

SpringBoot中使用@Async

对于SpringBoot项目，要使@Async能异步执行，必须在配置类或启动上加上@EnableAsync注解（否则还是会同步执行）。

package com.lzumetal.springboot.async;

import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
@EnableAsync
public class BootStrap {

    public static void main(String[] args) {
        new SpringApplicationBuilder().run(args);
    }
}

@Async的使用还是一样的：

package com.lzumetal.springboot.async.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class AsyncService {

    @Async
    public void saveToCache() throws InterruptedException {
        log.info("将数据存入缓存|start...");
        TimeUnit.SECONDS.sleep(2L);
        log.info("将数据存入缓存|end...");
    }

}

测试类：

package com.lzumetal.springboot.async.test;

import com.lzumetal.springboot.async.BootStrap;
import com.lzumetal.springboot.async.service.AsyncService;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.concurrent.TimeUnit;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = BootStrap.class)
@Slf4j
public class ServiceTest {

    @Autowired
    private AsyncService asyncService;

    @Test
    public void testAsync() throws InterruptedException {
        log.info("do some bussiness|start...");
        asyncService.saveToCache();
        TimeUnit.SECONDS.sleep(1);
        log.info("do some bussiness|end...");
        TimeUnit.SECONDS.sleep(2);
    }

}

控制台日志输出结果：

2020-05-03 19:55:57.121  INFO 15368 --- [           main] c.l.springboot.async.test.ServiceTest    : do some bussiness|start...
2020-05-03 19:55:57.134  INFO 15368 --- [         task-1] c.l.s.async.service.AsyncService         : 将数据存入缓存|start...
2020-05-03 19:55:58.128  INFO 15368 --- [           main] c.l.springboot.async.test.ServiceTest    : do some bussiness|end...
2020-05-03 19:55:59.135  INFO 15368 --- [         task-1] c.l.s.async.service.AsyncService         : 将数据存入缓存|end...

注意：@Async默认异步配置使用的是SimpleAsyncTaskExecutor，该线程池默认来一个任务创建一个线程，这种方式效率较低，并且不可控，可能导致OOM问题。可以解决的方案是自己配置一个执行异步任务的线程池。

package com.lzumetal.springboot.async.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@Slf4j
public class AsyncConfig {

    @Bean(name = "myexecutor")
    public ThreadPoolTaskExecutor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setQueueCapacity(1000);
        executor.setKeepAliveSeconds(600);
        executor.setMaxPoolSize(20);
        executor.setThreadNamePrefix("taskExecutor-");
        executor.setRejectedExecutionHandler((Runnable r, ThreadPoolExecutor exe) -> {
            log.warn("当前任务线程池队列已满.");
        });
        executor.initialize();
        return executor;
    }
}

然后在@Async注解的value指定为’myexecutor’。

1	@Async(value = "myexecutor")

运行结果：

2020-05-03 20:17:47.722  INFO 8516 --- [           main] c.l.springboot.async.test.ServiceTest    : do some bussiness|start...
2020-05-03 20:17:47.735  INFO 8516 --- [ taskExecutor-1] c.l.s.async.service.AsyncService         : 将数据存入缓存|start...
2020-05-03 20:17:48.729  INFO 8516 --- [           main] c.l.springboot.async.test.ServiceTest    : do some bussiness|end...
2020-05-03 20:17:49.737  INFO 8516 --- [ taskExecutor-1] c.l.s.async.service.AsyncService         : 将数据存入缓存|end...

@Async的使用注意点

返回值：不需要返回值直接返回void；需要返回值则用AsyncResult或者CompletableFuture
可自定义执行器并指定例如：@Async(“otherExecutor”)
@Async必须不同类间调用： A类—>B类.C方法()（@Async注释在B类/方法中），如果在同一个类中调用，会变同步执行,例如:A类.B()—>A类.@Async C()。
@Async也可以加到类，表示这个类的所有方法都是异步执行，并且方法上的注解会覆盖类上的注解。但一般不这么用！
@Async方法必需是实例方法，因为静态方法不能被override重写，因为@Async异步方法的实现原理是通过注入一个代理类到Bean中，该代理类集成这个Bean并且需要重写这个异步方法，所以需要是实例方法。
事物：在@Async注解的方法上再使用@Transaction注解是无效的，在@Async注解的方法中调用Service层的事务方法是有效的。
如果项目是以debug的方式启动，异步功能也是不生效的。