Java并发编程核心方法与框架-TheadPoolExecutor的使用

2022-02-02 15:52:39

类ThreadPoolExecutor最常使用的构造方法是

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, unit, BlockingDeque<Runnable> workQueue)

corePoolSize 线程池中所保存的线程数，包括空闲线程，也就是核心池的大小。当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有基本线程。
maximumPoolSize 池中允许的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用了*的任务队列这个参数就没什么效果。
keepAliveTime 当线程数量大于corePoolSize值时，在没有超过指定的时间内是不从线程池中将空闲线程删除的。如果超过此时间，则删除。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。
unit keepAliveTime的时间单位
workQueue 执行前用于保持任务的队列，此队列仅保持由execute方法提交的Runnable任务

ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。

LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO（先进先出）排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。

SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。

PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

其他常用的构造方法参数：

ThreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。
RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。以下是JDK1.5提供的四种策略。

AbortPolicy：直接抛出异常。
CallerRunsPolicy：只用调用者所在线程来运行任务。
DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
DiscardPolicy：不处理，丢弃掉。

当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。

当提交一个新任务到线程池时，线程池的处理流程如下：

线程池判断基本线程池是否已满？没满，创建一个工作线程来执行任务。满了，则进入下个流程。
线程池判断工作队列是否已满？没满，则将新提交的任务存储在工作队列里。满了，则进入下个流程。
线程池判断整个线程池是否已满？没满，则创建一个新的工作线程来执行任务，满了，则交给饱和策略来处理这个任务。

/*

* 获取基本属性corePoolSize和maximumPoolSize

*/

public class Run {

	public static void main(String[] args) {

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		LinkedBlockingDeque<Runnable> workQueue = new LinkedBlockingDeque<>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		System.out.println(executor.getCorePoolSize());//7

		System.out.println(executor.getMaximumPoolSize());//8

		System.out.println("---------------------------");

		SynchronousQueue<Runnable> synchronousQueue = new SynchronousQueue<>();

		executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, synchronousQueue);

		System.out.println(executor.getCorePoolSize());//7

		System.out.println(executor.getMaximumPoolSize());//8

	}

}

如果欲执行的runnable的数量<=corePoolSize，则马上创建线程运行这个任务，并且不放入扩展队列Queue中。

/**

 * 队列使用LinkedBlockingDeque类，并且线程数量<=corePoolSize

 * 所以keepAliveTime>5时也不清除空闲线程

 */

public class Run2_1 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		LinkedBlockingDeque<Runnable> workQueue = new LinkedBlockingDeque<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//7

		System.out.println("A:" + executor.getQueue().size());//0

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

//如果欲执行的runnable的数量<=corePoolSize，则马上创建线程运行这个任务，并且不放入扩展队列Queue中。

/**

 * 队列使用SynchronousQueue类，并且线程数量<=corePoolSize

 * 所以keepAliveTime>5时也不清除空闲线程

 */

public class Run2_2 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " begin " + System.currentTimeMillis());

					TimeUnit.SECONDS.sleep(1);

					System.out.println(Thread.currentThread().getName() + " end " + System.currentTimeMillis());

				} catch (Exception e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<>();

//		LinkedBlockingDeque<Runnable> workQueue = new LinkedBlockingDeque<>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

//		TimeUnit.SECONDS.sleep(10);

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//7

		System.out.println("A:" + executor.getQueue().size());//0

		TimeUnit.SECONDS.sleep(10);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

数量>corePoolSize&&数量<=maxmimumPoolSize的情况

/*

 * BlockingQueue是一个接口，常用的实现类有LinkedBlockingQueue和ArrayBlockingQueue.

 * LinkedBlockingQueue的好处在于没有大小限制，队列容量非常大，所以执行execute()不会抛出异常

 * 线程池中运行的线程的数量永远也不会超过corePoolSize的值，

 * 因为多余的线程被放入LinkedBlockingQueue队列中，keepAliveTime参数也就没有意义了。

 */

public class Run3_1 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		//队列使用LinkedBlockingDeque类，如果线程数量>corePoolSize时将其余的任务放入队列中。

		//同一时间最多只能有7个线程执行。

		//使用LinkedBlockingDeque类时，maximumPoolSize参数被忽略

		LinkedBlockingDeque<Runnable> workQueue = new LinkedBlockingDeque<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		//executor.execute(runnable);//8

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//7

		System.out.println("A:" + executor.getQueue().size());//1

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

以上代码运行结果如下：

pool-1-thread-1 run!1470813110727

pool-1-thread-5 run!1470813110728

pool-1-thread-4 run!1470813110728

pool-1-thread-7 run!1470813110728

pool-1-thread-2 run!1470813110727

pool-1-thread-3 run!1470813110727

pool-1-thread-6 run!1470813110728

A:7

A:7

A:1

pool-1-thread-1 run!1470813111731

B:7

B:7

B:0

可见，线程pool-1-thread-1执行了两个任务。

public class Run3_2 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		/*

		 *队列使用SynchronousQueue类并且线程数量>corePoolSize时，将其余的线程放入池中，总数量为8

		 *并且线程总数量没有超过maximumPoolSize值8

		 *由于运行线程数为8，数量上>corePoolSize的值7

		 *所以keepAliveTime>5时清除空闲线程

		 *如果使用SynchronousQueue类则maximumPoolSize参数的作用将有效

		 */

		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		//executor.execute(runnable);//8

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//8

		System.out.println("A:" + executor.getQueue().size());//0

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

		//删除的是>corePoolSize的多余线程

	}

}

以上代码运行结果如下：

pool-1-thread-3 run!1470832826522

pool-1-thread-6 run!1470832826522

pool-1-thread-7 run!1470832826523

pool-1-thread-5 run!1470832826522

pool-1-thread-4 run!1470832826522

pool-1-thread-1 run!1470832826522

pool-1-thread-2 run!1470832826522

pool-1-thread-8 run!1470832826523

A:7

A:8

A:0

B:7

B:7

B:0

数量>maximumPoolSize的情况

public class Run4_1 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		/*

		 *队列使用LinkedBlockingDeque类并且线程数量>corePoolSize时，将其余的线程放入池中

		 *同一时间内只有corePoolSize个线程在运行

		 *所以keepAliveTime>5时不清除空闲线程

		 */

		LinkedBlockingDeque<Runnable> workQueue = new LinkedBlockingDeque<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		executor.execute(runnable);//9

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//7

		System.out.println("A:" + executor.getQueue().size());//2

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

运行结果如下：

pool-1-thread-1 run!1470833462484

pool-1-thread-4 run!1470833462484

pool-1-thread-3 run!1470833462484

pool-1-thread-2 run!1470833462484

pool-1-thread-6 run!1470833462484

pool-1-thread-5 run!1470833462484

pool-1-thread-7 run!1470833462485

A:7

A:7

A:2

pool-1-thread-5 run!1470833463489

pool-1-thread-4 run!1470833463489

B:7

B:7

B:0

public class Run4_2 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 10;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		/*

		 *队列使用SynchronousQueue类并且线程数量>=corePoolSize

		 *并且线程数量<=maximumPoolSize

		 *所以keepAliveTime>5时清除空闲线程

		 */

		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		executor.execute(runnable);//9

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//9

		System.out.println("A:" + executor.getQueue().size());//0

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

以上代码运行结果如下：

pool-1-thread-2 run!1470833720785

pool-1-thread-5 run!1470833720786

pool-1-thread-4 run!1470833720785

pool-1-thread-3 run!1470833720785

pool-1-thread-1 run!1470833720785

pool-1-thread-6 run!1470833720786

pool-1-thread-7 run!1470833720786

pool-1-thread-8 run!1470833720786

pool-1-thread-9 run!1470833720786

A:7

A:9

A:0

B:7

B:7

B:0

public class Run4_3 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 8;

		int keepAliveTime = 5;

		TimeUnit unit = TimeUnit.SECONDS;

		/*

		 *队列使用SynchronousQueue类并且线程数量>corePoolSize

		 *并且线程数量>maximumPoolSize

		 *所以出现异常

		 */

		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		executor.execute(runnable);//9

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//

		System.out.println("A:" + executor.getPoolSize());//

		System.out.println("A:" + executor.getQueue().size());//

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//

		System.out.println("B:" + executor.getPoolSize());//

		System.out.println("B:" + executor.getQueue().size());//

	}

}

运行结果如下：

pool-1-thread-2 run!1470833900655

pool-1-thread-6 run!1470833900655

pool-1-thread-5 run!1470833900655

pool-1-thread-4 run!1470833900655

pool-1-thread-1 run!1470833900655

pool-1-thread-7 run!1470833900656

pool-1-thread-3 run!1470833900655

pool-1-thread-8 run!1470833900656

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.concurrent.chapter4.concurrent07.Run4_3$1@c33f45e rejected from java.util.concurrent.ThreadPoolExecutor@6a754384[Running, pool size = 8, active threads = 8, queued tasks = 0, completed tasks = 0]

	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)

	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)

	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)

	at com.concurrent.chapter4.concurrent07.Run4_3.main(Run4_3.java:42)

参数keepAliveTime为0时的实验

public class Run5 {

	public static void main(String[] args) throws InterruptedException {

		Runnable runnable = new Runnable() {

			@Override

			public void run() {

				try {

					System.out.println(Thread.currentThread().getName() + " run!" + System.currentTimeMillis());

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					e.printStackTrace();

				}

			}

		};

		int corePoolSize = 7;

		int maximumPoolSize = 10;

		int keepAliveTime = 0;

		TimeUnit unit = TimeUnit.SECONDS;

		/*

		 *队列使用SynchronousQueue类并且线程数量<corePoolSize

		 *并且线程数量<=maximumPoolSize

		 *并且keepAliveTime值为0时 线程执行完毕后立即清除

		 */

		SynchronousQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();

		ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);

		executor.execute(runnable);//1

		executor.execute(runnable);//2

		executor.execute(runnable);//3

		executor.execute(runnable);//4

		executor.execute(runnable);//5

		executor.execute(runnable);//6

		executor.execute(runnable);//7

		executor.execute(runnable);//8

		executor.execute(runnable);//9

		Thread.sleep(300);

		System.out.println("A:" + executor.getCorePoolSize());//7

		System.out.println("A:" + executor.getPoolSize());//9

		System.out.println("A:" + executor.getQueue().size());//0

		Thread.sleep(10000);

		System.out.println("B:" + executor.getCorePoolSize());//7

		System.out.println("B:" + executor.getPoolSize());//7

		System.out.println("B:" + executor.getQueue().size());//0

	}

}

运行结果如下：

pool-1-thread-3 run!1470834464969

pool-1-thread-5 run!1470834464969

pool-1-thread-4 run!1470834464969

pool-1-thread-6 run!1470834464969

pool-1-thread-1 run!1470834464969

pool-1-thread-2 run!1470834464969

pool-1-thread-7 run!1470834464970

pool-1-thread-8 run!1470834464970

pool-1-thread-9 run!1470834464970

A:7

A:9

A:0

B:7

B:7

B:0

码农公寓

相关文章