zoukankan      html  css  js  c++  java

  • 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表达式是定义任务周期的一种表达式
    • 这里不多介绍，可以参考这篇博客
  • 相关阅读:
    ios代码大全
    MYSQL数据库之如何在已经建立好表之后重构数据表
    关于cookie在一个页面设置但是在另外一个页面获取不到的原因
    cookie的那点事儿
    关于a标签不能调用js方法的小细节，你注意到了么？
    关于mysql预处理技术的一点小心得
    关于delete使用limit的一些注意事项
    DP1 等分连续1-N个数的划分种数
    Spring 编程式事务和声明式事务管理
    java https client信任所有证书
  • 原文地址：https://www.cnblogs.com/xiaogblog/p/14011290.html
Copyright © 2011-2022 走看看