SpringBoot中的异步操作与线程池

SpringBoot中的异步操作与线程池
强烈推介IDEA2021.1.3破解激活,IntelliJ IDEA 注册码,2021.1.3IDEA 激活码  

大家好,我是架构君,一个会写代码吟诗的架构师。今天说一说SpringBoot中的异步操作与线程池,希望能够帮助大家进步!!!

线程池类型

Java通过 java.util.concurrent.Executors 的静态方法提供五种线程池

  1. newCachedThreadPool 创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。
  2. newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。
  3. newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。
  4. newSingleThreadExecutor 创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
  5. newWorkStealingPool 这是java8新增的线程池类型。创建一个含有足够多线程的线程池,来维持相应的并行级别,它会通过工作窃取的方式,使得多核的CPU不会闲置,总会有活着的线程让CPU去运行。

五种线程池的底层实现

  • ThreadPoolExecutor 是CachedThreadPool、FixedThreadPool、SingleThreadExecutor、ScheduledThreadPool 这四种类型线程池的底层实现
  • ForkJoinPool (java7已有) 是WorkStealingPool线程池的底层实现

使用线程池的优点

  • 重用存在的线程,减少对象创建、消亡的开销,性能佳。
  • 可有效控制最大并发线程数,提高系统资源的使用率,同时避免过多资源竞争,避免堵塞。
  • 提供定时执行、定期执行、单线程、并发数控制等功能。

如何在SpringBoot中优雅的使用线程池

注册线程池

在config目录下创建 AsyncConfig 配置类,在配置类中定义线程池

package com.example.async_demo.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Lazy; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.SchedulingConfigurer; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.config.ScheduledTaskRegistrar; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @Configuration @EnableAsync @EnableScheduling public class AsyncConfig implements SchedulingConfigurer { //第一种线程池定义方式,可代替CachedThreadPool、FixedThreadPool、SingleThreadExecutor这三种 // Spring线程池 @Lazy //线程池懒加载 @Bean(name="threadPoolTaskExecutor",destroyMethod="shutdown") //name为线程池名称,destroyMethod="shutdown"在spring bean回收后释放资源 public ThreadPoolTaskExecutor threadPoolTaskExecutor() { //封装的是原生的ThreadPoolExecutor类型线程池 ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //核心线程数(获取硬件):线程池创建时候初始化的线程数 int corePoolSize = Runtime.getRuntime().availableProcessors(); System.out.println(corePoolSize); executor.setCorePoolSize(corePoolSize); //最大线程数+5:线程池最大的线程数,只有在缓冲队列满了之后才会申请超过核心线程数的线程 executor.setMaxPoolSize(corePoolSize+5); //缓冲队列500:用来缓冲执行任务的队列 executor.setQueueCapacity(500); //允许线程的空闲时间60秒:当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 executor.setKeepAliveSeconds(60); //线程池名的前缀:设置好了之后可以方便我们定位处理任务所在的线程池 executor.setThreadNamePrefix("MyAsync-"); executor.initialize(); return executor; } //第二种线程池定义方式,使用的是WorkStealingPool //java8 抢占式线程池 @Lazy @Bean(name="workStealingPool",destroyMethod="shutdown") public ExecutorService workStealingPool(){ ExecutorService executorService = Executors.newWorkStealingPool(); return executorService; } //第三种线程池定义方式,为周期任务线程池 //周期任务线程池 @Lazy @Bean(name="scheduledThreadPool",destroyMethod="shutdown") public ExecutorService scheduledThreadPool() { return Executors.newScheduledThreadPool(3); } @Override public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) { scheduledTaskRegistrar.setScheduler(scheduledThreadPool()); } } 

我在上述案例代码中定义了三种类型的线程池

  1. 第一种是ThreadPoolTaskExecutor线程池,他是Spring中的 org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor 线程池,
    底层是对 java.util.concurrent.ThreadPoolExecutor 的封装,综合了CachedThreadPool、FixedThreadPool、SingleThreadExecutor这三种线程池的优点;
  2. 第二种是java8新增的 workStealingPool 线程池。第一种和第二种使用时可以在配置类上使用@EnableAsync注解,这样就能优雅的使用@Async注解方法来实现线程run逻辑了;
  3. 第三种是ScheduledThreadPool线程池,不过在Spring中使用需要配置类实现SchedulingConfigurer接口,重写configureTasks方法。在配置类上使用
    @EnableScheduling注解,就可以优雅的使用@Scheduled注解方法来实现周期逻辑了

使用线程池

对第一种和第二种线程池在service中实现线程run的逻辑

package com.example.async_demo.service; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.stereotype.Service; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; @Service public class AsyncService { //使用名为threadPoolTaskExecutor的线程池,返回Future @Async("threadPoolTaskExecutor") public Future<Double> service1(){ double result = getRand(3000); return AsyncResult.forValue(result); } //使用名为threadPoolTaskExecutor的线程池,返回CompletableFuture @Async("threadPoolTaskExecutor") public CompletableFuture<Double> service2(){ double result = getRand(3000); return CompletableFuture.completedFuture(result); } //使用名为workStealingPool的线程池,返回CompletableFuture @Async("workStealingPool") public CompletableFuture<Double> service3(){ double result = getRand(3000); return CompletableFuture.completedFuture(result); } private double getRand(long sleep){ System.out.println(Thread.currentThread().getId()+"-start"); try { Thread.sleep(sleep); } catch (InterruptedException e) { e.printStackTrace(); } double result = Math.random();//方法返回的结果 return result; } } 

测试第一种和第二种线程池

@SpringBootTest class AsyncDemoApplicationTests { @Autowired private AsyncService asyncService; @Test void test1() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); Future<Double> result1 = asyncService.service1(); Future<Double> result2 = asyncService.service1(); Future<Double> result3 = asyncService.service1(); //让主线程等待子线程结束之后才能继续运行 while (!(result1.isDone()&&result2.isDone()&&result3.isDone())){ Thread.sleep(500); } long end = System.currentTimeMillis(); System.out.println(end-start+"ms"); System.out.println(result1.get()); System.out.println(result2.get()); System.out.println(result3.get()); } @Test void test2() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); CompletableFuture<Double> result1 = asyncService.service2(); CompletableFuture<Double> result2 = asyncService.service2(); CompletableFuture<Double> result3 = asyncService.service2(); //join() 的作用:让主线程等待子线程结束之后才能继续运行 CompletableFuture.allOf(result1,result2,result3).join(); long end = System.currentTimeMillis(); System.out.println(end-start+"ms"); System.out.println(result1.get()); System.out.println(result2.get()); System.out.println(result3.get()); } @Test void test3() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); CompletableFuture<Double> result1 = asyncService.service3(); CompletableFuture<Double> result2 = asyncService.service3(); CompletableFuture<Double> result3 = asyncService.service3(); //join() 的作用:让主线程等待子线程结束之后才能继续运行 CompletableFuture.allOf(result1,result2,result3).join(); long end = System.currentTimeMillis(); System.out.println(end-start+"ms"); System.out.println(result1.get()); System.out.println(result2.get()); System.out.println(result3.get()); } } 

test1测试结果

test1

test2测试结果

test1

test3测试结果

test1

通过测试发现Future返回类型不适合主线等待多个子线程全部完成的操作,
因为需要用到while循环去阻塞主线程,而CompletableFuture可以通过CompletableFuture.allOf(cf1,cf2,cf3).join()
去完成这个操作,所以推荐使用CompletableFuture作为返回类型

注意:@Async注解的方法不能在本类中被调用,只能在其他类中调用,如Controller类

对第三种线程池在service中实现线程的逻辑

package com.example.async_demo.service; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; @Service public class AsyncService { //cron表达式 每5秒执行一次 //@Scheduled(cron = "*/5 * * * * ?") @Scheduled(cron = "${cron.sec5}") //表达式写在application.yml文件中,则以这种方式取出。 public void service4(){ System.out.println("5s-"+Thread.currentThread().getId()+":"+System.currentTimeMillis()/1000); } //cron表达式 每3秒执行一次 @Scheduled(cron = "${cron.sec3}") public void service5(){ System.out.println("3s-"+Thread.currentThread().getId()+":"+System.currentTimeMillis()/1000); } } 

application.yml 文件

cron: sec5: '*/5 * * * * ?' sec3: '*/3 * * * * ?' 

周期任务测试结果(启动Application类)

test1

通过测试结果可发现两个周期任务使用了三个线程,
线程id分别是20、21、25。两个周期任务分别以3s和5s执行一次,
但不固定在某个线程中执行,而是哪个线程空闲则使用哪个线程

注意:若不为周期任务配置线程池,只使用@EnableScheduling和@Scheduled注解的话,
则所有周期任务共用一个子线程,若出现下一个周期开始上一个周期任务还没结束的情况,
则线程阻塞,直到前一个任务完成

CRON表达式

  • cron表达式是定义任务周期的一种表达式
  • 这里不多介绍,可以参考这篇 博客
本文来源huayang183,由架构君转载发布,观点不代表Java架构师必看的立场,转载请标明来源出处:https://javajgs.com/archives/18202

发表评论