Spring多线程-TaskExecutor任务调度
- spring通过任务执行器TaskExecutor来实现多线程与并发编程
注解方式
- 使用ThreadPoolTaskExecutor来实现一个基于线程池的TaskExecutor，首先实现AsyncConfigurer接口，开启一个线程池
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
@ComponentScan("com.yuansu.service")
@EnableAsync //利用@EnableAsync注解开启异步任务支持
public class TaskExecutorConfig implements AsyncConfigurer {
//配置类实现AsyncConfigurer接口并重写getAsyncExcutor方法，并返回一个ThreadPoolTaskExevutor
//这样我们就获得了一个基于线程池的TaskExecutor
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(5);//线程池维护线程的最少数量
taskExecutor.setMaxPoolSize(10);//线程池维护线程的最大数量
taskExecutor.setQueueCapacity(25);//线程池所使用的缓冲队列
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//设置测策略 CallerRunsPolicy策略
taskExecutor.initialize();
return taskExecutor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return null;
}
}
- 然后实现业务
- 通过@Async注解表明该方法是个异步方法，如果注解在类级别，则表明该类所有的方法都是异步方法。
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
public class TaskService {
@Async
//通过@Async注解表明该方法是个异步方法，如果注解在类级别，则表明该类所有的方法都是异步方法。
// 而这里的方法自动被注入使用ThreadPoolTaskExecutor作为TaskExecutor
public void executeAsyncTask(Integer i){
System.out.println("executeAsyncTask:"+i);
}
@Async
public void executeAsyncTaskPlus(Integer i){
System.out.println("executeAsyncTaskPlus:"+i);
}
}
- 测试
public class Main {
public static void main(String[] args) {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(TaskExecutorConfig.class);
TaskInfoService asyncTaskService = context.getBean(TaskInfoService.class);
for(int i=0;i<10;i++){
asyncTaskService.executeAsyncTask(i);
asyncTaskService.executeAsyncTaskPlus(i);
}
context.close();
}
}
- 测试结果
executeAsyncTaskPlus:0
executeAsyncTask:1
executeAsyncTask:2
executeAsyncTask:0
executeAsyncTaskPlus:3
executeAsyncTask:3
executeAsyncTaskPlus:4
executeAsyncTask:5
executeAsyncTaskPlus:2
executeAsyncTask:6
executeAsyncTaskPlus:6
executeAsyncTask:7
executeAsyncTaskPlus:7
executeAsyncTask:8
executeAsyncTaskPlus:8
executeAsyncTask:9
executeAsyncTaskPlus:9
executeAsyncTaskPlus:1
executeAsyncTaskPlus:5
executeAsyncTask:4
for循环里面，每运行一行调用方法的，就会开一个线程，异步执行
如果是同步执行，偶数行应该和奇数行值相等
配置xml方式实现
- applicationContext.xml
<bean id="xmlTaskService" class="com.yuansu.service.XmlTaskService">
<constructor-arg ref="taskExecutor" />
</bean>
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="5"/>
<property name="maxPoolSize" value="10"/>
<property name="queueCapacity" value="25"/>
</bean>
- 业务实现
import org.springframework.core.task.TaskExecutor;
import org.springframework.core.task.TaskRejectedException;
public class XmlTaskService {
private TaskExecutor taskExecutor;
public XmlTaskService(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void executeTask() {
for (int i = 0; i < 20; i++) {
try {
taskExecutor.execute(new executeAsyncTask("executeAsyncTask:" + i));
taskExecutor.execute(new executeAsyncTask("executeAsyncTaskPlus:" + i));
} catch (TaskRejectedException e) {
e.printStackTrace();
try {
Thread.sleep(200);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
}
}
class executeAsyncTask implements Runnable {
private String msg;
public executeAsyncTask(String msg) {
this.msg = msg;
}
public void run() {
System.out.println(msg);
}
}
其他实现(不依赖spring)
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class NormalTaskService {
public void executeTask() {
//有界队列
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(25);
//调用者运行策略
ThreadPoolExecutor tpe = new ThreadPoolExecutor(5, 10, 1000, TimeUnit.MILLISECONDS, queue, new ThreadPoolExecutor.CallerRunsPolicy());
for (int i = 0; i < 20; i++) {
tpe.execute(new executeAsyncTask("executeAsyncTask:" + i));
tpe.execute(new executeAsyncTask("executeAsyncTaskPlus:" + i));
}
tpe.shutdown();
}
}
class executeAsyncTask implements Runnable {
private String msg;
public executeAsyncTask(String msg) {
this.msg = msg;
}
public void run() {
System.out.println(msg);
}
}
TaskRejectedException
- Spring ThreadPoolTaskExecutor没有使用阻塞模式将任务加入到对象中，因此==对象满==的时候会抛出异常，对于这种情况，因此不得不我们自己阻塞队列的加入，在这里去==try-catch TaskRejectedException==,发现异常，Thread.sleep();模拟阻塞队列的效果
for (int i = 0; i < 10; i++) {
try {
taskService.executeAsyncTask(i);
taskService.executeAsyncTaskPlus(i);
} catch (TaskRejectedException e) {
e.printStackTrace();
try {
Thread.sleep(200);
} catch (InterruptedException e1) {
}
}
}
RejectedExecutionException
出现RejectedExecutionException异常有两种原因
- 当你设置的任务缓存队列过小，任务缓存队列也已经充满了等待的队列，向它提交任务，则会抛出这个异常
- 执行shutdown后再向线程池中添加任务
解决方案
饱和策略
-
AbortPolicy：终止策略是默认的饱和策略，当队列满时，会抛出一个RejectExecutionException,可以捕获这个异常，根据需求编写自己的处理代码
-
DiscardPolicy：策略会悄悄抛弃该任务。
-
DiscardOldestPolicy：策略将会抛弃下一个将要执行的任务，如果此策略配合优先队列PriorityBlockingQueue，该策略将会抛弃优先级最高的任务
-
CallerRunsPolicy：调用者运行策略，该策略不会抛出异常，不会抛弃任务，而是将任务回退给调用者线程执行，由于任务需要执行一段时间，所以在此期间不能提交任务，从而使工作线程有时间执行正在执行的任务。
