使用SpringBoot Async实现多线程操作

Spring异步任务

除自定义创建线程池外，Spring也提供了简单的异步任务实现，仅使用注解添加即可

异步任务配置

启动配置。对于使用注解的配置方法，配置类添加@EnableAsync注解。 对于XML配置方式，使用可以使用下面的配置，同时，对于线程池的大小，线程队列等参数也同样支持配置 <task:annotation-driven executor="annotationExecutor" /> <task:executor id="annotationExecutor"/> 方法配置。对于需要异步处理的方法，添加@Aync注解即可，值得注意的是被注解方法与调用方法不能在同一类内。

无返回异步任务

无返回异步任务仅在异步方法上添加注解即可：

@Slf4j @Component public class AsyncImpl implements AsyncInterface { @Async public void noResponseMethod() throws InterruptedException { log.info("this is a no response async method, thread id is:{}",Thread.currentThread().getId()); TimeUnit.SECONDS.sleep(5); } }

调用方按照正常方法的调用逻辑即可：

public class AsyncTestEntry implements AsyncTestEntryInterface { @Autowired private AsyncInterface asyncInterface; public void asyncNoResponseEntry(){ log.info("begin a no response method,thread id is:{}", Thread.currentThread().getId()); try { asyncInterface.noResponseMethod(); }catch (Exception e){ log.error("异步执行异常"); } log.info("end a no response method,thread id is:{}", Thread.currentThread().getId()); } }

带返回值的异步任务

同样，使用Spring自带的异步任务也需要使用Future来获取异步执行的结果：

@Async public Future<String> haveResponseMethod() throws InterruptedException { log.info("this is a have response async method, thread id is:{}",Thread.currentThread().getId()); TimeUnit.SECONDS.sleep(5); return AsyncResult.forValue("SUCCESS"); }

调用方使用get()方法获取异步任务的返回结果

public void asyncResponseEntry(){ log.info("begin a have response method,thread id is:{}", Thread.currentThread().getId()); try { log.info(asyncInterface.haveResponseMethod().get()); }catch (Exception e){ log.error("异步执行异常"); } log.info("end a have response method,thread id is:{}", Thread.currentThread().getId()); }

需要注意的是，获取异步任务的结果是阻塞的，异步任务没有返回前，get方法会阻塞直到异步方法抛出异常或正常返回结果。

带返回结果的异步任务回调处理

多个异步任务的情况下，对每个任务使用get方法会导致线程阻塞，避免这种情况，可以通过ListenableFuture，使用回调的形式来进行处理，为每个异步任务注册回调函数，在正常获取结果或异步抛出异常的终态处理业务；

调用方为异步任务注册回调函数，任务完成或异常时触发回调

public void asyncResponseEntry() throws InterruptedException { log.info("begin a have response method,thread id is:{}", Thread.currentThread().getId()); ListenableFuture<String> listenableFuture = asyncInterface.haveResponseMethod(); listenableFuture.addCallback(new SuccessCallback<String>() { @Override public void onSuccess(String result) { log.info("异步任务执行完成,result:{}",result); } }, new FailureCallback() { @Override public void onFailure(Throwable ex) { log.error("异步任务执行失败"); } }); try { asyncInterface.haveResponseMethod().get(); }catch (Exception e){ log.error("异步执行异常"); } log.info("end a have response method,thread id is:{}", Thread.currentThread().getId()); }

异步任务返回结果需要ListenableFuture

@Async public ListenableFuture<String> haveResponseMethod() throws InterruptedException { log.info("this is a have response async method, thread id is:{}",Thread.currentThread().getId()); TimeUnit.SECONDS.sleep(5); return AsyncResult.forValue("SUCCESS"); }

自定义线程池

除使用spring提供的默认线程池外，也可以自定义线程池并进行配置，例如：

@Configuration public class TaskConfiguration { @Bean("taskExecutor") public Executor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(20); executor.setQueueCapacity(200); executor.setKeepAliveSeconds(60); executor.setThreadNamePrefix("taskExecutor-"); executor.setRejectedExecutionHandler(new CallerRunsPolicy()); return executor; } }

使用时，仅需在@Async注解内标名线程池Bean名即可