SpringBoot中的异步操作与线程池

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表达式是定义任务周期的一种表达式
  • 这里不多介绍，可以参考这篇博客
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/xiaogblog/p/14011290.html