例如, 在某个调用中,需要顺序调用 A, B, C三个过程方法;如他们都是同步调用,则需要将他们都顺序执行完毕之后,方算作过程执行完毕; 如B为一个异步的调用方法,则在执行完A之后,调用B,并不等待B完成,而是执行开始调用C,待C执行完毕之后,就意味着这个过程执行完毕了。
在Java中,一般在处理类似的场景之时,都是基于创建独立的线程去完成相应的异步调用逻辑,通过主线程和不同的线程之间的执行流程,从而在启动独立的线程之后,主线程继续执行而不会产生停滞等待的情况。或是使用TaskExecutor执行异步线程。关于TaskExecutor可看:http://www.cnblogs.com/wihainan/p/6098970.html
Spring提供了@Async来使用异步方法。
在spring boot应用中使用@Async很简单:
1、启动类加上@EnableAsync
2、在需要被异步调用的方法外加上@Async
第二种使用@Async的方式是通过xml配置,后面有说到。
一、异步线程池TaskExecutor
其实质还是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的包装
二、@Async
spring对过@Async定义异步任务
异步的方法有3种
1. 最简单的异步调用,返回值为void
2. 带参数的异步调用异步方法可以传入参数
3. 异步调用返回Future
@Component public class AsyncDemo { private static final Logger log = LoggerFactory.getLogger(AsyncDemo.class); /** * 最简单的异步调用,返回值为void */ @Async public void asyncInvokeSimplest() { log.info("asyncSimplest"); } /** * 带参数的异步调用 异步方法可以传入参数 * * @param s */ @Async public void asyncInvokeWithParameter(String s) { log.info("asyncInvokeWithParameter, parementer={}", s); } /** * 异步调用返回Future * * @param i * @return */ @Async public Future<String> asyncInvokeReturnFuture(int i) { log.info("asyncInvokeReturnFuture, parementer={}", i); Future<String> future; try { Thread.sleep(1000 * 1); future = new AsyncResult<String>("success:" + i); } catch (InterruptedException e) { future = new AsyncResult<String>("error"); } return future; } }以上的异步方法和普通的方法调用相同
asyncDemo.asyncInvokeSimplest(); asyncDemo.asyncInvokeWithException("test"); Future<String> future = asyncDemo.asyncInvokeReturnFuture(100); System.out.println(future.get());
三、Spring 开启异步配置
Spring有两种方法启动配置
1. 注解:@EnableAsync
2. XML
1、通过注解开启异步
1. 在启动类处加上@EnableAsync
2. public AsyncTaskExecutor taskExecutor() 方法用于自定义自己的线程池,线程池前缀”Anno-Executor”。如果不定义,则使用系统默认的线程池。
@SpringBootApplication @EnableAsync // 启动异步调用 public class AsyncApplicationWithAnnotation { private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithAnnotation.class); /** * 自定义异步线程池 * @return */ @Bean public AsyncTaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setThreadNamePrefix("Anno-Executor"); executor.setMaxPoolSize(10); // 设置拒绝策略 executor.setRejectedExecutionHandler(new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // ..... } }); // 使用预定义的异常处理类 // executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); return executor; } public static void main(String[] args) { log.info("Start AsyncApplication.. "); SpringApplication.run(AsyncApplicationWithAnnotation.class, args); } }
2、通过XML实现
Bean文件配置: spring_async.xml
1. 线程的前缀为xmlExecutor
2. 启动异步线程池配置
<!-- 等价于 @EnableAsync, executor指定线程池 --> <task:annotation-driven executor="xmlExecutor"/> <!-- id指定线程池产生线程名称的前缀 --> <task:executor id="xmlExecutor" pool-size="5-25" queue-capacity="100" keep-alive="120" rejection-policy="CALLER_RUNS"/>
启动类导入xml文件
@SpringBootApplication @ImportResource("classpath:/async/spring_async.xml") public class AsyncApplicationWithXML { private static final Logger log = LoggerFactory.getLogger(AsyncApplicationWithXML.class); public static void main(String[] args) { log.info("Start AsyncApplication.. "); SpringApplication.run(AsyncApplicationWithXML.class, args); } }
四、对异步方法的异常处理
在调用方法时,可能出现方法中抛出异常的情况。在异步中主要有有两种异常处理方法:
1. 对于方法返回值是Futrue的异步方法: a) 一种是在调用future的get时捕获异常; b) 在异常方法中直接捕获异常
2. 对于返回值是void的异步方法:通过AsyncUncaughtExceptionHandler处理异常
AsyncExceptionDemo:
@Component public class AsyncExceptionDemo { private static final Logger log = LoggerFactory.getLogger(AsyncExceptionDemo.class); /** * 最简单的异步调用,返回值为void */ @Async public void asyncInvokeSimplest() { log.info("asyncSimplest"); } /** * 带参数的异步调用 异步方法可以传入参数 * 对于返回值是void,异常会被AsyncUncaughtExceptionHandler处理掉 * @param s */ @Async public void asyncInvokeWithException(String s) { log.info("asyncInvokeWithParameter, parementer={}", s); throw new IllegalArgumentException(s); } /** * 异常调用返回Future * 对于返回值是Future,不会被AsyncUncaughtExceptionHandler处理,需要我们在方法中捕获异常并处理 * 或者在调用方在调用Futrue.get时捕获异常进行处理 * * @param i * @return */ @Async public Future<String> asyncInvokeReturnFuture(int i) { log.info("asyncInvokeReturnFuture, parementer={}", i); Future<String> future; try { Thread.sleep(1000 * 1); future = new AsyncResult<String>("success:" + i); throw new IllegalArgumentException("a"); } catch (InterruptedException e) { future = new AsyncResult<String>("error"); } catch(IllegalArgumentException e){ future = new AsyncResult<String>("error-IllegalArgumentException"); } return future; } }实现AsyncConfigurer接口对异常线程池更加细粒度的控制
a) 创建线程自己的线程池
b) 对void方法抛出的异常处理的类AsyncUncaughtExceptionHandler
/** * 通过实现AsyncConfigurer自定义异常线程池,包含异常处理 * * @author hry * */ @Service public class MyAsyncConfigurer implements AsyncConfigurer{ private static final Logger log = LoggerFactory.getLogger(MyAsyncConfigurer.class); @Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor(); threadPool.setCorePoolSize(1); threadPool.setMaxPoolSize(1); threadPool.setWaitForTasksToCompleteOnShutdown(true); threadPool.setAwaitTerminationSeconds(60 * 15); threadPool.setThreadNamePrefix("MyAsync-"); threadPool.initialize(); return threadPool; } @Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return new MyAsyncExceptionHandler(); } /** * 自定义异常处理类 * @author hry * */ class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler { @Override public void handleUncaughtException(Throwable throwable, Method method, Object... obj) { log.info("Exception message - " + throwable.getMessage()); log.info("Method name - " + method.getName()); for (Object param : obj) { log.info("Parameter value - " + param); } } } }
五、@Async调用中的事务处理机制
在@Async标注的方法,同时也适用了@Transactional进行了标注;在其调用数据库操作之时,将无法产生事务管理的控制,原因就在于其是基于异步处理的操作。
那该如何给这些操作添加事务管理呢?可以将需要事务管理操作的方法放置到异步方法内部,在内部被调用的方法上添加@Transactional.
例如: 方法A,使用了@Async/@Transactional来标注,但是无法产生事务控制的目的。方法B,使用了@Async来标注, B中调用了C、D,C/D分别使用@Transactional做了标注,则可实现事务控制的目的。
参考资料:
https://blog.csdn.net/hry2015/article/details/67640534
https://www.cnblogs.com/wihainan/p/6516858.html