Spring中的几种线程池使用总结

爱被打了一巴掌 2023-09-30 19:14 39阅读 0赞

第一种：常用的Executor线程池

package com.glodon.litddemo.config.threadPool;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.*;
    
    /**
     * @className: ThreadExecutorPool.java
     * @description: TODO 线程池使用
     * @author: litd
     * @date: 2021/7/20 17:00
     **/
    @Configuration
    @EnableAsync
    public class ThreadExecutorPool {
    
        private Logger log = LoggerFactory.getLogger(this.getClass());
        @Bean("asyncServiceExecutor")
        public ThreadPoolTaskExecutor asyncServiceExecutor() {
            log.info("start asyncServiceExecutorCulateChild");
            int coreCpuNum = (Runtime.getRuntime().availableProcessors());
            int maxPoolSize = (Runtime.getRuntime().availableProcessors())*2;
            //使用自定义的VisiableThreadPoolTaskExecutor
            ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
            //配置核心线程数
            executor.setCorePoolSize(coreCpuNum);
            //配置最大线程数
            executor.setMaxPoolSize(maxPoolSize);
            //配置队列大小
            executor.setQueueCapacity(99999999);
            //配置线程池中的线程的名称前缀
            executor.setThreadNamePrefix("j-pool-");
            executor.setKeepAliveSeconds(180);
    
            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.setTaskDecorator(runnable -> runnable);
            //执行初始化
            executor.initialize();
            return executor;
        }
    
        /**
         * @description: 只负责执行，不带返回值
         * @author: litd
         * @date: 2021/7/22 16:07
         **/
        @Bean("Executor-thread-void")
        public Executor ExecutorVoid() {
            int coreCpuNum = (Runtime.getRuntime().availableProcessors());
            int maxPoolSize = (Runtime.getRuntime().availableProcessors())*2;
            ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
            //配置核心线程数
            executor.setCorePoolSize(coreCpuNum);
            //配置最大线程数
            executor.setMaxPoolSize(maxPoolSize);
            //配置队列大小
            executor.setQueueCapacity(10000);
            //配置线程池中的线程的名称前缀
            executor.setThreadNamePrefix("Executor-thread-");
            executor.setKeepAliveSeconds(60);
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.setTaskDecorator(runnable -> runnable);
            //执行初始化
            executor.initialize();
            return executor;
        }
    
        /**
         * @description: 单例线程池，主要用例打印频繁的日志和独立执行不重要的任务。不阻塞主程序的性能
         * @author: litd
         * @date: 2021/7/23 9:48
         **/
        @Bean("singleThreadExecutor")
        public ExecutorService singleThreadExecutor(){
            ExecutorService singleThreadExecutor= new ThreadPoolExecutor(1,1,15, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(200000),
                    new ThreadPoolExecutor.CallerRunsPolicy());
            return singleThreadExecutor;
        }

第二种：装饰器线程池（采用goole提供的Guava工具包实现）

package com.glodon.litddemo.config.threadPool;
    
    import com.google.common.util.concurrent.ListeningExecutorService;
    import com.google.common.util.concurrent.MoreExecutors;
    import org.apache.commons.lang3.concurrent.BasicThreadFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    
    
    import java.util.concurrent.*;
    
    /**
     * @author
     * @version 1.0
     * @description
     * @updateRemark
     * @updateUser
     * @createDate 2021/7/20 18:03
     * @updateDate 2021/7/20 18:03
     **/
    @Configuration
    @EnableAsync
    public class ThreadPoolGuava {
    
        private Logger log = LoggerFactory.getLogger(this.getClass());
        /**
         * 线程工厂名称
         */
        private static final ThreadFactory FACTORY = new BasicThreadFactory.Builder().namingPattern("guava-pool-")
                .daemon(false).build();
        @Bean("ExecutorGuava")
        public ListeningExecutorService  asyncServiceExecutor() {
            log.info("start asyncServiceExecutorCulateChild");
            //使用自定义的VisiableThreadPoolTaskExecutor
            ThreadPoolExecutor  executor = new ThreadPoolExecutor(10,20,120, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(200),
                    new ThreadPoolExecutor.CallerRunsPolicy());
            executor.setThreadFactory(FACTORY);
    
            ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executor);
            return listeningExecutorService;
        }
    }

以上两种的使用方式和区别

public class ExampleController1 {
        private Logger log = LoggerFactory.getLogger(this.getClass());
        @Autowired
        @Qualifier("asyncServiceExecutor")
        private ThreadPoolTaskExecutor executor;
        @Autowired
        @Qualifier("ExecutorGuava")
        private ListeningExecutorService listeningExecutorService;
    
        public String threadTest(){
    
            try {
                Future<Integer> submit = executor.submit(new Callable<Integer>() {
                    @Override
                    public Integer call() throws Exception {
                        /**
                         * 业务实现
                         */
    
                        return 66;
                    }
                });
                Integer integer = submit.get();
                log.info(integer.toString());
            }catch (Exception e){
                e.printStackTrace();
            }
    
            try {
                //threadService.bbs(990)  代表业务实现，需要异步执行的业务
                ListenableFuture<Integer> submit = listeningExecutorService.submit(() -> threadService.bbs(990));
                //异步监听任务是否执行成功，如果不需要监听，可以不写
                Futures.addCallback(submit, new FutureCallback<Integer>() {
                    @Override
                    public void onSuccess(@Nullable Integer integer) {
                       log.info("装饰线程执行成功了");
                    }
    
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("执行失败了，重新调用");
                        //threadService.bbs(990);
                    }
                },listeningExecutorService);
                Integer integer = submit.get();
                log.info("获得最终结果为："+integer);
            }catch (Exception e){
                e.printStackTrace();
            }
    }

区别：

第一种，属于Spring自带，传统的jdk封装实现，对线程的异步执行后的结果，需要自己实现监听或者阻塞，相对麻烦。

第二种，装饰线程池，其原理也是实现的第一种，只不过他的好处是更便捷，尤其对线程的监听，随时随地可以监听此线程任务的内部状态，做出相应的处理。在使用上，更为推荐这种方式。

第三种：ForkJoinPool线程池。

ForkJoinPool 不是为了替代 ExecutorService，而是它的补充，在某些应用场景下性能比 ExecutorService 更好。

ForkJoinPool 主要用于实现“分而治之”的算法，特别是分治之后递归调用的函数，例如 quick sort 等。

ForkJoinPool 最适合的是计算密集型的任务，如果存在 I/O，线程间同步，sleep() 等会造成线程长时间阻塞的情况时。

ForkJoinPool 的每个工作线程都维护着一个工作队列（WorkQueue），这是一个双端队列（Deque），里面存放的对象是任务（ForkJoinTask）。

每个工作线程在运行中产生新的任务（通常是因为调用了 fork()）时，会放入工作队列的队尾，并且工作线程在处理自己的工作队列时，使用的是 LIFO 方式来执行。

每个工作线程在处理自己的工作队列同时，会尝试窃取一个任务（或是来自于刚刚提交到 pool 的任务，或是来自于其他工作线程的工作队列），窃取的任务位于其他线程的工作队列的队首，也就是说工作线程在窃取其他工作线程的任务时，使用的是 FIFO 方式。

在遇到 join() 时，如果需要 join 的任务尚未完成，则会先处理其他任务，并等待其完成。

从以上分析可以看出，再CPU物理线程数满足的情况下，它的性能是最好的。

事例：

/**
     * @className: ForkJoinPoolFiber.java
     * @description: 并行流线程池
     * @author: litd
     * @date: 2021/8/4 15:15
     **/
    public class ForkJoinPoolFiber {
    
        private ForkJoinPoolFiber(){}
    
        //这是默认的方式。当然你也可以自己创建，如：ForkJoinPool pool = new ForkJoinPool(16); 16代表初始化的线程数
        private static ForkJoinPool pool =ForkJoinPool.commonPool();
        public static ForkJoinPool getForkJoinPool(){
            if(pool==null){
                pool =ForkJoinPool.commonPool();
            }
            return pool;
        }
    
    
        /**
         * @description: TODO 使用demo
         * @author: litd
         * @date: 2021/8/4 15:20
         **/
        static public void forkjoinTest() throws Exception{
            long time=System.currentTimeMillis();
            ForkJoinPool pool =ForkJoinPoolFiber.getForkJoinPool();
            List<ForkJoinTask<Integer>> fiberQueue = new ArrayList<>();
            AtomicInteger sum= new AtomicInteger();
            for (int j = 0; j < 200000; j++) {
                int finalI = j;
                ForkJoinTask<Integer> submit = pool.submit(() -> {
                    sum.addAndGet(finalI);
                    return sum.get();
                });
                fiberQueue.add(submit);
            }
            fiberQueue.forEach(item ->{
                try {
                    item.get();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            long time2=System.currentTimeMillis();
        }
    
    }

第四种：CompletableFuture 属于线程模板执行器。

1，Java8新增了CompletableFuture 提供对异步计算的支持，可以通过回调的方式处理计算结果

2，它也是一个线程框架，异步执行器。它默认使用的线程池为ForkJoinPool ,可以默认指定ExecutorService接口下的线程池

CompletableFuture用来解决什么问题？

1，解决future模式的缺点

1）为了解决Future虽然可以实现异步获取线程的执行结果，但是future没有提供通知机制，调用方无法得知future什么时候执行完的问题。

2）要么使用阻塞， 在future.get()的地方等待future返回结果，这时会变成同步操作。如果使用isDone()方法进行循环判断，就会消耗cpu资源

2，CompletableFuture能够将回调放到与任务不同的线程中执行，也能将回调作为继续执行的同步函数。

3，他避免了传统回调的最大问题，就是能够将控制流分离到不同的事件处理器中

事例：

/**
     * @className: CompletableFutureAsync.java
     * @description: TODO 类描述
     * @author: litd
     * @date: 2021/8/10 16:26
     **/
    public class CompletableFutureAsync {
    
        private static final Logger logger = LogManager.getLogger(CompletableFutureAsync.class);
    
        public static void main(String[] args) throws Exception {
            completableTest();
        }
    
        /**
        * @Description  异步回调
        * @Author  litd
        * @Date   2021/8/10 16:33
        * @Param
        * @Return
        * @Exception
        */
        public static void completableTest() throws Exception{
            ForkJoinPool pool=new ForkJoinPool(16);
            CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return 3.9;
            },pool);
    
    
            CompletableFuture cf2=cf.thenApply(result ->{
                //将上一个任务的返回结果作为参数传入，其实是创建了一个新的CompletableFuture执行
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.error("cf2得到上一个任务的结果={}",result);
                String str = result+"cf1结果经过cf2进行了加工";
                return str;
            }).thenAccept(result->{
                logger.error("得到上一个任务的返回值={},但我的就是貔貅只进不出！",result);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).thenRun(()->{
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.error("我是被遗忘的那个，不给狗粮，我也不拉狗屎");
            });
            logger.error("我是不是比cf2还执行的早");
            logger.error("cf2执行完成了，得到的结果={}",cf2.get());
    
            //当某个任务异常时进行处理
            CompletableFuture<Double> cf3=cf.exceptionally(ex ->{
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                logger.error("cf婚内出轨了，被我处理了，哈哈");
                ex.printStackTrace();
                return -2.3;
            });
        }
    
    
    }

大家可通过简单的demo验证进行掌握

以上集中线程再使用上相对有区别，从性能的角度来说，基本都差不多。但是如果是是IO密集型的处理，建议采用ForkJoin线程池。计算密集型可以采用其他三种。性能差距不大，都是和CPU进行的交互。