Jdk提供线程池-原理、解析和使用

Love The Way You Lie 2023-07-15 05:55 4阅读 0赞

#### 1.背景 ####

我们都很清楚java创建线程的方式：1.继承Thread 2.实现Runnable接口，但是在项目中，我们几乎从来不会这样使用，因为线程创建和销毁都很耗费系统资源（CPU、IO、内存等），那么问题来了，如果能有一种工具能在我们需要的时候（eg：前端页面请求、后端定时任务等）提供创建线程，当线程执行完成以后不进行销毁，直接被工具回收回去，进入闲置状态，等待被再次使用。这样的话，资源利用率就大大提高了，而jdk已经为我们考虑到了这些，jdk1.5开始出现了线程池，我们通过调用线程池中线程供我们使用，当使用完成以后，让线程池回收该线程。

#### 2.线程池ThreadPoolExecutor ####

jdk1.5开始，jdk提供线程池ThreadPoolExecutor，下面我们来一起看看ThreadPoolExecutor

##### 2.1 类图 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70]

*  **Executor** 根接口，提供executor方法执行任务，如下

package java.util.concurrent;
    
    public interface Executor { 
        void execute(Runnable command);
    }

*  **ExecutorService** 线程池接口，提供了线程池生命周期方法，继承自Executor接口
 *  **ThreadPoolExecutor** 线程池实现类，提供了线程池的维护操作等相关方法，继承自AbstractExecutorService，AbstractExecutorService实现了ExecutorService接口。

##### 2.2 体系结构 #####

java.util.concurrent.Executor：负责线程的使用和调度的根接口  
|–ExecutorService 子接口： 线程池的主要接口  
|–ThreadPoolExecutor：线程池的实现类  
|–ScheduledExceutorService 子接口： 负责线程的调度  
|–ScheduledThreadPoolExecutor ：继承ThreadPoolExecutor，实现了ScheduledExecutorService

##### 2.3 类构造函数和参数说明 #####

这里参数含义请求参考方法注释（不会英文的程序员肯定是最low的）

/** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters. * * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached * @throws IllegalArgumentException if one of the following holds: * {@code corePoolSize < 0} * {@code keepAliveTime < 0} * {@code maximumPoolSize <= 0} * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} * or {@code threadFactory} or {@code handler} is null */
 public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 ThreadFactory threadFactory,
 RejectedExecutionHandler handler) { 
 if (corePoolSize < 0 ||
 maximumPoolSize <= 0 ||
 maximumPoolSize < corePoolSize ||
 keepAliveTime < 0)
 throw new IllegalArgumentException();
 if (workQueue == null || threadFactory == null || handler == null)
 throw new NullPointerException();
 this.acc = System.getSecurityManager() == null ?
 null :
 AccessController.getContext();
 this.corePoolSize = corePoolSize;
 this.maximumPoolSize = maximumPoolSize;
 this.workQueue = workQueue;
 this.keepAliveTime = unit.toNanos(keepAliveTime);
 this.threadFactory = threadFactory;
 this.handler = handler;
 }
 
 public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue) { 
 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
 Executors.defaultThreadFactory(), defaultHandler);
 }
 
 /** * The default rejected execution handler */
 private static final RejectedExecutionHandler defaultHandler =
 new AbortPolicy();

**说明**

*  默认线程工厂：Executors.defaultThreadFactory()，Executors工具类下一节介绍
 *  线程池ThreadPoolExecutor参数RejectedExecutionHandler：当线程池已满，执行被阻塞，无法创建线程来执行任务的处理方式，默认：AbortPolicy。处理方式总共4种，如下：
    
     *  AbortPolicy：用于被拒绝任务的处理程序，它将抛出RejectedExecutionException
     *  CallerRunsPolicy：用于被拒绝任务的处理程序，它直接在execute方法的调用线程中运行被拒绝的任务。
     *  DiscardOldestPolicy：用于被拒绝任务的处理程序，它放弃最旧的未处理请求，然后重试execute。
     *  DiscardPolicy：用于被拒绝任务的处理程序，它后台丢弃被拒绝的任务。

##### 2.4 线程池处理流程 #####

当一个任务被提交到线程池时，执行以下步骤：

1.  查看核心线程池是否已满，不满就创建一条线程执行任务，否则执行第二步。
2.  查看任务队列是否已满，不满就将任务存储在任务队列中，否则执行第三步。
3.  查看线程池是否已满（线程池线程梳理是否大于maximumPoolSize），不满就创建一条线程执行任务，否则就按照策略处理无法执行的任务。

#### 3.创建线程池工具类Executors ####

我们当然可以项目通过构造方法创建线程池，jdk为了方便我们创建线程池，提供了工具类Executors，Executors创建常用4种线程池源码如下：

public static ExecutorService newFixedThreadPool(int nThreads) { 
 return new ThreadPoolExecutor(nThreads, nThreads,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());
 }
 
 public static ExecutorService newSingleThreadExecutor() { 
 return new FinalizableDelegatedExecutorService
 (new ThreadPoolExecutor(1, 1,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>()));
 }
 
 public static ExecutorService newCachedThreadPool() { 
 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
 60L, TimeUnit.SECONDS,
 new SynchronousQueue<Runnable>());
 }
 
 /** * Creates a thread pool that can schedule commands to run after a * given delay, or to execute periodically. * @param corePoolSize the number of threads to keep in the pool, * even if they are idle * @return a newly created scheduled thread pool * @throws IllegalArgumentException if {@code corePoolSize < 0} */
 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
 return new ScheduledThreadPoolExecutor(corePoolSize);
 }

##### 3.1 说明 #####

ExecutorService\#newFixedThreadPool() : 创建固定大小的线程池  
ExecutorService\#newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。  
ExecutorService\#newSingleThreadExecutor() : 创建单个线程池。 线程池中只有一个线程

ScheduledExecutorService\#newScheduledThreadPool() : 创建固定大小的线程，可以延迟或定时的执行任务

根据Executors.java的源码比较性能:
 
 	
 	/** * Creates a thread pool that reuses a fixed number of threads * operating off a shared unbounded queue. At any point, at most * {@code nThreads} threads will be active processing tasks. * If additional tasks are submitted when all threads are active, * they will wait in the queue until a thread is available. * If any thread terminates due to a failure during execution * prior to shutdown, a new one will take its place if needed to * execute subsequent tasks. The threads in the pool will exist * until it is explicitly {@link ExecutorService#shutdown shutdown}. * * @param nThreads the number of threads in the pool * @return the newly created thread pool * @throws IllegalArgumentException if {@code nThreads <= 0} */
 public static ExecutorService newFixedThreadPool(int nThreads) { 
 return new ThreadPoolExecutor(nThreads, nThreads,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());
 }
 => LinkedBlockingQueue.java构造方法
 /** * Creates a {@code LinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}. */
 public LinkedBlockingQueue() { 
 this(Integer.MAX_VALUE);
 }
 	/** * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity} is not greater * than zero */
 public LinkedBlockingQueue(int capacity) { 
 if (capacity <= 0) throw new IllegalArgumentException();
 this.capacity = capacity;
 last = head = new Node<E>(null);
 }
 => 1)创建固定大小的线程池,核心线程数和最大线程数相等，且为指定大小
 2)非核心线程数空闲超时时间为0L，意味着:一旦非核心线程数空闲不执行任务，立马回收
 3)阻塞队列BlockingQueue为LinkedBlockingQueue，队列容量为Integer.MAX_VALUE
 缺点:允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
 
 	/** * Creates an Executor that uses a single worker thread operating * off an unbounded queue. (Note however that if this single * thread terminates due to a failure during execution prior to * shutdown, a new one will take its place if needed to execute * subsequent tasks.) Tasks are guaranteed to execute * sequentially, and no more than one task will be active at any * given time. Unlike the otherwise equivalent * {@code newFixedThreadPool(1)} the returned executor is * guaranteed not to be reconfigurable to use additional threads. * * @return the newly created single-threaded Executor */
 public static ExecutorService newSingleThreadExecutor() { 
 return new FinalizableDelegatedExecutorService
 (new ThreadPoolExecutor(1, 1,
 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>()));
 }
 => 1)创建线程池核心线程数和最大线程数都为1
 2)非核心线程数空闲超时时间为0L，意味着:一旦非核心线程数空闲不执行任务，立马回收
 3)阻塞队列BlockingQueue实现为LinkedBlockingQueue，队列容量为Integer.MAX_VALUE
 缺点:允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
 注意:跟newFixedThreadPool(1)是有区别的，返回线程池执行器不允许重新配置
 
 	/** * Creates a thread pool that creates new threads as needed, but * will reuse previously constructed threads when they are * available. These pools will typically improve the performance * of programs that execute many short-lived asynchronous tasks. * Calls to {@code execute} will reuse previously constructed * threads if available. If no existing thread is available, a new * thread will be created and added to the pool. Threads that have * not been used for sixty seconds are terminated and removed from * the cache. Thus, a pool that remains idle for long enough will * not consume any resources. Note that pools with similar * properties but different details (for example, timeout parameters) * may be created using {@link ThreadPoolExecutor} constructors. * * @return the newly created thread pool */
 public static ExecutorService newCachedThreadPool() { 
 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
 60L, TimeUnit.SECONDS,
 new SynchronousQueue<Runnable>());
 }
 => 1)核心线程数为0，最大线程数为Integer.VALUE
 2)非核心线程数空闲超时时间为60L，意味着:一旦非核心线程数空闲不执行任务，立马回收
 3)阻塞队列BlockingQueue采用SynchronousQueue，SynchronousQueue队列并没有内部容量，它仅仅起到一个新建进程的过渡作用，仅仅当另一个线程移除元素时，它才存储一个元素
 缺点:允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。
 
 	/** * Creates a thread pool that can schedule commands to run after a * given delay, or to execute periodically. * @param corePoolSize the number of threads to keep in the pool, * even if they are idle * @return a newly created scheduled thread pool * @throws IllegalArgumentException if {@code corePoolSize < 0} */
 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
 return new ScheduledThreadPoolExecutor(corePoolSize);
 }
 
 ==> ScheduledThreadPoolExecutor.java的构造方法
 /** * Creates a new {@code ScheduledThreadPoolExecutor} with the * given core pool size. * * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @throws IllegalArgumentException if {@code corePoolSize < 0} */
 public ScheduledThreadPoolExecutor(int corePoolSize) { 
 super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
 new DelayedWorkQueue());
 }
 => 1)核心线程数固定大小，最大线程数:Integer.MAX_VALUE
 2)用来创建线程延迟或周期性执行定时任务
 3)阻塞队列BlockingQueue采用DelayedWorkQueue，DelayedWorkQueue是特定的延迟队列，为了配合ThreadPoolExecutor的构造函数，需要实现BlockingQueue，初始容量大小:16
 缺点:允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

##### 3.2 阿里开发手册不推荐通过Executors创建线程池 #####

[阿里开发手册（华山版）.pdf 链接][.pdf] 提取码: 82xk ，编程规约->并发处理，如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70 1]

##### 3.3 Executors方法运用 #####

实际项目中，我们很少会自己通过new ThreadPoolExecutor方法创建线程池，通过我们会采用Spring对ThreadPoolExecutor封装的ThreadPoolTaskExecutor对其进行线程池的配置(在下一节做介绍)。这里以Executors提供的方法创建线程池编写测试方法，如下：

package com.wisely.highlight_springmvc4.thread.pool;
 
 import java.util.concurrent.*;
 
 public class ThreadPoolTest1 { 
 
 public static void main(String[] args) { 
 // Step1:创建固定大小的线程池
 ExecutorService exec = Executors.newFixedThreadPool(10);
 Callable<String> call = new Callable<String>() { 
 public String call() throws Exception { 
 //开始执行耗时操作
 Thread.sleep(1000 * 2);
 return "线程执行完成.";
 }
 };
 
 try { 
 Future<String> future = exec.submit(call);
 // 任务处理超时时间设为 1 秒
 String obj = future.get(1000 * 1, TimeUnit.MILLISECONDS);
 System.out.println("任务成功返回:" + obj);
 } catch (TimeoutException ex) { 
 System.out.println("处理超时啦....");
 ex.printStackTrace();
 } catch (Exception e) { 
 System.out.println("处理失败.");
 e.printStackTrace();
 }
 // 关闭线程池
 exec.shutdown();
 }
 
 }

**注意：**

*  Callable回调函数，Future\#get(long timeout, TimeUnit unit)可以设置任务执行超时时间，可以结合分布式锁（redis+zookeeper实现的分布式锁）一起使用

#### 4. Spring线程池封装ThreadPoolTaskExecutor ####

Spring对jdk线程池ThreadPoolExecutor进行了封装，封装类为ThreadPoolTaskExecutor

##### 4.1 结构 #####

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport
    		implements AsyncListenableTaskExecutor, SchedulingTaskExecutor { 
    
    	private final Object poolSizeMonitor = new Object();
    
    	private int corePoolSize = 1;
    
    	private int maxPoolSize = Integer.MAX_VALUE;
    
    	private int keepAliveSeconds = 60;
    
    	private int queueCapacity = Integer.MAX_VALUE;
    
    	private boolean allowCoreThreadTimeOut = false;
    
    	private ThreadPoolExecutor threadPoolExecutor;

为什么说Spring是对ThreadPoolExecutor进行了封装，对ThreadPoolTaskExecutor配置，就是对ThreadPoolTaskExecutor配置，我们先来看下ThreadPoolTaskExecutor容器初始化方法

##### 4.2 类图![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70 2] #####

从类图中我们可以看到，ThreadPoolTaskExecutor继承了ExecutorConfigurationSupport，ThreadPoolTaskExecutor提供了初始化线程池方法initializeExecutor，继承了ExecutorConfigurationSupport方法afterPropertiesSet和initialize方法，容器初始化时会调用afterPropertiesSet方法如下：

ThreadPoolTaskExecutor.java

/** * Note: This method exposes an {@link ExecutorService} to its base class * but stores the actual {@link ThreadPoolExecutor} handle internally. * Do not override this method for replacing the executor, rather just for * decorating its {@code ExecutorService} handle or storing custom state. */
 @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) { 
 super.execute(taskDecorator.decorate(command));
 }
 };
 }
 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.java

/** * Calls {@code initialize()} after the container applied all property values. * @see #initialize() */
    @Override
    public void afterPropertiesSet() { 
        initialize();
    }
    
    /** * Set up the ExecutorService. */
    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);
    }

以上是Spring提供线程池封装类ThreadPoolTaskExecutor初始化线程池ThreadPoolExecutor过程，ThreadPoolTaskExecutor提供的方法都是对内置成员变量线程池ThreadPoolExecutor访问，这里以线程池核心线程数的方法做介绍，ThreadPoolTaskExecutor中corePoolSize的Getter和Setter方法如下：

/** * Set the ThreadPoolExecutor's core pool size. * Default is 1. * This setting can be modified at runtime, for example through JMX. */
 	public void setCorePoolSize(int corePoolSize) { 
 		synchronized (this.poolSizeMonitor) { // 线程池锁
 			this.corePoolSize = corePoolSize;
 			if (this.threadPoolExecutor != null) { 
 				this.threadPoolExecutor.setCorePoolSize(corePoolSize);
 			}
 		}
 	}
 
 	/** * Return the ThreadPoolExecutor's core pool size. */
 	public int getCorePoolSize() { 
 		synchronized (this.poolSizeMonitor) { 
 			return this.corePoolSize;
 		}
 	}

从上面方法，我们可以看到这里采用了**代理模式**，ThreadPoolTaskExecutor帮我们代理访问ThreadPoolExecutor，对ThreadPoolTaskExecutor访问就是对ThreadPoolExecutor的访问。

##### 4.3 案例 #####

**1）线程池配置**

这里以基于java配置为例（基于xml配置存在参数限制，配置信息不全）

package com.wisely.highlight_springmvc4.thread.pool;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.lang.reflect.Method;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * Created by dujiayong on 2020/3/1. */
    @Configuration
    @EnableAsync // 开启对异步任务支持
    public class MyThreadPoolExecutor implements AsyncConfigurer{ 
    
        private static final Logger log = LoggerFactory.getLogger(MyThreadPoolExecutor.class);
    
        /** * 实现AsyncConfigurer接口，重写getAsyncExecutor方法，自定义线程池 * 重写getAsyncUncaughtExceptionHandler，自定义异常处理类 * @return */
        @Bean(name = "asyncExecutor")
        @Override
        public Executor getAsyncExecutor() { 
            ThreadPoolTaskExecutor asyncExecutor = new ThreadPoolTaskExecutor();
            asyncExecutor.setCorePoolSize(5);
            asyncExecutor.setMaxPoolSize(100);
            asyncExecutor.setKeepAliveSeconds(60);
            asyncExecutor.setQueueCapacity(100);
            asyncExecutor.setThreadNamePrefix("asyncExecutor-custom-");
            // 当pool已经达到max size的时候，如何处理新任务，不在新线程中执行，而是由调用者所在的线程执行
            asyncExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
            return asyncExecutor;
        }
    
        /** * 自定义线程池:ThreadPoolTaskExecutor，返回参数类型:ThreadPoolTaskExecutor * @return */
        @Bean(name = "threadPoolTaskExecutor")
        public ThreadPoolTaskExecutor threadPoolTaskExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(10);
            executor.setMaxPoolSize(100);
            executor.setKeepAliveSeconds(60);
            executor.setQueueCapacity(100);
            executor.setThreadNamePrefix("MyThreadPoolExecutor-custom-");
            // 当pool已经达到max size的时候，如何处理新任务，不在新线程中执行，而是由调用者所在的线程执行
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            return executor;
        }
    
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { 
            return new MyAsyncExceptionHandler();
        }
    
        /** * 自定义异常处理类 */
        static class MyAsyncExceptionHandler implements AsyncUncaughtExceptionHandler { 
            @Override
            public void handleUncaughtException(Throwable throwable, Method method, Object... objects) { 
                log.error("Exception message - " + throwable.getMessage());
                log.error("Method name - " + method.getName());
                for(Object param:objects){ 
                    log.error("Parameter value -" + param);
                }
            }
        }
    }

**2）线程池使用**

package com.wisely.highlight_springmvc4.web.threadpool;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 /** * Created by dujiayong on 2020/3/2. */
 @RestController
 @RequestMapping(value = "/threadPoolTest")
 public class ThreadPoolTestController { 
 
 @Autowired
 ThreadPoolTaskExecutor threadPoolTaskExecutor;
 
 @RequestMapping(value = "/stat")
 public String stat() { 
 for (int i = 0; i < 15; i++) { 
 threadPoolTaskExecutor.execute(() -> { 
 try { 
 Thread.sleep(1000 * 2);
 System.out.println("当前时间:" + System.currentTimeMillis() + ",线程:" + Thread.currentThread().getName());
 } catch (InterruptedException e) { 
 e.printStackTrace();
 }
 });
 }
 System.out.println("结束...,时间:" + System.currentTimeMillis());
 return "success";
 }
 }

执行结果如下：

结束...,时间:1584235693194
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-1
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-3
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-2
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-4
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-5
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-6
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-10
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-8
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-7
    当前时间:1584235695207,线程:MyThreadPoolExecutor-custom-9
    当前时间:1584235697218,线程:MyThreadPoolExecutor-custom-1
    当前时间:1584235697218,线程:MyThreadPoolExecutor-custom-3
    当前时间:1584235697218,线程:MyThreadPoolExecutor-custom-2
    当前时间:1584235697218,线程:MyThreadPoolExecutor-custom-5
    当前时间:1584235697218,线程:MyThreadPoolExecutor-custom-4

由此可以ThreadPoolTaskExecutor有10个核心线程，15个并发任务，前10个由核心线程处理，另外5个对队列，等前10个任务执行完毕，核心线程再去执行任务队列中的任务。

**完毕**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70]: /images/20230528/4003ad0c87154693b1c858c20ec3c57c.png
[.pdf]: https://pan.baidu.com/s/1gwPBGTTAPC-XJj0qf0rbXg
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70 1]: /images/20230528/c053a7af422443fb80ad3c02b8e65fc7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RqeV8yMDE3_size_16_color_FFFFFF_t_70 2]: /images/20230528/3784a93f44494d6f93e94af4c9d788c9.png