springboot-线程池

疫情大数据平台可能会出现短时大量访问的情况，就需要线程池进行处理，本篇主要描述手动实现线程池

实现：

首先启动类需要添加这个注解@EnableAsync

@SpringBootApplication@EnableAsyncpublicclassDemoApplication{publicstaticvoidmain(String[] args){SpringApplication.run(DemoApplication.class, args);}@BeanpublicRestTemplaterestTemplate(){returnnewRestTemplate();}}

然后，新增threadPoolTaskExecutor类，用于创建线程池

@Configuration@EnableAsyncpublicclassThreadExecutorConfig{privatefinalLogger logger=LoggerFactory.getLogger(this.getClass());privateint corePoolSize=20;privateint maxPoolSize=20;privateint queueCapacity=20;@BeanpublicExecutorthreadPoolTaskExecutor(){System.out.println("创建线程池");ThreadPoolTaskExecutor executor=newThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setThreadNamePrefix("threadPoolTaskExecutor-》》》");

        executor.setRejectedExecutionHandler(newThreadPoolExecutor.AbortPolicy());
        executor.initialize();return executor;}

使用线程池

任务类

@Componentpublicclass test{@Async("threadPoolTaskExecutor")publicvoidtestThread()throwsInterruptedException{System.out.println("内部1"+Thread.currentThread().getName());System.out.println("2s执行方法睡眠前");Thread.sleep(2000);System.out.println("2s执行方法睡眠后");}@Async("threadPoolTaskExecutor")publicvoidtestThread1()throwsInterruptedException{System.out.println("内部2"+Thread.currentThread().getName());System.out.println("10s执行方法睡眠前");Thread.sleep(10000);System.out.println("10s执行方法睡眠后");}}

补充线程池代码

新增一个配置类

publicclassMyThreadPoolTaskExecutorextendsThreadPoolTaskExecutor{Logger logger=LoggerFactory.getLogger(MyThreadPoolTaskExecutor.class);@Overridepublicvoidexecute(Runnable task){logThreadPoolStatus();super.execute(task);}@Overridepublicvoidexecute(Runnable task,long startTimeout){logThreadPoolStatus();super.execute(task, startTimeout);}@OverridepublicFuture<?>submit(Runnable task){logThreadPoolStatus();returnsuper.submit(task);}@Overridepublic<T>Future<T>submit(Callable<T> task){logThreadPoolStatus();returnsuper.submit(task);}@OverridepublicListenableFuture<?>submitListenable(Runnable task){logThreadPoolStatus();returnsuper.submitListenable(task);}@Overridepublic<T>ListenableFuture<T>submitListenable(Callable<T> task){logThreadPoolStatus();returnsuper.submitListenable(task);}privatevoidlogThreadPoolStatus(){
        logger.info("核心线程数:{}, 最大线程数:{}, 当前线程数: {}, 活跃的线程数: {}",getCorePoolSize(),getMaxPoolSize(),getPoolSize(),getActiveCount());}}

补充bean

@BeanpublicExecutormyThreadPool(){int corePoolSize=5;int maxPoolSize=5;int queueCapacity=2000;long keepAliveTime=30;String threadNamePrefix="myThreadPool-->";RejectedExecutionHandler rejectedExecutionHandler=newRejectedExecutionHandler(){@OverridepublicvoidrejectedExecution(Runnable r,ThreadPoolExecutor executor){thrownewRejectedExecutionException("自定义的RejectedExecutionHandler");}};ThreadFactory threadFactory=newThreadFactory(){privateint i=1;@OverridepublicThreadnewThread(Runnable r){Thread thread=newThread(r);
                thread.setName(threadNamePrefix+ i);
                i++;return thread;}};ThreadPoolExecutor threadPoolExecutor=newThreadPoolExecutor(corePoolSize, maxPoolSize,
                keepAliveTime,TimeUnit.SECONDS,newLinkedBlockingQueue<>(queueCapacity),
                threadFactory, rejectedExecutionHandler);return threadPoolExecutor;}@BeanpublicExecutormyThreadPoolTaskExecutor(){ThreadPoolTaskExecutor threadPoolTaskExecutor=newMyThreadPoolTaskExecutor();

        threadPoolTaskExecutor.setCorePoolSize(5);
        
        threadPoolTaskExecutor.setMaxPoolSize(5);
      
        threadPoolTaskExecutor.setQueueCapacity(2000);
 threadPoolTaskExecutor.setThreadNamePrefix("myThreadPoolTaskExecutor-->");

        threadPoolTaskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.AbortPolicy());
        
        threadPoolTaskExecutor.initialize();return threadPoolTaskExecutor;}