跟我学Java多线程——ThreadPoolExecutor(线程池)

小鱼儿 2022-06-12 09:52 210阅读 0赞

# **什么是线程池** #

多线程开发中，由于线程数量多，并且每个线程执行一段时间就结束，所以要频繁的创建线程，但是这样频繁的创建线程会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。在这种情况下，人们就想要一种可以线程执行完后不用销毁，同时该线程还可以去执行其他任务，在这样的情况下线程池就出现了。

线程池就是线程的池子，任务提交到线程池后，就从线程池中取出一个空闲的线程为之服务，服务完后不销毁该线程，而是将该线程还回到线程池中。

在线程池的编程模式下，任务是提交给整个线程池，而不是直接交给某个线程，线程池在拿到任务后，它就在内部找有无空闲的线程，再把任务交给内部某个空闲的线程，如果没有空闲的进程，任务就处于等待状态，这就是封装。

使用了线程池后减少了创建和销毁线程的次数，每个线程都可以被重复利用，可执行多个任务；同时可以根据系统的承受能力，调整线程池中线程的数目，避免出现将系统内存消耗完毕这样的情况出现。

# **ThreadPoolExecuto例子** #

我们先通过一个简单的线程池的例子先来对ThreadPoolExecuto进行一个整体的认识，我们构造了一个正常线程数量为5，最大线程池数量为10，任务缓存队列5的线程池，让这个线程池来执行15个任务的demo。

先来看下MyRunnable类，用来记录当前正在执行的任务数：

package com.tgb.threadpool;
    
    /**
     * 线程执行
     * @author kang
     *
     */
    public class MyRunnable implements Runnable {
    
    	// 正在执行的任务数
    	private int num;
    
    	public MyRunnable(int num) {
    		this.num = num;
    	}
    
    	@Override
    	public void run() {
    		System.out.println("正在执行的MyRunnable " + num);
    		try {
    			Thread.currentThread().sleep(4000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		System.out.println("MyRunnable " + num + "执行完毕");
    
    	}
    
    }

线程池客户端：

package com.tgb.threadpool;
 
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 /**
 * 线程池调用客户端
 * @author kang
 *
 */
 public class ThreadPoolExecutorTest {
 
 	//池中所保存的线程数，包括空闲线程。
 	final static int corePoolSize = 5;
 	//池中允许的最大线程数。
 	final static int maximumPoolSize = 10;
 	//当线程数大于核心线程时，此为终止前多余的空闲线程等待新任务的最长时间
 	final static long keepAliveTime = 200;
 	//执行前用于保持任务的队列5，即任务缓存队列
 	final static ArrayBlockingQueue<Runnable> workQueue =new ArrayBlockingQueue<Runnable>(5);
 	
 	public static void main(String[] args) {
 		//构建一个线程池，正常线程数量为5，最大线程数据为10，等待时间200
 		ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
 				corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES,	workQueue);
 		
 		//线程池去执行15个任务
 		for (int i = 0; i < 15; i++) {
 			MyRunnable myRunnable = new MyRunnable(i);
 			threadPoolExecutor.execute(myRunnable);
 			System.out.println("线程池中现在的线程数目是："+threadPoolExecutor.getPoolSize()+", 队列中正在等待执行的任务数量为："+
 					threadPoolExecutor.getQueue().size());
 		}
 		//关掉线程池
 		threadPoolExecutor.shutdown();
 		
 	}
 	
 }

执行结果：

正在执行的MyRunnable 0
    线程池中现在的线程数目是：1,  队列中正在等待执行的任务数量为：0
    线程池中现在的线程数目是：2,  队列中正在等待执行的任务数量为：0
    正在执行的MyRunnable 1
    线程池中现在的线程数目是：3,  队列中正在等待执行的任务数量为：0
    正在执行的MyRunnable 2
    线程池中现在的线程数目是：4,  队列中正在等待执行的任务数量为：0
    正在执行的MyRunnable 3
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：0
    正在执行的MyRunnable 4
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：1
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：2
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：3
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：4
    线程池中现在的线程数目是：5,  队列中正在等待执行的任务数量为：5
    线程池中现在的线程数目是：6,  队列中正在等待执行的任务数量为：5
    正在执行的MyRunnable 10
    线程池中现在的线程数目是：7,  队列中正在等待执行的任务数量为：5
    正在执行的MyRunnable 11
    线程池中现在的线程数目是：8,  队列中正在等待执行的任务数量为：5
    正在执行的MyRunnable 12
    线程池中现在的线程数目是：9,  队列中正在等待执行的任务数量为：5
    正在执行的MyRunnable 13
    线程池中现在的线程数目是：10,  队列中正在等待执行的任务数量为：5
    正在执行的MyRunnable 14
    MyRunnable 0执行完毕
    正在执行的MyRunnable 5
    MyRunnable 3执行完毕
    MyRunnable 4执行完毕
    MyRunnable 2执行完毕
    正在执行的MyRunnable 7
    正在执行的MyRunnable 6
    正在执行的MyRunnable 8
    MyRunnable 1执行完毕
    正在执行的MyRunnable 9
    MyRunnable 12执行完毕
    MyRunnable 11执行完毕
    MyRunnable 10执行完毕
    MyRunnable 14执行完毕
    MyRunnable 13执行完毕
    MyRunnable 5执行完毕
    MyRunnable 9执行完毕
    MyRunnable 8执行完毕
    MyRunnable 6执行完毕
    MyRunnable 7执行完毕

从执行结果来看，当线程池中线程的数量大于 5 时，便将任务放入任务缓存队列里面，任务缓存队列满了以后，变创建新的线程，来执行，如果让任务数量大于 15 ，程序就会报错，抛出任务拒绝异常，这时可以增大任务缓存队列，有兴趣的同学可以去试试。

通过这个demo应该可以对线程池有了一个宏观的理解了吧，下面我们就继续深入的来学习ThreadPoolExecutor对象。

# **ThreadPoolExecutor源码** #

java.uitl.concurrent.ThreadPoolExecutor这个类是线程池中最核心的一个类，所以我们就通过来ThreadPoolExecutor的源码来进一步理解线程池：

public class ThreadPoolExecutor extends AbstractExecutorService {
 
 // Public constructors and methods
 
 /**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters and default thread factory and rejected execution handler.
 * It may be more convenient to use one of the {@link Executors} factory
 * methods instead of this general purpose constructor.
 *
 * @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.
 * @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} is null
 */
 public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue) {
 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
 Executors.defaultThreadFactory(), defaultHandler);
 }
 
 /**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters and default rejected execution handler.
 *
 * @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
 * @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} is null
 */
 public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 ThreadFactory threadFactory) {
 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
 threadFactory, defaultHandler);
 }
 
 /**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters and default thread factory.
 *
 * @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 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 handler} is null
 */
 public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 RejectedExecutionHandler handler) {
 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
 Executors.defaultThreadFactory(), handler);
 }
 
 /**
 * 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.corePoolSize = corePoolSize;
 this.maximumPoolSize = maximumPoolSize;
 this.workQueue = workQueue;
 this.keepAliveTime = unit.toNanos(keepAliveTime);
 this.threadFactory = threadFactory;
 this.handler = handler;
 }

通过代码我们发现，ThreadPoolExecutor继承了AbstractExecutorService类继承了AbstractExecutorService类，并且提供了四个构造方法，大家看完就会发现，其实前面的三个构造方法调用的都是第四个构造方法进行的初始化工作。

我们接下来看下构造器中各个参数的定义，结合jdk-api加入自己的理解:

corePoolSize- 池中所保存的核心线程数，包括空闲线程，默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；

maximumPoolSize -池中允许的最大线程数。

keepAliveTime- 当线程数大于核心数时，线程没有任务执行时最多保持多久时间就会终止，默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用；如果线程池中的线程数小于corePoolSize，一个线程空闲的时间达到keepAliveTime，就会终止

unit -参数keepAliveTime的时间单位，在TimeUnit类中有有七种静态属性：

TimeUnit.DAYS;               //天
    TimeUnit.HOURS;             //小时
    TimeUnit.MINUTES;           //分钟
    TimeUnit.SECONDS;           //秒
    TimeUnit.MILLISECONDS;      //毫秒
    TimeUnit.MICROSECONDS;      //微妙
    TimeUnit.NANOSECONDS;       //纳秒

workQueue- 执行前用于保持任务的队列 ，也就是用来存储等待执行的任务 。此队列仅保持由 execute 方法提交的 Runnable 任务。 一般我们也称它为阻塞队列，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，关于阻塞队列的介绍我们放到下篇文章；

threadFactory -执行程序创建新线程时使用的工厂；

handler -由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序，常见有以下几种：

ThreadPoolExecutor.AbortPolicy:放弃任务抛异常。
    ThreadPoolExecutor.DiscardPolicy：放弃任务不抛异常。
    ThreadPoolExecutor.DiscardOldestPolicy：放弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
    ThreadPoolExecutor.CallerRunsPolicy：调用线程处理该任务

通过代码我们发现，我们的ThreadPoolExecutor类继承了AbstractExecutorService抽象类，我们接下来看下AbstractExecutorService抽象类 :

public abstractclass AbstractExecutorService implements ExecutorService {
    …………………………………………………..

关于AbstractExecutorService类的详细代码我们就不再展示了，它实现了ExecutorService接口，提供ExecutorService 执行方法的默认实现。此类使用 newTaskFor 返回的 RunnableFuture实现 submit、invokeAny和 invokeAll 方法。

我们接着来看ExecutorService接口：

public interfaceExecutorService extends Executor {
    …………………………………………………..

ExecutorService接口又继承了 Executor 接口，ExecutorService接口是真正的线程池接口，ExecutorService是可以关闭的，这将导致其拒绝新任务。提供两个方法来关闭ExecutorService。shutdown()方法在终止前允许执行以前提交的任务，而 shutdownNow()方法阻止等待任务启动并试图停止当前正在执行的任务。在终止时，执行程序没有任务在执行，也没有任务在等待执行，并且无法提交新任务。应该关闭未使用的ExecutorService 以允许回收其资源。

ExecutorService还可以通过创建并返回一个可用于取消执行和/或等待完成的Future，方法 submit 扩展了基本方法 Executor.execute(java.lang.Runnable)。方法 invokeAny 和invokeAll 是批量执行的最常用形式，它们执行任务 collection，然后等待至少一个，或全部任务完成（可使用ExecutorCompletionService 类来编写这些方法的自定义变体）。

接下来看Executor接口

public interface Executor {
    
        /**
         * Executes the given command at some time in the future.  The command
         * may execute in a new thread, in a pooled thread, or in the calling
         * thread, at the discretion of the {@code Executor} implementation.
         *
         * @param command the runnable task
         * @throws RejectedExecutionException if this task cannot be
         * accepted for execution
         * @throws NullPointerException if command is null
         */
        void execute(Runnable command);
    }

这个接口只有一个 execute 方法，执行已提交的Runnable 任务的对象，所以 Executor  接口并不是一个线程池只是一个执行线程的工具。

到这里大家应该明白ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了Executor是最顶层的接口，是来执行线程任务的；ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；最后我们的ThreadPoolExecutor继承了类AbstractExecutorService。

在我们的ThreadPoolExecutor类中有几个常用的方法：

execute()、submit()、shutdown()、shutdownNow()，其他方法大家可以去查api

其中execute()方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。

submit()方法是在ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。

shutdown()和shutdownNow()比较简单是用来关闭线程的。

# 总结 #

关于线程池ThreadPoolExecutor我们本篇文章讲解了线程池是什么以及使用线程池的好处，并且通过一个简单的demo来让大家对线程池有一个整体认识，最后介绍了线程池ThreadPoolExecutor类的与父类以及接口的关系，关于ThreadPoolExecutor的介绍就先到这里。

博客来源：[http://blog.csdn.net/zwk626542417/article/details/47084925][http_blog.csdn.net_zwk626542417_article_details_47084925]

[http_blog.csdn.net_zwk626542417_article_details_47084925]: http://blog.csdn.net/zwk626542417/article/details/47084925