kim.zhang

1.Spring异步任务

Spring的异步任务使用步骤:

1. 通常是在service层的某个方法中需要使用异步操作，且方法通常返回CompletableFuture类型。在方法上标注@Async注解表示该方法以异步执行。(标注@Async的类必须加入到Spring IOC容器中才有效)
2. 在配置类上添加@EnableAsync,开启允许异步操作的开关。(也可以在主启动类上添加@EnableAsync注解,因为主启动类也是一个配置类)。经过测试,如果不添加该注解,会以同步的方式执行。
3. (可选)可以配置一个Executor的bean来配置线程池的参数,且bean的方法名称为taskExecutor(官方好像有要求,但是换成其他名字好像也ok)。如果不自定义配置,Spring会提供一个默认的SimpleSyncTaskExecutor.
4. 在Spring启动的时候(实现CommandLineRunner接口)或在定时器中调用需要执行的异步方法。

经过测试,最简单的使用异步任务的方法是编写一个方法,在方法上添加@Async,开启@EnableAsync,通过定时器或者容器启动时执行该异步方法。

2.DEMO

在service中,异步调用Github的api查找用户信息:

@Service
public class LookupService {

    private static final Logger log = LoggerFactory.getLogger(LookupService.class);
    private RestTemplate restTemplate;

    public LookupService(RestTemplateBuilder builder) {
        this.restTemplate = builder.build();
    }

    @Async
    public CompletableFuture<User> findUser(String name) throws InterruptedException {
        log.info("start aysnc to Looking up user: []" , name);
        String url = String.format("https://api.github.com/users/%s", name);
        User user = null;
        try {
            user = restTemplate.getForObject(url, User.class);
        } catch (RestClientException e) {
            log.warn("remote call is encounter a issue...");
        }
        log.info("query user:" + user);
        Thread.sleep(1000L);
        return CompletableFuture.completedFuture(user);
    }
}

配置类配置Executor,@EnableSync开启异步执行任务:

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("GithubLookup-");
        // if add @Bean annotation,it is no longer necessary to manually call the executor.initialize() method
        // as this will be invoked automatically when the bean is initialized.
        executor.initialize();
        return executor;
    }
}

在容器启动的时候,执行异步任务:

@Component
public class AppRunner implements CommandLineRunner {

    private final Logger log = LoggerFactory.getLogger(AppRunner.class);
    @Autowired
    private LookupService lookupService;

    @Override
    public void run(String... args) throws Exception {
        log.info("application start and run the AppRunner class..");
        long start = System.currentTimeMillis();

        // to lookup github user info
        CompletableFuture<User> user1 = lookupService.findUser("kim.zhang");
        CompletableFuture<User> user2 = lookupService.findUser("PivotalSoftware");
        CompletableFuture<User> user3 = lookupService.findUser("Spring-Projects");

        // waiting all aysnc task executors finished
        CompletableFuture.allOf(user1, user2, user3);

        log.info("finished find user,elapsed time is []", System.currentTimeMillis() - start);
        log.info("user1 is []", user1.get());
        log.info("user2 is []", user2.get());
        log.info("user3 is []", user3.get());
    }
}

3.Executor

Spring异步线程池的接口类，其实质是java.util.concurrent.Executor。

Spring 已经实现的异常线程池：

1. SimpleAsyncTaskExecutor：
   不是真的线程池，这个类不重用线程，每次调用都会创建一个新的线程。

2. SyncTaskExecutor:

   这个类没有实现异步调用，只是一个同步操作。只适用于不需要多线程的地方

3. ConcurrentTaskExecutor：

   Executor的适配类，不推荐使用。如果ThreadPoolTaskExecutor不满足要求时，才用考虑使用这个类

4. SimpleThreadPoolTaskExecutor：

   是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用，才需要使用此类

5. ThreadPoolTaskExecutor：

   最常使用，推荐。其实质是对java.util.concurrent.ThreadPoolExecutor的包装。

需要特别注意的是Executor的initialize方法,如果添加了@Bean注解,是可以不调用这个方法的。但如果是通过实现AsyncConfigurer的方式来创建Executor,就需要调用initialize().

4.异常处理

• 对于异步方法返回值是Future的情况:
  • 方法一是在异步方法中捕获异常进行处理
  • 方法二是在调用future.get()方法时捕获异常进行处理
• 对于异步方法返回值是void的情况:
  • 通过自定义的AsyncUncaughtExceptionHandler处理异常

异步方法返回值是void:

@Async
public void findUserNoReturn(String name) throws Exception {
    log.info("start aysnc to Looking up user: []" , name);
    String url = String.format("https://api.github.com/users/%s", name);
	
    // 这里会抛出异常
    User user = restTemplate.getForObject(url, User.class);

    log.info("query user:" + user);
    Thread.sleep(1000L);
}

调用异步方法:

@Component
public class AppRunner implements CommandLineRunner {

    private final Logger log = LoggerFactory.getLogger(AppRunner.class);
    @Autowired
    private LookupService lookupService;

    @Override
    public void run(String... args) throws Exception {
        log.info("application start and run the AppRunner class..");
        long start = System.currentTimeMillis();

        lookupService.findUserNoReturn("kim.zhang");
        lookupService.findUserNoReturn("PivotalSoftware");
        lookupService.findUserNoReturn("Spring-Projects");
     
        log.info("finished find user,elapsed time is []", System.currentTimeMillis() - start);
        
    }
}

异步方法返回值是void,处理需要以下两个步骤：

• 实现AsyncConfigurer接口,并实现该接口的方法,并将其加入到IOC容器
• 自定义一个类实现AsyncUncaughtExceptionHandler接口,处理异常处理的逻辑

AsyncUncaughtExceptionHandler只会处理异步方法返回值是void的异常,不会处理返回值是Future的异常。

AsyncConfig:(不一定要是配置类，只要将实现AsyncConfigurer接口的类加入到IOC容器,并开启@EnableSync即可)

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(2);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("GithubLookup-");
        // 这里没有添加@Bean注解,需要执行initialize方法
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncExceptionHandler();
    }
}

AsyncExceptionHandler:

public class AsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

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

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        log.error("ex: {}" , ex);
        log.error("method: {}", method);
        for (int i = 0; i < params.length; i++) {
            log.error("params : {}", params[i]);
        }
    }
}