Spring自带线程池（ThreadPoolTaskExecutor）用在Spring Event，理解拒绝策略

墨蓝 2022-12-31 01:11 328阅读 0赞

#### 背景 ####

*  基础代码逻辑[Spring自带线程池（ThreadPoolTaskExecutor）用在Spring Event中，过程与理解][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：由调用线程（提交任务的线程）处理该任务
 *  线程池不能处理更多任务了，给出拒绝策略。在理解上，这不是错误，但是开发者需要对出现丢弃任务情况，有应对策略。比如记录日志、持久化，以便后续的定时任务机制进行补偿或者手动机制进行补偿。
 *  结论：无论什么框架对线程池进行了二次封装，底层的大致工作过程都是一样的。

[Spring_ThreadPoolTaskExecutor_Spring Event]: https://blog.csdn.net/outsanding/article/details/111756871