Spring Boot 2.x基础教程:配置线程池的拒绝策略

简介: Spring Boot 2.x基础教程:配置线程池的拒绝策略

通过之前三篇关于Spring Boot异步任务实现的博文,我们分别学会了用@Async创建异步任务为异步任务配置线程池使用多个线程池隔离不同的异步任务。今天这篇,我们继续对上面的知识进行完善和优化!

如果你已经看过上面几篇内容并已经掌握之后,一起来思考下面这个问题:

假设,线程池配置为核心线程数2、最大线程数2、缓冲队列长度2。此时,有5个异步任务同时开始,会发生什么?

场景重现

我们先来把上面的假设用代码实现一下:

第一步:创建Spring Boot应用,根据上面的假设写好线程池配置。

@EnableAsync
@SpringBootApplication
public class Chapter78Application {
    public static void main(String[] args) {
        SpringApplication.run(Chapter78Application.class, args);
    }
    @EnableAsync
    @Configuration
    class TaskPoolConfig {
        @Bean
        public Executor taskExecutor1() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(2);
            executor.setMaxPoolSize(2);
            executor.setQueueCapacity(2);
            executor.setKeepAliveSeconds(60);
            executor.setThreadNamePrefix("executor-1-");
            return executor;
        }
    }
}

第二步:用@Async注解实现一个部分任务

@Slf4j
@Component
public class AsyncTasks {
    public static Random random = new Random();
    @Async("taskExecutor1")
    public CompletableFuture<String> doTaskOne(String taskNo) throws Exception {
        log.info("开始任务:{}", taskNo);
        long start = System.currentTimeMillis();
        Thread.sleep(random.nextInt(10000));
        long end = System.currentTimeMillis();
        log.info("完成任务:{},耗时:{} 毫秒", taskNo, end - start);
        return CompletableFuture.completedFuture("任务完成");
    }
}

第三步:编写测试用例

@Slf4j
@SpringBootTest
public class Chapter78ApplicationTests {
    @Autowired
    private AsyncTasks asyncTasks;
    @Test
    public void test2() throws Exception {
        // 线程池配置:core-2,max-2,queue=2,同时有5个任务,出现下面异常:
        // org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@59901c4d[Running, pool size = 2,
        // active threads = 0, queued tasks = 2, completed tasks = 4]] did not accept task: java.util.concurrent.CompletableFuture$AsyncSupply@408e96d9
        long start = System.currentTimeMillis();
        // 线程池1
        CompletableFuture<String> task1 = asyncTasks.doTaskOne("1");
        CompletableFuture<String> task2 = asyncTasks.doTaskOne("2");
        CompletableFuture<String> task3 = asyncTasks.doTaskOne("3");
        CompletableFuture<String> task4 = asyncTasks.doTaskOne("4");
        CompletableFuture<String> task5 = asyncTasks.doTaskOne("5");
        // 一起执行
        CompletableFuture.allOf(task1, task2, task3, task4, task5).join();
        long end = System.currentTimeMillis();
        log.info("任务全部完成,总耗时:" + (end - start) + "毫秒");
    }
}

执行一下,可以类似下面这样的日志信息:

2021-09-22 17:33:08.159  INFO 21119 --- [   executor-1-2] com.didispace.chapter78.AsyncTasks       : 开始任务:2
2021-09-22 17:33:08.159  INFO 21119 --- [   executor-1-1] com.didispace.chapter78.AsyncTasks       : 开始任务:1
org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@3e1a3801[Running, pool size = 2, active threads = 2, queued tasks = 2, completed tasks = 0]] did not accept task: java.util.concurrent.CompletableFuture$AsyncSupply@64968732
  at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:324)
  at java.util.concurrent.CompletableFuture.asyncSupplyStage(CompletableFuture.java:1604)
  at java.util.concurrent.CompletableFuture.supplyAsync(CompletableFuture.java:1830)
  at org.springframework.aop.interceptor.AsyncExecutionAspectSupport.doSubmit(AsyncExecutionAspectSupport.java:274)
  at org.springframework.aop.interceptor.AsyncExecutionInterceptor.invoke(AsyncExecutionInterceptor.java:129)
  at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
  at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:750)
  at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:692)
  at com.didispace.chapter78.AsyncTasks$$EnhancerBySpringCGLIB$$c7e8d57b.doTaskOne(<generated>)
  at com.didispace.chapter78.Chapter78ApplicationTests.test2(Chapter78ApplicationTests.java:51)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
  at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
  at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
  at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
  at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
  at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
  at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
  at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
  at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
  at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
  at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
  at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
  at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
  at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
  at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$6(TestMethodTestDescriptor.java:210)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:206)
  at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:131)
  at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:65)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
  at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
  at java.util.ArrayList.forEach(ArrayList.java:1255)
  at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
  at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
  at java.util.ArrayList.forEach(ArrayList.java:1255)
  at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
  at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
  at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
  at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
  at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:32)
  at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
  at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:51)
  at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:108)
  at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
  at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
  at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
  at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
  at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:96)
  at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:75)
  at com.intellij.junit5.JUnit5IdeaTestRunner.startRunnerWithArgs(JUnit5IdeaTestRunner.java:71)
  at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
  at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
  at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$AsyncSupply@64968732 rejected from java.util.concurrent.ThreadPoolExecutor@3e1a3801[Running, pool size = 2, active threads = 2, queued tasks = 2, 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 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:321)
  ... 74 more

从异常信息org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@3e1a3801[Running, pool size = 2, active threads = 2, queued tasks = 2, completed tasks = 0]] did not accept task:中,可以很明确的知道,第5个任务因为超过了执行线程+缓冲队列长度,而被拒绝了。

所有,默认情况下,线程池的拒绝策略是:当线程池队列满了,会丢弃这个任务,并抛出异常。

配置拒绝策略

虽然线程池有默认的拒绝策略,但实际开发过程中,有些业务场景,直接拒绝的策略往往并不适用,有时候我们可能会选择舍弃最早开始执行而未完成的任务、也可能会选择舍弃刚开始执行而未完成的任务等更贴近业务需要的策略。所以,为线程池配置其他拒绝策略或自定义拒绝策略是很常见的需求,那么这个要怎么实现呢?

下面就来具体说说今天的正题,如何为线程池配置拒绝策略、如何自定义拒绝策略。

看下面这段代码的最后一行,setRejectedExecutionHandler方法就是为线程池设置拒绝策略的方法:

ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
//...其他线程池配置
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
在ThreadPoolExecutor中提供了4种线程的策略可以供开发者直接使用,你只需要像下面这样设置即可:
// AbortPolicy策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
// DiscardPolicy策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
// DiscardOldestPolicy策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
// CallerRunsPolicy策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

这四个策略对应的含义分别是:

  • AbortPolicy策略:默认策略,如果线程池队列满了丢掉这个任务并且抛出RejectedExecutionException异常。
  • DiscardPolicy策略:如果线程池队列满了,会直接丢掉这个任务并且不会有任何异常。
  • DiscardOldestPolicy策略:如果队列满了,会将最早进入队列的任务删掉腾出空间,再尝试加入队列。
  • CallerRunsPolicy策略:如果添加到线程池失败,那么主线程会自己去执行该任务,不会等待线程池中的线程去执行。

而如果你要自定义一个拒绝策略,那么可以这样写:

executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 拒绝策略的逻辑
    }
});

当然如果你喜欢用Lamba表达式,也可以这样写:

executor.setRejectedExecutionHandler((r, executor1) -> {
    // 拒绝策略的逻辑
});

好了,今天的学习就到这里!

如果您学习过程中如遇困难?可以加入我们超高质量的Spring技术交流群,参与交流与讨论,更好的学习与进步!更多Spring Boot教程可以点击直达!,欢迎收藏与转发支持!

代码示例

本文的完整工程可以查看下面仓库中2.x目录下的chapter7-8工程:

如果您觉得本文不错,欢迎Star支持,您的关注是我坚持的动力!

目录
相关文章
|
4月前
|
Java Spring
Spring Boot配置的优先级?
在Spring Boot项目中,配置可通过配置文件和外部配置实现。支持的配置文件包括application.properties、application.yml和application.yaml,优先级依次降低。外部配置常用方式有Java系统属性(如-Dserver.port=9001)和命令行参数(如--server.port=10010),其中命令行参数优先级高于系统属性。整体优先级顺序为:命令行参数 &gt; Java系统属性 &gt; application.properties &gt; application.yml &gt; application.yaml。
913 0
|
1月前
|
JavaScript Java Maven
【SpringBoot(二)】带你认识Yaml配置文件类型、SpringMVC的资源访问路径 和 静态资源配置的原理!
SpringBoot专栏第二章,从本章开始正式进入SpringBoot的WEB阶段开发,本章先带你认识yaml配置文件和资源的路径配置原理,以方便在后面的文章中打下基础
264 3
|
2月前
|
缓存 Java 应用服务中间件
Spring Boot配置优化:Tomcat+数据库+缓存+日志,全场景教程
本文详解Spring Boot十大核心配置优化技巧,涵盖Tomcat连接池、数据库连接池、Jackson时区、日志管理、缓存策略、异步线程池等关键配置,结合代码示例与通俗解释,助你轻松掌握高并发场景下的性能调优方法,适用于实际项目落地。
516 5
|
2月前
|
传感器 Java 数据库
探索Spring Boot的@Conditional注解的上下文配置
Spring Boot 的 `@Conditional` 注解可根据不同条件动态控制 Bean 的加载,提升应用的灵活性与可配置性。本文深入解析其用法与优势,并结合实例展示如何通过自定义条件类实现环境适配的智能配置。
170 0
探索Spring Boot的@Conditional注解的上下文配置
|
8月前
|
缓存 Java API
微服务——SpringBoot使用归纳——Spring Boot集成 Swagger2 展现在线接口文档——Swagger2 的配置
本文介绍了在Spring Boot中配置Swagger2的方法。通过创建一个配置类,添加`@Configuration`和`@EnableSwagger2`注解,使用Docket对象定义API文档的详细信息,包括标题、描述、版本和包路径等。配置完成后,访问`localhost:8080/swagger-ui.html`即可查看接口文档。文中还提示了可能因浏览器缓存导致的问题及解决方法。
989 0
微服务——SpringBoot使用归纳——Spring Boot集成 Swagger2 展现在线接口文档——Swagger2 的配置
|
3月前
|
安全 算法 Java
在Spring Boot中应用Jasypt以加密配置信息。
通过以上步骤,可以在Spring Boot应用中有效地利用Jasypt对配置信息进行加密,这样即使配置文件被泄露,其中的敏感信息也不会直接暴露给攻击者。这是一种在不牺牲操作复杂度的情况下提升应用安全性的简便方法。
953 10
|
8月前
|
Java 关系型数据库 数据库
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——Spring Boot 事务配置
本文介绍了 Spring Boot 中的事务配置与使用方法。首先需要导入 MySQL 依赖,Spring Boot 会自动注入 `DataSourceTransactionManager`,无需额外配置即可通过 `@Transactional` 注解实现事务管理。接着通过创建一个用户插入功能的示例,展示了如何在 Service 层手动抛出异常以测试事务回滚机制。测试结果表明,数据库中未新增记录,证明事务已成功回滚。此过程简单高效,适合日常开发需求。
1107 0
|
8月前
|
Java 测试技术 微服务
微服务——SpringBoot使用归纳——Spring Boot中的项目属性配置——少量配置信息的情形
本课主要讲解Spring Boot项目中的属性配置方法。在实际开发中,测试与生产环境的配置往往不同,因此不应将配置信息硬编码在代码中,而应使用配置文件管理,如`application.yml`。例如,在微服务架构下,可通过配置文件设置调用其他服务的地址(如订单服务端口8002),并利用`@Value`注解在代码中读取这些配置值。这种方式使项目更灵活,便于后续修改和维护。
147 0
|
8月前
|
SQL Java 数据库连接
微服务——SpringBoot使用归纳——Spring Boot使用slf4j进行日志记录—— application.yml 中对日志的配置
在 Spring Boot 项目中,`application.yml` 文件用于配置日志。通过 `logging.config` 指定日志配置文件(如 `logback.xml`),实现日志详细设置。`logging.level` 可定义包的日志输出级别,例如将 `com.itcodai.course03.dao` 包设为 `trace` 级别,便于开发时查看 SQL 操作。日志级别从高到低为 ERROR、WARN、INFO、DEBUG,生产环境建议调整为较高级别以减少日志量。本课程采用 yml 格式,因其层次清晰,但需注意格式要求。
772 0

热门文章

最新文章