Spring自带线程池(ThreadPoolTaskExecutor)用在Spring Event,理解拒绝策略

背景

过程

  1. 线程池配置( corePoolSize=1;maxPoolSize=1;queueCapacity=1;)
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
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
@EnableAsync
public class ListenerAsyncConfiguration implements AsyncConfigurer
{
    /**
     * 获取异步线程池执行对象
     * @return
     */
    @Override
    @Bean("AsyncTaskExecutor")
    public Executor getAsyncExecutor() {
        //使用Spring内置线程池任务对象
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数
        taskExecutor.setCorePoolSize(1);
        //连接池中保留的最大连接数。
        taskExecutor.setMaxPoolSize(1);
        //queueCapacity 线程池所使用的缓冲队列
        taskExecutor.setQueueCapacity(0);
        taskExecutor.setThreadNamePrefix("Spring AsyncTaskExecutor Thread-");
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }

}
  1. test代码
import com.sanding.attachfile.AttachfileApplication;
import com.sanding.attachfile.event.AnotherEvent;
import com.sanding.attachfile.event.MyEvent;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.ApplicationContext;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Date;

@RunWith(SpringRunner.class)
@SpringBootTest(classes={AttachfileApplication.class})
public class SpringEventLisnerTest {
    @Autowired
    ApplicationContext applicationContext;

    @Test
    public void test() {
        // 业务实体
        MyEvent myEvent = new MyEvent(this, "北京", "范柳原", new Date());
        //发布事件
        this.applicationContext.publishEvent(myEvent);

        // 业务实体
        AnotherEvent anotherEvent = new AnotherEvent(this, "北京", "白流苏", new Date());
        //发布事件
        this.applicationContext.publishEvent(anotherEvent);
    }
}
  1. 测试结果
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@1af677f8 rejected from java.util.concurrent.ThreadPoolExecutor@7a55fb81[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:134)
	at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.submit(ThreadPoolTaskExecutor.java:341)
  1. 测试过程分析
    发布了两个事件(MyEvent, AnotherEvent)。
    在使用Springboot测试过程中,MyEvent一直在被线程执行,当发布AnotherEvent的时候,就触发了拒绝策略。线程池不接受AnotherEvent的处理过程。因此在实际情况中,需要把业务数据记录到数据库中,后续通过程序job定时任务去补偿这个AnotherEvent,且还需要做一个手动补偿机制。

小结

  • Spring自带线程池(ThreadPoolTaskExecutor),默认拒绝策略是:java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
  • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 ThreadPoolExecutor.DiscardPolicy:丢弃任务,但不抛异常。 ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新提交被拒绝的任务 ThreadPoolExecutor.CallerRunsPolicy:由调用线程(提交任务的线程)处理该任务
  • 线程池不能处理更多任务了,给出拒绝策略。在理解上,这不是错误,但是开发者需要对出现丢弃任务情况,有应对策略。比如记录日志、持久化,以便后续的定时任务机制进行补偿或者手动机制进行补偿。
  • 结论:无论什么框架对线程池进行了二次封装,底层的大致工作过程都是一样的。
上一篇:线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式


下一篇:线程池源码分析