网站后台文章编辑器,做内容网站好累,网站备案 做网站,wordpress dux商城很多业务场景需要使用异步去完成#xff0c;比如#xff1a;发送短信通知。要完成异步操作一般有两种#xff1a; 1、消息队列MQ
2、线程池处理。
我们来看看Spring框架中如何去使用线程池来完成异步操作#xff0c;以及分析背后的原理。
一. Spring异步线程池的接口类 … 很多业务场景需要使用异步去完成比如发送短信通知。要完成异步操作一般有两种 1、消息队列MQ
2、线程池处理。
我们来看看Spring框架中如何去使用线程池来完成异步操作以及分析背后的原理。
一. Spring异步线程池的接口类 TaskExecutor
在Spring4中Spring中引入了一个新的注解Async这个注解让我们在使用Spring完成异步操作变得非常方便。
Spring异步线程池的接口类其实质是 java.util.concurrent.Executor
Spring 已经实现的异常线程池
1. SimpleAsyncTaskExecutor不是真的线程池这个类不重用线程每次调用都会创建一个新的线程。
2. SyncTaskExecutor这个类没有实现异步调用只是一个同步操作。只适用于不需要多线程的地方
3. ConcurrentTaskExecutorExecutor的适配类不推荐使用。如果ThreadPoolTaskExecutor不满足要求时才用考虑使用这个类
4. SimpleThreadPoolTaskExecutor是Quartz的SimpleThreadPool的类。线程池同时被quartz和非quartz使用才需要使用此类
5. ThreadPoolTaskExecutor 最常使用推荐。 其实质是对java.util.concurrent.ThreadPoolExecutor的包装
二、简单使用说明
Spring中用Async注解标记的方法称为异步方法。在spring boot应用中使用Async很简单
1、调用异步方法类上或者启动类加上注解EnableAsync
2、在需要被异步调用的方法外加上Async
3、所使用的Async注解方法的类对象应该是Spring容器管理的bean对象
启动类加上注解EnableAsync
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;SpringBootApplication
EnableAsync
public class CollectorApplication {public static void main(String[] args) throws Exception {SpringApplication.run(CollectorApplication.class, args);}
}
在需要被异步调用的方法外加上Async同时类AsyncService加上注解Service或者Component使其对象成为Spring容器管理的bean对象
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;Service
Transactional
public class AsyncService {Asyncpublic void asyncMethod(String s) {System.out.println(receive: s);}public void test() {System.out.println(test);asyncMethod();//同一个类里面调用异步方法}Asyncpublic void test2() {AsyncService asyncService context.getBean(AsyncService.class);asyncService.asyncMethod();//异步}/*** 异布调用返回Future*/Asyncpublic FutureString asyncInvokeReturnFuture(int i) {System.out.println(asyncInvokeReturnFuture, parementer i);FutureString future;try {Thread.sleep(1000 * 1);future new AsyncResultString(success: i);} catch (InterruptedException e) {future new AsyncResultString(error);}return future;}
}//异步方法和普通的方法调用相同
asyncService.asyncMethod(123);
FutureString future asyncService.asyncInvokeReturnFuture(100);
System.out.println(future.get());
如果将一个类声明为异步类Async那么这个类对外暴露的方法全部成为异步方法。
Async
Service
public class AsyncClass {public AsyncClass() {System.out.println(----init AsyncClass----);}volatile int index 0;public void foo() {System.out.println(asyncclass foo, index: index);}public void foo(int i) {this.index i;System.out.println(asyncclass foo, index: i);}public void bar(int i) {this.index i;System.out.println(asyncclass bar, index: i);}
}
这里需要注意的是
1、同一个类里面调用异步方法不生效原因默认类内的方法调用不会被aop拦截即调用方和被调用方是在同一个类中是无法产生切面的该对象没有被Spring容器管理。即Async方法不生效。
解决办法如果要使同一个类中的方法之间调用也被拦截需要使用spring容器中的实例对象而不是使用默认的this因为通过bean实例的调用才会被spring的aop拦截
本例使用方法AsyncService asyncService context.getBean(AsyncService.class); 然后使用这个引用调用本地的方法即可达到被拦截的目的
备注这种方法只能拦截protecteddefaultpublic方法private方法无法拦截。这个是spring aop的一个机制。
2、如果不自定义异步方法的线程池默认使用SimpleAsyncTaskExecutor。SimpleAsyncTaskExecutor不是真的线程池这个类不重用线程每次调用都会创建一个新的线程。并发大的时候会产生严重的性能问题。
3、异步方法返回类型只能有两种void和 java.util.concurrent.Future。
1当返回类型为void的时候方法调用过程产生的异常不会抛到调用者层面
可以通过注 AsyncUncaughtExceptionHandler来捕获此类异常
2当返回类型为Future的时候方法调用过程产生的异常会抛到调用者层面
三、定义通用线程池
自定义线程池可对系统中线程池更加细粒度的控制方便调整线程池大小配置线程执行异常控制和处理。在设置系统自定义线程池代替默认线程池时虽可通过多种模式设置但替换默认线程池最终产生的线程池有且只能设置一个不能设置多个类继承 AsyncConfigurer。自定义线程池有如下方式
重新实现接口 AsyncConfigurer继承 AsyncConfigurerSupport配置由自定义的 TaskExecutor 替代内置的任务执行器。
通过查看 Spring 源码关于 Async 的默认调用规则会优先查询源码中实现 AsyncConfigurer 这个接口的类实现这个接口的类为 AsyncConfigurerSupport。但默认配置的线程池和异步处理方法均为空所以无论是继承或者重新实现接口都需指定一个线程池。且重新实现 public Executor getAsyncExecutor () 方法。
实现接口 AsyncConfigurer
Configurationpublic class AsyncConfiguration implements AsyncConfigurer {Bean(taskExecutor)public ThreadPoolTaskExecutor executor() {ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();int corePoolSize 10;executor.setCorePoolSize(corePoolSize);int maxPoolSize 50;executor.setMaxPoolSize(maxPoolSize);int queueCapacity 10;executor.setQueueCapacity(queueCapacity);executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.setThreadNamePrefix( asyncServiceExecutor-);executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(awaitTerminationSeconds);executor.initialize();return executor;}Overridepublic Executor getAsyncExecutor() {return executor();}}
继承 AsyncConfigurerSupport
Configuration
EnableAsync
class SpringAsyncConfigurer extends AsyncConfigurerSupport { Bean public ThreadPoolTaskExecutor asyncExecutor() { ThreadPoolTaskExecutor threadPool new ThreadPoolTaskExecutor(); threadPool.setCorePoolSize(3); threadPool.setMaxPoolSize(3); threadPool.setWaitForTasksToCompleteOnShutdown(true); threadPool.setAwaitTerminationSeconds(60 * 15); return threadPool; } Override public Executor getAsyncExecutor() { return asyncExecutor; }
}
配置自定义的 TaskExecutor (建议采用方式)
在Spring Boot主类中定义一个线程池public Executor taskExecutor() 方法用于自定义自己的线程池线程池前缀”taskExecutor-”。如果不定义则使用系统默认的线程池。
SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}EnableAsyncConfigurationclass TaskPoolConfig {Beanpublic Executor taskExecutor1() {ThreadPoolTaskExecutor pool new ThreadPoolTaskExecutor();pool.setCorePoolSize(5); //线程池活跃的线程数pool.setMaxPoolSize(10); //线程池最大活跃的线程数pool.setWaitForTasksToCompleteOnShutdown(true);pool.setThreadNamePrefix(defaultExecutor);return pool;}Bean(taskExecutor)public Executor taskExecutor2() {ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(20);executor.setQueueCapacity(200);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix(taskExecutor-);executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(60);return executor;}}
}
上面我们通过ThreadPoolTaskExecutor创建了一个线程池同时设置了如下参数
核心线程数10线程池创建时初始化的线程数
最大线程数20线程池最大的线程数只有在缓冲队列满了之后才会申请超过核心线程数的线程
缓冲队列200用来缓冲执行任务的队列
允许线程的空闲时间60秒超过了核心线程数之外的线程在空闲时间到达之后会被销毁
线程池名的前缀设置好了之后可以方便我们定位处理任务所在的线程池
线程池对拒绝任务的处理策略此处采用了CallerRunsPolicy策略当线程池没有处理能力的时候该策略会直接在execute方法的调用线程中运行被拒绝的任务如果执行程序已被关闭则会丢弃该任务
设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
设置线程池中任务的等待时间如果超过这个时候还没有销毁就强制销毁以确保应用最后能够被关闭而不是阻塞住
也可以单独类来配置线程池
/*** 线程池参数配置多个线程池实现线程池隔离Async注解默认使用系统自定义线程池可在项目中设置多个线程池在异步调用的时候指明需要调用的线程池名称比如Async(taskName)**/
EnableAsync
Configuration
public class TaskPoolConfig {Bean(taskExecutor)public Executor taskExecutor() {//返回可用处理器的Java虚拟机的数量 12int i Runtime.getRuntime().availableProcessors();System.out.println(系统最大线程数 i);ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor();//核心线程池大小executor.setCorePoolSize(16);//最大线程数executor.setMaxPoolSize(20);//配置队列容量默认值为Integer.MAX_VALUEexecutor.setQueueCapacity(99999);//活跃时间executor.setKeepAliveSeconds(60);//线程名字前缀executor.setThreadNamePrefix(asyncServiceExecutor -);//设置此执行程序应该在关闭时阻止的最大秒数以便在容器的其余部分继续关闭之前等待剩余的任务完成他们的执行executor.setAwaitTerminationSeconds(60);//等待所有的任务结束后再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);return executor;}
}
四、异步方法使用线程池
Async(“线程池名称”)指定value使用自己定义的线程池
只需要在Async注解中指定线程池名即可
Component
public class Task {//默认使用线程池Asyncpublic void doTaskOne() throws Exception {System.out.println(开始做任务);long start System.currentTimeMillis();Thread.sleep(random.nextInt(10000));long end System.currentTimeMillis();System.out.println(完成任务耗时 (end - start) 毫秒);}//根据Bean Name指定特定线程池Async(taskExecutor)public void doTaskOne() throws Exception {System.out.println(开始做任务);long start System.currentTimeMillis();Thread.sleep(random.nextInt(10000));long end System.currentTimeMillis();System.out.println(完成任务耗时 (end - start) 毫秒);}
}
注意事项一定注意
在使用Async注解时很多小伙伴都会发现异步使用失败。主要原因是异步方法的定义出了问题。
1、异步方法不能使用static修饰
2、异步类没有使用Component注解或其他注解导致spring无法扫描到异步类
3、异步方法和调用异步方法的方法不能在同一个类
4、类中需要使用Autowired或Resource等注解自动注入不能自己手动new对象
5、如果使用SpringBoot框架必须在启动类中增加EnableAsync注解
