经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Spring Boot » 查看文章
SpringBoot线程池和Java线程池的用法和实现原理
来源：cnblogs　　作者：twilight0402　　时间：2023/4/12 11:16:50　　对本文有异议

使用默认的线程池

方式一：通过@Async注解调用

  1. public class AsyncTest {
  2.     @Async
  3.     public void async(String name) throws InterruptedException {
  4.         System.out.println("async" + name + " " + Thread.currentThread().getName());
  5.         Thread.sleep(1000);
  6.     }
  7. }

启动类上需要添加@EnableAsync注解，否则不会生效。

  1. @SpringBootApplication
  2. //@EnableAsync
  3. public class Test1Application {
  4.    public static void main(String[] args) throws InterruptedException {
  5.       ConfigurableApplicationContext run = SpringApplication.run(Test1Application.class, args);
  6.       AsyncTest bean = run.getBean(AsyncTest.class);
  7.       for(int index = 0; index <= 10; ++index){
  8.          bean.async(String.valueOf(index));
  9.       }
  10.    }
  11. }

方式二：直接注入 ThreadPoolTaskExecutor

此时可不加 @EnableAsync注解

  1. @SpringBootTest
  2. class Test1ApplicationTests {
  4.    @Resource
  5.    ThreadPoolTaskExecutor threadPoolTaskExecutor;
  7.    @Test
  8.    void contextLoads() {
  9.       Runnable runnable = () -> {
  10.          System.out.println(Thread.currentThread().getName());
  11.       };
  13.       for(int index = 0; index <= 10; ++index){
  14.          threadPoolTaskExecutor.submit(runnable);
  15.       }
  16.    }
  18. }

线程池默认配置信息

SpringBoot线程池的常见配置：

  1. spring:
  2.   task:
  3.     execution:
  4.       pool:
  5.         core-size: 8
  6.         max-size: 16                          # 默认是 Integer.MAX_VALUE
  7.         keep-alive: 60s                       # 当线程池中的线程数量大于 corePoolSize 时，如果某线程空闲时间超过keepAliveTime，线程将被终止
  8.         allow-core-thread-timeout: true       # 是否允许核心线程超时,默认true
  9.         queue-capacity: 100                   # 线程队列的大小,默认Integer.MAX_VALUE
  10.       shutdown:
  11.         await-termination: false              # 线程关闭等待
  12.       thread-name-prefix: task-               # 线程名称的前缀

SpringBoot 线程池的实现原理

TaskExecutionAutoConfiguration 类中定义了 ThreadPoolTaskExecutor,该类的内部实现也是基于java原生的 ThreadPoolExecutor类。initializeExecutor()方法在其父类中被调用，但是在父类中 RejectedExecutionHandler 被定义为了 private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); ，并通过initialize()方法将AbortPolicy传入initializeExecutor()中。

注意在TaskExecutionAutoConfiguration 类中，ThreadPoolTaskExecutor类的bean的名称为： applicationTaskExecutor 和 taskExecutor。

  1. // TaskExecutionAutoConfiguration#applicationTaskExecutor()
  2. @Lazy
  3. @Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,
  4.       AsyncAnnotationBeanPostProcessor.DEFAUL
  5.           T_TASK_EXECUTOR_BEAN_NAME })
  6. @ConditionalOnMissingBean(Executor.class)
  7. public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
  8.    return builder.build();
  9. }
  1. // ThreadPoolTaskExecutor#initializeExecutor()
  2. @Override
  3. protected ExecutorService initializeExecutor(
  4.       ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
  6.    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
  8.    ThreadPoolExecutor executor;
  9.    if (this.taskDecorator != null) {
  10.       executor = new ThreadPoolExecutor(
  11.             this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
  12.             queue, threadFactory, rejectedExecutionHandler) {
  13.          @Override
  14.          public void execute(Runnable command) {
  15.             Runnable decorated = taskDecorator.decorate(command);
  16.             if (decorated != command) {
  17.                decoratedTaskMap.put(decorated, command);
  18.             }
  19.             super.execute(decorated);
  20.          }
  21.       };
  22.    }
  23.    else {
  24.       executor = new ThreadPoolExecutor(
  25.             this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
  26.             queue, threadFactory, rejectedExecutionHandler);
  28.    }
  30.    if (this.allowCoreThreadTimeOut) {
  31.       executor.allowCoreThreadTimeOut(true);
  32.    }
  34.    this.threadPoolExecutor = executor;
  35.    return executor;
  36. }
  1. // ExecutorConfigurationSupport#initialize()
  2. public void initialize() {
  3.    if (logger.isInfoEnabled()) {
  4.       logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
  5.    }
  6.    if (!this.threadNamePrefixSet && this.beanName != null) {
  7.       setThreadNamePrefix(this.beanName + "-");
  8.    }
  9.    this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
  10. }

覆盖默认的线程池

覆盖默认的 taskExecutor对象，bean的返回类型可以是ThreadPoolTaskExecutor也可以是Executor。

  1. @Configuration
  2. public class ThreadPoolConfiguration {
  4.     @Bean("taskExecutor")
  5.     public ThreadPoolTaskExecutor taskExecutor() {
  6.         ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
  7.         //设置线程池参数信息
  8.         taskExecutor.setCorePoolSize(10);
  9.         taskExecutor.setMaxPoolSize(50);
  10.         taskExecutor.setQueueCapacity(200);
  11.         taskExecutor.setKeepAliveSeconds(60);
  12.         taskExecutor.setThreadNamePrefix("myExecutor--");
  13.         taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  14.         taskExecutor.setAwaitTerminationSeconds(60);
  15.         //修改拒绝策略为使用当前线程执行
  16.         taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  17.         //初始化线程池
  18.         taskExecutor.initialize();
  19.         return taskExecutor;
  20.     }
  21. }

管理多个线程池

如果出现了多个线程池，例如再定义一个线程池 taskExecutor2，则直接执行会报错。此时需要指定bean的名称即可。

  1. @Bean("taskExecutor2")
  2. public ThreadPoolTaskExecutor taskExecutor2() {
  3.     ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
  4.     //设置线程池参数信息
  5.     taskExecutor.setCorePoolSize(10);
  6.     taskExecutor.setMaxPoolSize(50);
  7.     taskExecutor.setQueueCapacity(200);
  8.     taskExecutor.setKeepAliveSeconds(60);
  9.     taskExecutor.setThreadNamePrefix("myExecutor2--");
  10.     taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  11.     taskExecutor.setAwaitTerminationSeconds(60);
  12.     //修改拒绝策略为使用当前线程执行
  13.     taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  14.     //初始化线程池
  15.     taskExecutor.initialize();
  16.     return taskExecutor;
  17. }

引用线程池时，需要将变量名更改为bean的名称，这样会按照名称查找。

  1. @Resource
  2. ThreadPoolTaskExecutor taskExecutor2;

对于使用@Async注解的多线程则在注解中指定bean的名字即可。

  1.     @Async("taskExecutor2")
  2.     public void async(String name) throws InterruptedException {
  3.         System.out.println("async" + name + " " + Thread.currentThread().getName());
  4.         Thread.sleep(1000);
  5.     }

线程池的四种拒绝策略

JAVA常用的四种线程池

ThreadPoolExecutor 类的构造函数如下：

  1. public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
  2.                           BlockingQueue<Runnable> workQueue) {
  3.     this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
  4.          Executors.defaultThreadFactory(), defaultHandler);
  5. }

newCachedThreadPool

不限制最大线程数（maximumPoolSize=Integer.MAX_VALUE）,如果有空闲的线程超过需要，则回收，否则重用已有的线程。

  1. new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  2.                               60L, TimeUnit.SECONDS,
  3.                               new SynchronousQueue<Runnable>());

newFixedThreadPool

定长线程池，超出线程数的任务会在队列中等待。

  1. return new ThreadPoolExecutor(nThreads, nThreads,
  2.                               0L, TimeUnit.MILLISECONDS,
  3.                               new LinkedBlockingQueue<Runnable>());

newScheduledThreadPool

类似于newCachedThreadPool，线程数无上限，但是可以指定corePoolSize。可实现延迟执行、周期执行。

  1. public ScheduledThreadPoolExecutor(int corePoolSize) {
  2.     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
  3.           new DelayedWorkQueue());
  4. }

周期执行：

  1. ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  2. scheduledThreadPool.scheduleAtFixedRate(()->{
  3.    System.out.println("rate");
  4. }, 1, 1, TimeUnit.SECONDS);

延时执行：

  1. scheduledThreadPool.schedule(()->{
  2.    System.out.println("delay 3 seconds");
  3. }, 3, TimeUnit.SECONDS);

newSingleThreadExecutor

单线程线程池，可以实现线程的顺序执行。

  1. public static ExecutorService newSingleThreadExecutor() {
  2.     return new FinalizableDelegatedExecutorService
  3.         (new ThreadPoolExecutor(1, 1,
  4.                                 0L, TimeUnit.MILLISECONDS,
  5.                                 new LinkedBlockingQueue<Runnable>()));
  6. }

Java 线程池中的四种拒绝策略

  • CallerRunsPolicy：线程池让调用者去执行。

  • AbortPolicy：如果线程池拒绝了任务，直接报错。

  • DiscardPolicy：如果线程池拒绝了任务，直接丢弃。

  • DiscardOldestPolicy：如果线程池拒绝了任务，直接将线程池中最旧的，未运行的任务丢弃，将新任务入队。

CallerRunsPolicy

直接在主线程中执行了run方法。

  1. public static class CallerRunsPolicy implements RejectedExecutionHandler {
  2.  
  3.     public CallerRunsPolicy() { }
  4.  
  5.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.         if (!e.isShutdown()) {
  7.             r.run();
  8.         }
  9.     }
  10. }

效果类似于：

  1. Runnable thread = ()->{
  2.    System.out.println(Thread.currentThread().getName());
  3.    try {
  4.       Thread.sleep(0);
  5.    } catch (InterruptedException e) {
  6.       throw new RuntimeException(e);
  7.    }
  8. };
  10. thread.run();

AbortPolicy

直接抛出RejectedExecutionException异常，并指示任务的信息，线程池的信息。、

  1. public static class AbortPolicy implements RejectedExecutionHandler {
  2.  
  3.     public AbortPolicy() { }
  4.  
  5.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.         throw new RejectedExecutionException("Task " + r.toString() +
  7.                                              " rejected from " +
  8.                                              e.toString());
  9.     }
  10. }

DiscardPolicy

什么也不做。

  1. public static class DiscardPolicy implements RejectedExecutionHandler {
  2.  
  3.     public DiscardPolicy() { }
  4.  
  5.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.     }
  7. }

DiscardOldestPolicy

  • e.getQueue().poll() : 取出队列最旧的任务。

  • e.execute(r) : 当前任务入队。

  1. public static class DiscardOldestPolicy implements RejectedExecutionHandler {
  2.  
  3.     public DiscardOldestPolicy() { }
  4.  
  5.     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.         if (!e.isShutdown()) {
  7.             e.getQueue().poll();
  8.             e.execute(r);
  9.         }
  10.     }
  11. }

Java 线程复用的原理

java的线程池中保存的是 java.util.concurrent.ThreadPoolExecutor.Worker 对象，该对象在 被维护在private final HashSet<Worker> workers = new HashSet<Worker>();。workQueue是保存待执行的任务的队列，线程池中加入新的任务时，会将任务加入到workQueue队列中。

  1. private final class Worker
  2.     extends AbstractQueuedSynchronizer
  3.     implements Runnable
  4. {
  5.     /**
  6.      * This class will never be serialized, but we provide a
  7.      * serialVersionUID to suppress a javac warning.
  8.      */
  9.     private static final long serialVersionUID = 6138294804551838833L;
  11.     /** Thread this worker is running in.  Null if factory fails. */
  12.     final Thread thread;
  13.     /** Initial task to run.  Possibly null. */
  14.     Runnable firstTask;
  15.     /** Per-thread task counter */
  16.     volatile long completedTasks;
  18.     /**
  19.      * Creates with given first task and thread from ThreadFactory.
  20.      * @param firstTask the first task (null if none)
  21.      */
  22.     Worker(Runnable firstTask) {
  23.         setState(-1); // inhibit interrupts until runWorker
  24.         this.firstTask = firstTask;
  25.         this.thread = getThreadFactory().newThread(this);
  26.     }
  28.     /** Delegates main run loop to outer runWorker  */
  29.     public void run() {
  30.         runWorker(this);
  31.     }
  33.     // Lock methods
  34.     //
  35.     // The value 0 represents the unlocked state.
  36.     // The value 1 represents the locked state.
  38.     protected boolean isHeldExclusively() {
  39.         return getState() != 0;
  40.     }
  42.     protected boolean tryAcquire(int unused) {
  43.         if (compareAndSetState(0, 1)) {
  44.             setExclusiveOwnerThread(Thread.currentThread());
  45.             return true;
  46.         }
  47.         return false;
  48.     }
  50.     protected boolean tryRelease(int unused) {
  51.         setExclusiveOwnerThread(null);
  52.         setState(0);
  53.         return true;
  54.     }
  56.     public void lock()        { acquire(1); }
  57.     public boolean tryLock()  { return tryAcquire(1); }
  58.     public void unlock()      { release(1); }
  59.     public boolean isLocked() { return isHeldExclusively(); }
  61.     void interruptIfStarted() {
  62.         Thread t;
  63.         if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
  64.             try {
  65.                 t.interrupt();
  66.             } catch (SecurityException ignore) {
  67.             }
  68.         }
  69.     }
  70. }

work对象的执行依赖于 runWorker()，与我们平时写的线程不同，该线程处在一个循环中，并不断地从队列中获取新的任务执行。因此线程池中的线程才可以复用，而不是像我们平常使用的线程一样执行完毕就结束。

  1. final void runWorker(Worker w) {
  2.     Thread wt = Thread.currentThread();
  3.     Runnable task = w.firstTask;
  4.     w.firstTask = null;
  5.     w.unlock(); // allow interrupts
  6.     boolean completedAbruptly = true;
  7.     try {
  8.         while (task != null || (task = getTask()) != null) {
  9.             w.lock();
  10.             // If pool is stopping, ensure thread is interrupted;
  11.             // if not, ensure thread is not interrupted.  This
  12.             // requires a recheck in second case to deal with
  13.             // shutdownNow race while clearing interrupt
  14.             if ((runStateAtLeast(ctl.get(), STOP) ||
  15.                  (Thread.interrupted() &&
  16.                   runStateAtLeast(ctl.get(), STOP))) &&
  17.                 !wt.isInterrupted())
  18.                 wt.interrupt();
  19.             try {
  20.                 beforeExecute(wt, task);
  21.                 Throwable thrown = null;
  22.                 try {
  23.                     task.run();
  24.                 } catch (RuntimeException x) {
  25.                     thrown = x; throw x;
  26.                 } catch (Error x) {
  27.                     thrown = x; throw x;
  28.                 } catch (Throwable x) {
  29.                     thrown = x; throw new Error(x);
  30.                 } finally {
  31.                     afterExecute(task, thrown);
  32.                 }
  33.             } finally {
  34.                 task = null;
  35.                 w.completedTasks++;
  36.                 w.unlock();
  37.             }
  38.         }
  39.         completedAbruptly = false;
  40.     } finally {
  41.         processWorkerExit(w, completedAbruptly);
  42.     }
  43. }

原文链接：https://www.cnblogs.com/twilight0402/p/17305328.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Spring Boot热门文章

  • 史上最详 Thymeleaf 使用教程
  • Spring Boot如何通过java -jar启动
  • Spring Boot 2.x（六）：优雅的统一返回值
  • Springboot使用Logback实现日志配置与异常记录
  • 使用SpringBoot+EasyExcel+Vue实现excel表格的导入和导出详解
  • spring boot上传文件超出大小异常无法捕获问题
  • paging组件的用法和意义
  • springboot2.0集成rabbitmq的示例代码
  • Spring MVC学习step1——框架熟悉
  • spring boot ${}占位符不起作用的解决方案

    • Spring Boot推荐文章

  • 3分钟带你搞定Spring Boot中Schedule
  • 不单独部署注册中心，又要具备注册中心的功能，我能上天！
  • 从零开始学Spring Boot系列-集成Spring Security实现用户认证与授权
  • 不单独部署注册中心，又要具备注册中心的功能，咋不让我上天？
  • EasyExcel 无法读取图片？用poi写了一个工具类
  • 【进阶篇】使用 Stream 流对比两个集合的常用操作分享
  • 关于DDD和COLA的一些总结和思考
  • SnakeYaml反序列化分析
  • RestTemplate进行https请求时适配信任证书
  • Kryo反序列化链分析

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号