设为首页收藏本站
开启辅助访问

SpringBoot使用线程池

 您所在的位置：网站首页 › 线程池spring › SpringBoot使用线程池

SpringBoot使用线程池

2023-11-27 10:47| 来源: 网络整理| 查看: 265

SpringBoot使用线程池 软件环境 名称版本号jdk1.8springboot2.1.6maven3.3.9 1.Java中创建线程池

只会介绍java中线程池的核心类ThreadPoolExecutor，其他用法请自行查询

1.1 ThreadPoolExecutor类介绍

jdk1.8 源码 删减部分内容

package java.util.concurrent; /** * @param corePoolSize 核心线程数 -> 线程池中保持的线程数量,即使它们是空闲的也不会销毁, * 除非设置了{@code allowCoreThreadTimeOut}核心线程超时时间 * @param maximumPoolSize 最大线程数 -> 线程池中允许接收的最大线程数量 * 如果设定的数量比系统支持的线程数还要大时,会抛出OOM(OutOfMemoryError)异常 * @param keepAliveTime 最大存活时间 -> 当前线程数大于核心线程数的时候, * 其他多余的线程接收新任务之前的最大等待时间,超过时间没有新任务就会销毁. * @param unit {@code keepAliveTime}最大存活时间的单位.eg:TimeUnit.SECONDS * @param workQueue 工作队列 -> 保存任务直到任务被提交到线程池的线程中执行. * @param threadFactory 线程工厂 -> 当线程池需要创建线程得时候会从线程工厂获取新的实例. * (自定义ThreadFactory可以跟踪线程池究竟何时创建了多少线程,也可以自定义线程的名称、 * 组以及优先级等信息,甚至可以任性的将线程设置为守护线程. * 总之,自定义ThreadFactory可以更加自由的设置线程池中所有线程的状态。) * @param handler 当线程数量等于最大线程数并且工作队列已满的时候,再有新的任务添加进来就会进入这个handler, * 可以理解为设置拒绝策略（此处不清楚的可以看一下ThreadPoolExecutor中的execute方法的注释） */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { }

ThreadPoolExecutor的执行流程如下：

提交新任务 YES NO YES NO YES NO 主线程 线程池 线程数 @Bean public Executor threadPoolTaskExecutor() { ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor(); // 设置核心线程数 threadPoolTaskExecutor.setCorePoolSize(5); // 设置最大线程数 threadPoolTaskExecutor.setMaxPoolSize(5); // 设置工作队列大小 threadPoolTaskExecutor.setQueueCapacity(2000); // 设置线程名称前缀 threadPoolTaskExecutor.setThreadNamePrefix("threadPoolTaskExecutor-->"); // 设置拒绝策略.当工作队列已满,线程数为最大线程数的时候,接收新任务抛出RejectedExecutionException异常 threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); // 初始化线程池 threadPoolTaskExecutor.initialize(); return threadPoolTaskExecutor; } } 调用方法Service import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; @Service public class HelloService { Logger logger = LoggerFactory.getLogger(HelloService.class); /** * @Async标注的方法，称之为异步方法；这些方法将在执行的时候， * 将会在独立的线程中被执行，调用者无需等待它的完成，即可继续其他的操作。 */ @Async // 使用异步方法 public void sayHello() { logger.info("start say hello"); System.out.println(Thread.currentThread().getName()); System.out.println("hello"); logger.info("end say hello"); } } 测试类 import com.cain.threadpool.ThreadPoolApplication; 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.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest(classes = ThreadPoolApplication.class) public class HelloServiceTest { @Autowired HelloService helloService; @Test public void testSayHello() throws Exception { helloService.sayHello(); } } 测试结果 2019-07-02 18:36:25.138 INFO 2868 --- [ main] t.c.e.demo.service.HelloServiceTest : Starting HelloServiceTest on DLC00R90RK7NBL with PID 2868 (started by jiaxin.chi in C:\Users\jiaxin.chi\Desktop\demo) 2019-07-02 18:36:25.140 INFO 2868 --- [ main] t.c.e.demo.service.HelloServiceTest : No active profile set, falling back to default profiles: default 2019-07-02 18:36:26.892 INFO 2868 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 2019-07-02 18:36:26.913 INFO 2868 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'threadPoolTaskExecutor' 2019-07-02 18:36:28.644 INFO 2868 --- [ main] t.c.e.demo.service.HelloServiceTest : Started HelloServiceTest in 3.98 seconds (JVM running for 6.103) 2019-07-02 18:36:29.047 INFO 2868 --- [askExecutor-->1] com.example.demo.service.HelloService : start say hello threadPoolTaskExecutor-->1 hello 2019-07-02 18:36:29.048 INFO 2868 --- [askExecutor-->1] com.example.demo.service.HelloService : end say hello 2019-07-02 18:36:29.051 INFO 2868 --- [ Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'threadPoolTaskExecutor'

从测试的结果可以清晰的看到sayHello方法是由我们定义的线程池中的线程执行的

注意 因为显示名称长度限制的原因我们看到的是askExecutor–>1, 但是通过在方法中打印当前线程的名字得知确实是我们设置的线程threadPoolTaskExecutor–>1

3.2 使用ThreadPoolExecutor 在config类中增加如下配置 @Bean public Executor myThreadPool() { // 设置核心线程数 int corePoolSize = 5; // 设置最大线程数 int maxPoolSize = 5; // 设置工作队列大小 int queueCapacity = 2000; // 最大存活时间 long keepAliveTime = 30; // 设置线程名称前缀 String threadNamePrefix = "myThreadPool-->"; // 设置自定义拒绝策略.当工作队列已满,线程数为最大线程数的时候,接收新任务抛出RejectedExecutionException异常 RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { throw new RejectedExecutionException("自定义的RejectedExecutionHandler"); } }; // 自定义线程工厂 ThreadFactory threadFactory = new ThreadFactory() { private int i = 1; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + i); i++; return thread; } }; // 初始化线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue(queueCapacity), threadFactory, rejectedExecutionHandler); return threadPoolExecutor; }

可以看到我们在config类中配置了两个线程池,如果我们想要指定使用其中一个线程池的需使用如下方式

当未指明使用哪个线程池的时候会优先使用ThreadPoolTaskExecutor

@Async("myThreadPool") // 参数为线程池配置时的方法名即对应的bean的id ① public void sayHello() { logger.info("start say hello"); System.out.println(Thread.currentThread().getName()); System.out.println("hello"); logger.info("end say hello"); } 测试结果 2019-07-03 10:26:55.515 INFO 13304 --- [ main] t.c.e.demo.service.HelloServiceTest : Starting HelloServiceTest on DLC00R90RK7NBL with PID 13304 (started by jiaxin.chi in C:\Users\jiaxin.chi\Desktop\demo) 2019-07-03 10:26:55.517 INFO 13304 --- [ main] t.c.e.demo.service.HelloServiceTest : No active profile set, falling back to default profiles: default 2019-07-03 10:26:56.768 INFO 13304 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 2019-07-03 10:26:56.789 INFO 13304 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'threadPoolTaskExecutor' 2019-07-03 10:26:57.997 INFO 13304 --- [ main] t.c.e.demo.service.HelloServiceTest : Started HelloServiceTest in 2.824 seconds (JVM running for 4.462) 2019-07-03 10:26:58.258 INFO 13304 --- [yThreadPool-->1] com.example.demo.service.HelloService : start say hello myThreadPool-->1 hello 2019-07-03 10:26:58.258 INFO 13304 --- [yThreadPool-->1] com.example.demo.service.HelloService : end say hello 2019-07-03 10:26:58.260 INFO 13304 --- [ Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'threadPoolTaskExecutor' 3.3 自定义ThreadPoolTaskExecutor 创建MyThreadPoolTaskExecutor import java.util.concurrent.Callable; import java.util.concurrent.Future; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.concurrent.ListenableFuture; public class MyThreadPoolTaskExecutor extends ThreadPoolTaskExecutor { Logger logger = LoggerFactory.getLogger(MyThreadPoolTaskExecutor.class); @Override public void execute(Runnable task) { logThreadPoolStatus(); super.execute(task); } @Override public void execute(Runnable task, long startTimeout) { logThreadPoolStatus(); super.execute(task, startTimeout); } @Override public Future submit(Runnable task) { logThreadPoolStatus(); return super.submit(task); } @Override public Future submit(Callable task) { logThreadPoolStatus(); return super.submit(task); } @Override public ListenableFuture submitListenable(Runnable task) { logThreadPoolStatus(); return super.submitListenable(task); } @Override public ListenableFuture submitListenable(Callable task) { logThreadPoolStatus(); return super.submitListenable(task); } /** * 在线程池运行的时候输出线程池的基本信息 */ private void logThreadPoolStatus() { logger.info("核心线程数:{}, 最大线程数:{}, 当前线程数: {}, 活跃的线程数: {}", getCorePoolSize(), getMaxPoolSize(), getPoolSize(), getActiveCount()); } }

我们可以在自定义的ThreadPoolTaskExecutor中,输出一些线程池的当前状态,包括所有上面介绍的参数.

在config类中增加配置 @Bean public Executor myThreadPoolTaskExecutor() { ThreadPoolTaskExecutor threadPoolTaskExecutor = new MyThreadPoolTaskExecutor(); // 设置核心线程数 threadPoolTaskExecutor.setCorePoolSize(5); // 设置最大线程数 threadPoolTaskExecutor.setMaxPoolSize(5); // 设置工作队列大小 threadPoolTaskExecutor.setQueueCapacity(2000); // 设置线程名称前缀 threadPoolTaskExecutor.setThreadNamePrefix("myThreadPoolTaskExecutor-->"); // 设置拒绝策略.当工作队列已满,线程数为最大线程数的时候,接收新任务抛出RejectedExecutionException异常 threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); // 初始化线程池 threadPoolTaskExecutor.initialize(); return threadPoolTaskExecutor; }

只需将ThreadPoolTaskExecutor的实例化对象换成自定义的即可

3.4 基于@Async返回值的调用

以上示例都是基于@Async无返回值的调用,下面介绍一下有返回值的调用

增加一个实体对象 public class HelloEntity { private String helloStr; public String getHelloStr() { return helloStr; } public void setHelloStr(String helloStr) { this.helloStr = helloStr; } } 在service中增加以下方法 @Async public Future getHelloString() { logger.info("start getHelloString"); HelloEntity helloEntity = new HelloEntity(); helloEntity.setHelloStr("Say hello to little wang"); System.out.println(Thread.currentThread().getName()); logger.info("end getHelloString"); return new AsyncResult(helloEntity); }

如果对Future的使用不熟悉的,建议学习一下jdk中的JUC包

AsyncResult 是spring封装的和@Async配合使用的异步返回结果

Tips: 网上有说: 在异步方法中，如果出现了异常，对于调用者而言是无法感知的。 如果确实需要处理异常，则需要自定义实现AsyncTaskExecutor。 查看AsyncResult的源码 [As of Spring 4.2, this class also supports passing execution exceptions back to the caller.] 也就说从Spring4.2开始已经支持传递异常的,所以说建议大家学习的时候不要太依赖网上的教程. 因为网上的教程和你当前的所使用的spring版本号不一定是一样的. 当自己的程序和网上的教程有所分歧的时候,建议大家看看源码. 测试类 @Test public void testGetHelloString() throws Exception { Future helloString = helloService.getHelloString(); HelloEntity helloEntity = helloString.get(); System.out.println(helloEntity.getHelloStr()); } 测试结果 2019-07-03 11:01:11.603 INFO 13492 --- [ main] t.c.e.demo.service.HelloServiceTest : Starting HelloServiceTest on DLC00R90RK7NBL with PID 13492 (started by jiaxin.chi in C:\Users\jiaxin.chi\Desktop\demo) 2019-07-03 11:01:11.604 INFO 13492 --- [ main] t.c.e.demo.service.HelloServiceTest : No active profile set, falling back to default profiles: default 2019-07-03 11:01:13.205 INFO 13492 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 2019-07-03 11:01:13.226 INFO 13492 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'threadPoolTaskExecutor' 2019-07-03 11:01:14.729 INFO 13492 --- [ main] t.c.e.demo.service.HelloServiceTest : Started HelloServiceTest in 3.613 seconds (JVM running for 5.579) 2019-07-03 11:01:15.070 INFO 13492 --- [askExecutor-->1] com.example.demo.service.HelloService : start getHelloString threadPoolTaskExecutor-->1 2019-07-03 11:01:15.071 INFO 13492 --- [askExecutor-->1] com.example.demo.service.HelloService : end getHelloString Say hello to little wang 2019-07-03 11:01:15.087 INFO 13492 --- [ Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'threadPoolTaskExecutor' 3.5 使用默认配置的线程池

我们不需要配置任何config类只需要在启动类中加上以下注解

ThreadPoolApplication import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; @SpringBootApplication @EnableAsync // 允许使用异步方法 public class ThreadPoolApplication { public static void main(String[] args) { SpringApplication.run(ThreadPoolApplication.class, args); } } service @Async public void sayHello() { logger.info("start say hello"); System.out.println(Thread.currentThread().getName()); System.out.println("hello"); logger.info("end say hello"); } 测试方法 @Test public void testSayHello() throws Exception { helloService.sayHello(); } 测试结果 2019-07-04 17:26:57.406 INFO 20620 --- [ main] c.c.threadpool.service.HelloServiceTest : Starting HelloServiceTest on DLC00R90RK7NBL with PID 20620 (started by jiaxin.chi in E:\threadpooltest) 2019-07-04 17:26:57.407 INFO 20620 --- [ main] c.c.threadpool.service.HelloServiceTest : No active profile set, falling back to default profiles: default 2019-07-04 17:26:59.757 INFO 20620 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor' 2019-07-04 17:27:00.458 INFO 20620 --- [ main] c.c.threadpool.service.HelloServiceTest : Started HelloServiceTest in 3.337 seconds (JVM running for 4.728) 2019-07-04 17:27:00.880 INFO 20620 --- [ task-1] c.cain.threadpool.service.HelloService : start say hello task-1 hello 2019-07-04 17:27:00.880 INFO 20620 --- [ task-1] c.cain.threadpool.service.HelloService : end say hello 2019-07-04 17:27:00.887 INFO 20620 --- [ Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor' Disconnected from the target VM, address: '127.0.0.1:49464', transport: 'socket'

通过日志可以看出默认使用的就是ThreadPoolTaskExecutor这个类

Tips: 当我们在项目中使用线程池的时候,还是要根据项目的实际情况来设置线程池的参数 4 实战演示

分享一个我在项目中的使用场景

有登录和人脸识别两个服务.

当用户登录的时候,需要去数据库查询用户的账号状态做一些业务逻辑的判断,同时也需要去进行在线人脸识别. 只有两个校验都通过的时候才会成功登陆. 这个时候我们就可以用异步的调用在线人脸识别的接口,从而加快系统的响应时间.

4.1 代码演示 LoginService import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import com.cain.threadpool.entity.CainResult; import com.cain.threadpool.entity.FaceVerificationResult; import com.cain.threadpool.entity.UserInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Service; @Service public class LoginService { Logger logger = LoggerFactory.getLogger(LoginService.class); @Autowired FaceVerifiationService faceVerifiationService; @Autowired ApplicationContext applicationContext; ② public CainResult login(UserInfo userInfo) { CainResult cainResult = new CainResult(); LoginService loginService = applicationContext.getBean(LoginService.class); Future onlineFaceVerificationResult = loginService.getOnlineFaceVerificationResult(userInfo); try { // 模拟从数据获取数据判断用户账号状态是否冻结的消耗时间 Thread.sleep(4000); } catch (InterruptedException e) { e.printStackTrace(); } FaceVerificationResult faceVerificationResult = null; try { // 如果3秒还没有结果直接人脸识别失败 faceVerificationResult = onlineFaceVerificationResult.get(3, TimeUnit.SECONDS); } catch (Exception e) { logger.error("获取人脸识别结果失败"); // 异常处理 } if (faceVerificationResult != null) { cainResult.setCode(faceVerificationResult.getCode()); cainResult.setMessage(faceVerificationResult.getMessage()); } return cainResult; } @Async public Future getOnlineFaceVerificationResult(UserInfo userInfo) { logger.info("异步执行"); FaceVerificationResult result = faceVerifiationService.getResult(userInfo); return new AsyncResult(result); } } 封装的Bean public class CainResult { private Integer code; private String message; } public class FaceVerificationResult { private int code; private String message; } public class UserInfo { private Integer id; private String name; private String photo; } FaceVerifiationService package com.cain.threadpool.service; import com.cain.threadpool.entity.FaceVerificationResult; import com.cain.threadpool.entity.UserInfo; import org.springframework.stereotype.Service; @Service public class FaceVerifiationService { public FaceVerificationResult getResult(UserInfo userInfo) { FaceVerificationResult faceVerificationResult = new FaceVerificationResult(); // 成功返回 100 faceVerificationResult.setCode(100); faceVerificationResult.setMessage("success"); try { // 模拟调用人脸识别接口的http请求消耗的时间 Thread.sleep(1000); // 如果需要模拟异常的情况那么时间应该在 // 4(主方法的逻辑处理时间) + 3(允许的超时时间) 也就是7s以上才可以 } catch (InterruptedException e) { e.printStackTrace(); } return faceVerificationResult; } }

代码中用Thread.sleep()模拟了两个方法.一个是调用数据库和业务逻辑处理,一个是调用人脸识别的接口.

如果两个方法并行,不考虑其他因素的情况下,login方法的执行时间为5s左右;示例中对调用人脸识别接口的方法异步调用,所以最终的执行时间应该在4s左右.

测试方法 @RunWith(SpringRunner.class) @SpringBootTest(classes = ThreadPoolApplication.class) public class LoginServiceTest { @Autowired LoginService loginService; @Test public void login() { UserInfo userInfo = new UserInfo(); long start = Instant.now().toEpochMilli(); loginService.login(userInfo); long end = Instant.now().toEpochMilli(); System.out.println(end - start); } } 测试结果 2019-07-04 18:51:36.968 INFO 28924 --- [ main] c.c.threadpool.service.LoginServiceTest : Starting LoginServiceTest on DLC00R90RK7NBL with PID 28924 (started by jiaxin.chi in E:\threadpooltest) 2019-07-04 18:51:36.970 INFO 28924 --- [ main] c.c.threadpool.service.LoginServiceTest : No active profile set, falling back to default profiles: default 2019-07-04 18:51:39.622 INFO 28924 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor' 2019-07-04 18:51:40.143 INFO 28924 --- [ main] c.c.threadpool.service.LoginServiceTest : Started LoginServiceTest in 3.728 seconds (JVM running for 6.121) 2019-07-04 18:51:40.505 INFO 28924 --- [ task-1] c.cain.threadpool.service.LoginService : 异步执行 4018 2019-07-04 18:51:44.514 INFO 28924 --- [ Thread-2] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor'

从日志中发现调用了线程池,方法是异步执行的,并且运行时间为4018毫秒即4秒

源码地址:https://gitee.com/cjx940216/springboot-thread-pool-executor

后记:

①

@Configuration public class ThreadPoolConfig { @Bean public HelloService helloService() { ... } }

这个配置就等同于之前在xml里的配置

②这个地方使用的是

LoginService loginService = applicationContext.getBean(LoginService.class); loginService.getOnlineFaceVerificationResult(userInfo);

不能

this.getOnlineFaceVerificationResult(userInfo);

因为spring-aop是通过代理来给方法增强的(增强方法的注解如:@Transactional 和 @Asyn等),这个地方如果直接调用 t h i s . \bf\color{purple}{this.} this. g e t O n l i n e F a c e V e r i f i c a t i o n R e s u l t \bf\color{blue}{getOnlineFaceVerificationResult} getOnlineFaceVerificationResult ( u s e r I n f o ) \bf\color{black}{(userInfo)} (userInfo);使用的是真实的方法而不是代理的,所以不会执行异步操作,也就是绕过了spring-aop的增强,具体细节我会在下篇文章继续讲解.

补充: 使用@Bean(“beanName”)定义线程池 然后在@Async(“beanName”)中引用指定的线程池


【本文地址】


今日新闻

推荐新闻

CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3