SpringBoot 自定义线程池

今天药忘吃喽~ 2021-12-14 20:15 371阅读 0赞

1.我们都知道spring只是为我们简单的处理线程池，每次用到线程总会new 一个新的线程，效率不高，所以我们需要自定义一个线程池。

2.自定义线程池有两种方法，第一种自定义线程池然后使用自己的自定义的，第二种重写spring默认的线程池，然后使用自己重写过的线程池

一：自定义线程池

1.1 修改application.yml

#线程池配置参数
    task:
      pool:
        corePoolSize: 5 #设置核心线程数
        maxPoolSize: 20  #设置最大线程数
        keepAliveSeconds: 300 #设置线程活跃时间（秒）
        queueCapacity: 50 #设置队列容量

1.2 线程池配置属性类TaskThreadPoolConfig .java

import org.springframework.boot.context.properties.ConfigurationProperties;
    
    /**
     * 线程池配置属性类
     */
    @ConfigurationProperties(prefix = "task.pool")
    public class TaskThreadPoolConfig {
        private int corePoolSize;
    
        private int maxPoolSize;
    
        private int keepAliveSeconds;
    
        private int queueCapacity;
        ...getter and setter methods...
    }

1.3  注意启动类上一定要开启线程异步支持

@EnableAsync
    @EnableConfigurationProperties({TaskThreadPoolConfig.class} ) // 开启配置属性支持

1.4 创建线程池 TaskExecutePool .java

/**
     * 创建线程池配置类
     */
    @Configuration
    public class TaskExecutePool {
    
        @Autowired
        private TaskThreadPoolConfig config;
    
        /**
         * 1.这种形式的线程池配置是需要在使用的方法上面@Async("taskExecutor"),
         * 2.如果在使用的方法上面不加该注解那么spring就会使用默认的线程池
         * 3.所以如果加@Async注解但是不指定使用的线程池，又想自己定义线程池那么就可以重写spring默认的线程池
         * 4.所以第二个方法就是重写默认线程池
         * 注意：完全可以把线程池的参数写到配置文件中
         */
    
        @Bean
        public Executor taskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //核心线程池大小
            executor.setCorePoolSize(config.getCorePoolSize());
            //最大线程数
            executor.setMaxPoolSize(config.getMaxPoolSize());
            //队列容量
            executor.setQueueCapacity(config.getQueueCapacity());
            //活跃时间
            executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
            //线程名字前缀
            executor.setThreadNamePrefix("TaskExecutePool-");
    
            // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
            // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
             // 等待所有任务结束后再关闭线程池
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.initialize();
            return executor;
        }
    }

1.5 测试方法

import io.swagger.annotations.Api;
    import io.swagger.annotations.ApiOperation;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.ResponseBody;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @author qijx
     */
    @Api(description = "测试控制类11111")
    @RestController
    @RequestMapping("/threadPoolController1")
    public class ThreadPoolController1 {
    
            @Autowired
            private ThreadPoolService1 threadPoolService;
    
    
            @ApiOperation(value = "测试方法")
            @ResponseBody
            @RequestMapping(value = "/test",method = RequestMethod.GET)
            public String threadPoolTest() {
                threadPoolService.executeAsync();
                return "hello word!";
            }
    
    
    }

1.6 service测试方法

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Service;
    
    /**
     * @author qijx
     * @date 2019-07-03 17:40
     */
    
    @Service
    public class ThreadPoolService1 {
        private static final Logger logger = LoggerFactory.getLogger(ThreadPoolService1.class);
    
        @Async("taskExecutor") //指定使用那个线程池配置，不然会使用spring默认的线程池
        public void executeAsync() {
            logger.info("start executeAsync");
            try {
                System.out.println("当前运行的线程名称：" + Thread.currentThread().getName());
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            logger.info("end executeAsync");
        }
    
    
    
    }

二：配置默认的线程池

2.1 第一种方式的那个线程池使用时候总要加注解`@Async("`taskExecutor`")`，而这种方式是重写spring默认线程池的方式，使用的时候只需要加`@Async`注解就可以，不用去声明线程池类。

2.2 这个和上面的TaskThreadPoolConfig类相同，这里不重复

2.3 NativeAsyncTaskExecutePool.java 装配线程池

**
     * 原生(Spring)异步任务线程池装配类，实现AsyncConfigurer重写他的两个方法，这样在使用默认的
     *  线程池的时候就会使用自己重写的
     */
    @Slf4j
    @Configuration
    public class NativeAsyncTaskExecutePool implements AsyncConfigurer{
    
    
        //注入配置类
        @Autowired
        TaskThreadPoolConfig config;
    
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            //核心线程池大小
            executor.setCorePoolSize(config.getCorePoolSize());
            //最大线程数
            executor.setMaxPoolSize(config.getMaxPoolSize());
            //队列容量
            executor.setQueueCapacity(config.getQueueCapacity());
            //活跃时间
            executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
            //线程名字前缀
            executor.setThreadNamePrefix("NativeAsyncTaskExecutePool-");
    
            // setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
            // CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            // 等待所有任务结束后再关闭线程池
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.initialize();
            return executor;
        }
    
    
        /**
         *  异步任务中异常处理
         * @return
         */
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new AsyncUncaughtExceptionHandler() {
    
                @Override
                public void handleUncaughtException(Throwable arg0, Method arg1, Object... arg2) {
                    log.error("=========================="+arg0.getMessage()+"=======================", arg0);
                    log.error("exception method:"+arg1.getName());
                }
            };
        }
    }

2.4 测试controller

/**
     * @author qijx
     */
    @Api(description = "测试控制类22222")
    @RestController
    @RequestMapping("/threadPoolController2")
    public class ThreadPoolController2 {
            @Autowired
            private ThreadPoolService2 threadPoolService;
    
            @ApiOperation(value = "测试方法")
            @ResponseBody
            @RequestMapping(value = "/test",method = RequestMethod.GET)
            public String threadPoolTest() {
                threadPoolService.executeAsync();
                return "hello word!";
            }
    }

2.5 测试service方法

/**
     * @author qijx
     */
    
    @Service
    public class ThreadPoolService2 {
        private static final Logger logger = LoggerFactory.getLogger(ThreadPoolService2.class);
    
        /**
         * @Async该注解不需要在指定任何bean
         */
        @Async
        public void executeAsync() {
            logger.info("start executeAsync");
            try {
                System.out.println("当前运行的线程名称：" + Thread.currentThread().getName());
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            logger.info("end executeAsync");
        }
    
    
    }