Comment utiliser le pool de threads SpringBoot et le pool de threads Java

Principes d'utilisation et de mise en œuvre du pool de threads SpringBoot et du pool de threads Java

Utilisez le pool de threads par défaut

Méthode 1 : Appel via l'annotation @Async

public class AsyncTest {
    @Async
    public void async(String name) throws InterruptedException {
        System.out.println("async" + name + " " + Thread.currentThread().getName());
        Thread.sleep(1000);
    }
}

Vous devez ajouter @EnableAsyncà la classe de démarrage >Annotation, sinon elle ne prendra pas effet. @EnableAsync注解，否则不会生效。

@SpringBootApplication
//@EnableAsync
public class Test1Application {
   public static void main(String[] args) throws InterruptedException {
      ConfigurableApplicationContext run = SpringApplication.run(Test1Application.class, args);
      AsyncTest bean = run.getBean(AsyncTest.class);
      for(int index = 0; index <= 10; ++index){
         bean.async(String.valueOf(index));
      }
   }
}

方式二：直接注入 ThreadPoolTaskExecutor

此时可不加 @EnableAsync注解

@SpringBootTest
class Test1ApplicationTests {

   @Resource
   ThreadPoolTaskExecutor threadPoolTaskExecutor;

   @Test
   void contextLoads() {
      Runnable runnable = () -> {
         System.out.println(Thread.currentThread().getName());
      };

      for(int index = 0; index <= 10; ++index){
         threadPoolTaskExecutor.submit(runnable);
      }
   }

}

线程池默认配置信息

SpringBoot线程池的常见配置：

spring:
  task:
    execution:
      pool:
        core-size: 8
        max-size: 16                          # 默认是 Integer.MAX_VALUE
        keep-alive: 60s                       # 当线程池中的线程数量大于 corePoolSize 时，如果某线程空闲时间超过keepAliveTime，线程将被终止
        allow-core-thread-timeout: true       # 是否允许核心线程超时,默认true
        queue-capacity: 100                   # 线程队列的大小,默认Integer.MAX_VALUE
      shutdown:
        await-termination: false              # 线程关闭等待
      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。

// TaskExecutionAutoConfiguration#applicationTaskExecutor()
@Lazy
@Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,
      AsyncAnnotationBeanPostProcessor.DEFAUL
          T_TASK_EXECUTOR_BEAN_NAME })
@ConditionalOnMissingBean(Executor.class)
public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
   return builder.build();
}

// ThreadPoolTaskExecutor#initializeExecutor()
@Override
protected ExecutorService initializeExecutor(
      ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

   BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

   ThreadPoolExecutor executor;
   if (this.taskDecorator != null) {
      executor = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler) {
         @Override
         public void execute(Runnable command) {
            Runnable decorated = taskDecorator.decorate(command);
            if (decorated != command) {
               decoratedTaskMap.put(decorated, command);
            }
            super.execute(decorated);
         }
      };
   }
   else {
      executor = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler);

   }

   if (this.allowCoreThreadTimeOut) {
      executor.allowCoreThreadTimeOut(true);
   }

   this.threadPoolExecutor = executor;
   return executor;
}

// ExecutorConfigurationSupport#initialize()
public void initialize() {
   if (logger.isInfoEnabled()) {
      logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
   }
   if (!this.threadNamePrefixSet && this.beanName != null) {
      setThreadNamePrefix(this.beanName + "-");
   }
   this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
}

覆盖默认的线程池

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

@Configuration
public class ThreadPoolConfiguration {

    @Bean("taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(50);
        taskExecutor.setQueueCapacity(200);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix("myExecutor--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(60);
        //修改拒绝策略为使用当前线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //初始化线程池
        taskExecutor.initialize();
        return taskExecutor;
    }
}

管理多个线程池

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

@Bean("taskExecutor2")
public ThreadPoolTaskExecutor taskExecutor2() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    //设置线程池参数信息
    taskExecutor.setCorePoolSize(10);
    taskExecutor.setMaxPoolSize(50);
    taskExecutor.setQueueCapacity(200);
    taskExecutor.setKeepAliveSeconds(60);
    taskExecutor.setThreadNamePrefix("myExecutor2--");
    taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    taskExecutor.setAwaitTerminationSeconds(60);
    //修改拒绝策略为使用当前线程执行
    taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    //初始化线程池
    taskExecutor.initialize();
    return taskExecutor;
}

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

@Resource
ThreadPoolTaskExecutor taskExecutor2;

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

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

线程池的四种拒绝策略

JAVA常用的四种线程池

ThreadPoolExecutor 类的构造函数如下：

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

newCachedThreadPool

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

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

newFixedThreadPool

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

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

newScheduledThreadPool

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

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

周期执行：

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

延时执行：

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

newSingleThreadExecutor

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

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

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

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

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

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

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

CallerRunsPolicy

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

public static class CallerRunsPolicy implements RejectedExecutionHandler {
 
    public CallerRunsPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

效果类似于：

Runnable thread = ()->{
   System.out.println(Thread.currentThread().getName());
   try {
      Thread.sleep(0);
   } catch (InterruptedException e) {
      throw new RuntimeException(e);
   }
};

thread.run();

AbortPolicy

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

public static class AbortPolicy implements RejectedExecutionHandler {
 
    public AbortPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

DiscardPolicy

什么也不做。

public static class DiscardPolicy implements RejectedExecutionHandler {
 
    public DiscardPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

DiscardOldestPolicy

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

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

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
 
    public DiscardOldestPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

Java 线程复用的原理

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

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

work对象的执行依赖于 runWorker()

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

Méthode 2 : injecter directement ThreadPoolTaskExecutor🎜🎜Vous ne pouvez pas ajouter l'annotation @EnableAsync pour le moment🎜rrreee🎜Informations de configuration par défaut du pool de threads🎜🎜Configuration commune du pool de threads SpringBoot :🎜rrreee🎜Implémentation du principe du pool de threads SpringBoot 🎜🎜 La classe TaskExecutionAutoConfiguration définit ThreadPoolTaskExecutor, et l'implémentation interne de cette classe est également basée sur la classe native ThreadPoolExecutor de Java. La méthode initializeExecutor() est appelée dans sa classe parent, mais dans la classe parent RejectedExecutionHandler est définie comme private RejectedExecutionHandler rejetéExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); et transmettez <code>AbortPolicy dans initializeExecutor() via la méthode initialize(). 🎜🎜Notez que dans la classe TaskExecutionAutoConfiguration, les noms de bean de la classe ThreadPoolTaskExecutor sont : applicationTaskExecutor et taskExecutor. 🎜rrreeerrreeerrreee🎜Remplacer le pool de threads par défaut🎜🎜Remplacer l'objet taskExecutor par défaut. Le type de retour du bean peut être ThreadPoolTaskExecutor ou Executor. 🎜rrreee🎜Gérer plusieurs pools de threads🎜🎜S'il existe plusieurs pools de threads, par exemple en définissant un autre pool de threads taskExecutor2, l'exécution directe signalera une erreur. A ce stade, vous devez spécifier le nom du bean. 🎜rrreee🎜Lors du référencement du pool de threads, vous devez remplacer le nom de la variable par le nom du bean, afin qu'il soit recherché par nom. 🎜rrreee🎜Pour le multithreading utilisant l'annotation @Async, spécifiez simplement le nom du bean dans l'annotation. 🎜rrreee🎜Quatre stratégies de rejet des pools de threads🎜🎜Quatre pools de threads couramment utilisés en JAVA🎜🎜ThreadPoolExecutor Le constructeur de la classe est le suivant : 🎜rrreee🎜newCachedThreadPool🎜🎜Aucune limite sur le nombre maximum de threads ( maximumPoolSize=Integer.MAX_VALUE), s'il y a plus de threads inactifs que nécessaire, ils seront recyclés, sinon les threads existants seront réutilisés. 🎜rrreee🎜newFixedThreadPool🎜🎜Pool de threads de longueur fixe, les tâches qui dépassent le nombre de threads attendront dans la file d'attente. 🎜rrreee🎜newScheduledThreadPool🎜🎜Semblable à newCachedThreadPool, il n'y a pas de limite supérieure sur le nombre de threads, mais corePoolSize peut être spécifié. Une exécution différée et une exécution périodique peuvent être obtenues. 🎜rrreee🎜Exécution périodique : 🎜rrreee🎜Exécution retardée : 🎜rrreee🎜newSingleThreadExecutor🎜🎜Pool de threads à thread unique, qui peut réaliser l'exécution séquentielle des threads. 🎜rrreee🎜Quatre stratégies de rejet dans le pool de threads Java🎜

• 🎜CallerRunsPolicy : Le pool de threads permet à l'appelant de s'exécuter. 🎜
• 🎜AbortPolicy : Si le pool de threads rejette la tâche, une erreur sera directement signalée. 🎜
• 🎜DiscardPolicy : Si le pool de threads rejette la tâche, elle sera directement rejetée. 🎜
• 🎜DiscardOldestPolicy : si le pool de threads rejette une tâche, la tâche la plus ancienne et non exécutée du pool de threads sera directement ignorée et la nouvelle tâche sera mise en file d'attente. 🎜
🎜🎜CallerRunsPolicy🎜🎜exécute la méthode run directement dans le thread principal. 🎜rrreee🎜L'effet est similaire à : 🎜rrreee🎜AbortPolicy🎜🎜 lève directement l'exception RejectedExecutionException et indique les informations sur la tâche et les informations sur le pool de threads. , 🎜rrreee🎜DiscardPolicy🎜🎜 ne fait rien. 🎜rrreee🎜DiscardOldestPolicy🎜
• 🎜e.getQueue().poll() : Supprime le plus ancien de la tâche de file d'attente. 🎜
• 🎜e.execute(r) : La tâche en cours est ajoutée à la file d'attente. 🎜
🎜rrreee🎜Le principe de la réutilisation des threads Java🎜🎜Le pool de threads de java enregistre l'objet java.util.concurrent.ThreadPoolExecutor.Worker. est conservé dans private final HashSet7b8f3a9cc722774601ac1c36ccacd929 Workers = new HashSet7b8f3a9cc722774601ac1c36ccacd929();. workQueue est une file d'attente qui stocke les tâches à exécuter Lorsqu'une nouvelle tâche est ajoutée au pool de threads, la tâche sera ajoutée à la file d'attente workQueue. 🎜rrreee🎜L'exécution de l'objet de travail repose sur runWorker(). Différent des threads que nous écrivons habituellement, ce thread est en boucle et obtient en permanence de nouvelles tâches de la file d'attente pour exécution. Par conséquent, les threads du pool de threads peuvent être réutilisés, au lieu de se terminer après l’exécution comme les threads que nous utilisons habituellement. 🎜rrreee

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!