线程池Thread Pool，是一个容纳多个线程的容器，里面的线程可以反复使用。
好处：
- 降低资源消耗：通过重复利用已创建的线程，降低线程创建和销毁造成的资源消耗。
- 提高响应速度：当有任务时，无需等待线程创建，就能立即执行。
- 提高线程的可管理性：线程池可以对线程进行统一的分配、调优和监控。
应用场景：Tomcat服务器、数据库连接池等。
代码：
javase/src/code16/test10/Task.java：
package code16.test10;
public class Task implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
javase/src/code16/test10/Test01.java：
package code16.test10;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Test01 {
public static void main(String[] args) {
Task task1 = new Task();
Task task2 = new Task();
Task task3 = new Task();
// Thread thread = new Thread(mr1, mr2, mr3); // 没有该写法，一个线程只能执行一个任务
// Thread thread1 = new Thread(mr1);
// Thread thread2 = new Thread(mr2);
// Thread thread3 = new Thread(mr3);
// // 线程不能复用，损耗资源
// 创建只有一个线程的线程池
ExecutorService threadPool = Executors.newSingleThreadExecutor();
//
// thread.start();
// thread1.start();
// thread2.start();
// thread3.start();
threadPool.execute(task1);
threadPool.execute(task2);
threadPool.execute(task3);
// 线程得到复用，降低资源损耗
// 关闭线程池
threadPool.shutdown();
}
}
java执行结果：
2. 线程池的API线程池相关的API都在java.util.concurrent包中。
|- Executor
|- ExecutorService
|- AbstractExecutorService
|- ThreadPoolExecutor // 核心线程池
|- ScheduledThreadPoolExecutor
|- ScheduledExecutorService // 调度线程池
|- ScheduledThreadPoolExecutor
3. 创建线程池
线程池的基本属性：
| 属性 | 含义 | 描述 |
|---|---|---|
| corePoolSize | 核心线程数 | 核心线程：只要线程池不关闭，就不会被销毁的线程 |
| maximumPoolSize | 最大线程数 | 线程池中最多允许有的线程数 |
| keepAliveTime | 空闲线程存活时间 | 非核心线程（线程池中除了核心线程以外的线程）没有执行任务时，超过空闲线程存活时间，就会被清理 |
| unit | 时间单位 | keepAliveTime的时间单位 |
| workQueue | 任务队列 | 存放任务（Runnable实例）的容器，提交给线程池的任务都位于这里，线程池中的线程也都是在该容器中领取任务。常用的有LinkedBlockingQueue、ArrayBlockingQueue等 |
| threadFactory | 线程工厂 | 指定线程的创建过程 |
| handler | 任务拒绝策略 | 指定线程池拒绝提交的任务后，执行的操作 |
核心线程池：ThreadPoolExecutor。
其它线程池：
- FixedThreadPool：固定线程数量的线程池。
- SingleThreadExecutor：只有一个线程的线程池。
- CachedThreadPool：可缓存的线程池。
这三种线程池都是由Executors（线程池的工具类），对ThreadPoolExecutor进行包装，从而创建的。
注：实际应用中，一般不使用这三种线程池，存在风险。
代码：
javase/src/code16/test10/Test02.java：
package code16.test10;
import java.util.concurrent.*;
public class Test02 {
public static void main(String[] args) {
// 创建任务
Task task1 = new Task();
Task task2 = new Task();
Task task3 = new Task();
// 1. ThreadPoolExecutor
// ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
// 10, // 核心线程数
// 25, // 最大线程数
// 10, // 空闲线程存活时间
// TimeUnit.SECONDS, // keepAliveTime的单位，这里为秒
// new LinkedBlockingQueue<>(), // 任务队列
// new CustomThreadFactory(), // 线程工厂
// // Executors.defaultThreadFactory() // 默认使用的线程工厂
// // new ThreadPoolExecutor.AbortPolicy() // 默认使用的任务拒绝策略，这里为直接抛出RejectedExecutionException
// );
// 2. FixedThreadPool
// ExecutorService fixedThreadPool = new ThreadPoolExecutor(
// 10, 10, // 核心线程数和最大线程数相同，即线程池中的线程都是核心线程。由于核心线程在线程池关闭之前不会被销毁，因此线程数量固定
// 0L, TimeUnit.MILLISECONDS, // 空闲线程存活时间为0ms
// new LinkedBlockingQueue<>() // 任务队列使用的是LinkedBlockingQueue，该队列的容量为Integer.MAX_VALUE，堆积大量请求时，会导致资源耗尽，从而报错OOM（OutOfMemoryError）
// );
// ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);
// 3. SingleThreadExecutor
// ExecutorService singleThreadPool = new Executors.FinalizableDelegatedExecutorService(
// new ThreadPoolExecutor(
// 1, 1, // 核心线程数和最大线程数都为1
// 0L, TimeUnit.MILLISECONDS, // 空闲线程存活时间为0ms
// new LinkedBlockingQueue<>()
// )
// );
// ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
// 4. CachedThreadPool
// ExecutorService CachedThreadPool =new ThreadPoolExecutor(
// 0, Integer.MAX_VALUE, // 核心线程数为0，最大线程数为Integer.MAX_VALUE（创建大量线程时会导致OOM），即线程池中都是非核心线程。由于非核心线程在没有执行任务时，超过空闲线程存活时间就会被清理，因此是可缓存的
// 60L, TimeUnit.SECONDS, // 空闲线程存活时间为60s
// new SynchronousQueue<>() // 任务队列使用的是SynchronousQueue，该队列是一个同步队列
// );
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 提交任务
cachedThreadPool.execute(task1);
cachedThreadPool.execute(task2);
cachedThreadPool.execute(task3);
// 关闭线程池
cachedThreadPool.shutdown();
}
}
class CustomThreadFactory implements ThreadFactory {
private static int count = 1;
@Override
public Thread newThread(Runnable r) { // r代表新线程要执行的任务
Thread thread = new Thread(r);
thread.setName("线程" + count++ + "号");
return thread;
}
}
java执行结果：
自行测试。
4. 任务 4.1. 简介任务的类型有两种：
- Runnable：实现了Runnable接口的类，在重写的run方法中定义任务内容，run方法无返回值。
- Callable：实现了Callable接口的类，在重写的call方法中定义任务内容，call方法有返回值。
向线程池提交任务，由线程池中的线程来执行这些任务。
提交任务的两种方法：
-
execute：
- 提交一个Runnable类型的任务（实现了Runnable接口的类）。
- 返回值类型为void，无任务执行结果。
- 当任务被拒绝时，抛出RejectedExecutionException异常。
-
submit：
- 提交一个Callable类型（实现了Callable接口的类）或Runnable类型的任务。
- 返回值类型为Future，封装了任务执行结果。
实际开发中一般都使用submit提交任务。
代码：
javase/src/code16/test10/Test03.java：
package code16.test10;
import java.util.concurrent.*;
public class Test03 {
public static void main(String[] args) {
// test01();
test02();
}
public static void test01() {
RunnableTask runnableTask = new RunnableTask();
CallableTask callableTask = new CallableTask();
ExecutorService threadPool = Executors.newSingleThreadExecutor();
// 1. execute
// threadPool.execute(runnableTask);
// threadPool.shutdown();
// 2. submit
// 提交Runnable任务
// Future future = threadPool.submit(runnableTask); // 泛型是任务执行结果的类型
// 提交Runnable任务并指定执行结果
// Future future = threadPool.submit(runnableTask, "任务完成");
// 提交Callable任务
Future future = threadPool.submit(callableTask);
try {
// Object result = future.get(); // null
// String result = future.get(); // 任务完成
// Integer result = future.get(); // 2
Integer result = future.get(3, TimeUnit.SECONDS); // 设置获取任务执行结果的超时时间，超过该时间后，线程就不再继续等待，直接抛出TimeoutException异常
System.out.println(result);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
public static void test02() {
ExecutorService threadPool = Executors.newFixedThreadPool(5);
CompletionService completionService = new ExecutorCompletionService<>(threadPool);
for (int i = 5; i >= 1; i--) {
int timeout = i; // 执行时间
completionService.submit(() -> {
System.out.println("执行顺序：" + timeout);
Thread.sleep(timeout * 100L);
return timeout;
});
}
for (int i = 5; i >= 1; i--) {
Future future;
try {
future = completionService.take();
System.out.println("返回顺序：" + future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
}
}
class RunnableTask implements Runnable {
// 定义任务内容
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
class CallableTask implements Callable { // 泛型是任务执行结果的类型
@Override
public Integer call() throws Exception {
Thread.sleep(5000); // 模拟任务执行时间长的情况
return 1 + 1;
}
}
java执行结果：
自行测试。
4.3. 取消任务通过Future对象的cancel方法取消任务。
任务状态：
- 任务未执行：submit—>进入任务队列（入队）。
- 任务正在执行：出队—>任务完成。
- 任务完成。
cancel只能取消正在执行的任务。
代码：
javase/src/code16/test10/Test04.java：
package code16.test10;
import java.util.concurrent.*;
public class Test04 {
public static void main(String[] args) {
CallableTask2 callableTask = new CallableTask2();
ExecutorService threadPool = Executors.newSingleThreadExecutor();
// Future future1 = threadPool.submit(callableTask);
Future future2 = threadPool.submit(callableTask);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任务是否完成：" + future2.isDone());
boolean cancel = future2.cancel(true); // 只会影响正在执行的任务，而不会影响未执行和已完成的任务
System.out.println("任务是否取消成功：" + cancel);
// System.out.println("任务是否取消成功：" + future2.isCancelled());
try {
System.out.println(future2.get()); // 取消任务后，获取任务结果会抛出异常CancellationException
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
}
class CallableTask2 implements Callable {
@Override
public Integer call() throws Exception {
int count = 0;
while (!Thread.interrupted()) { // 线程被中断时退出循环
count++;
}
System.out.println(count);
return 1 + 1;
}
}
java执行结果：
自行测试。
4.4. 任务拒绝策略任务被拒绝的时机：
- 线程池关闭。
- 线程池无空闲线程，且任务队列已满。
任务拒绝策略：指定线程池拒绝提交的任务后，执行的操作。
| 任务拒绝策略 | 描述 |
|---|---|
| AborPolicy | 默认策略（最常用），抛出RejectedException异常 |
| DiscardPolicy | 直接丢弃任务 |
| CallerRunsPolicy | 使用调用者线程再执行被拒绝的任务 |
| DiscardOldestPolicy | 丢弃处于任务队列头部的任务，并在队尾添加被拒绝的任务 |
代码：
javase/src/code16/test10/Test05.java：
package code16.test10;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class Test05 {
public static void main(String[] args) {
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
1, 1,
0, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1),
// new ThreadPoolExecutor.AbortPolicy() // 默认策略（最常用），抛出RejectedException异常
// new ThreadPoolExecutor.DiscardPolicy() // 直接丢弃任务
// new ThreadPoolExecutor.CallerRunsPolicy() // 使用调用者线程再执行被拒绝的任务
new ThreadPoolExecutor.DiscardOldestPolicy() // 丢弃处于任务队列头部的任务，并在队尾添加被拒绝的任务
);
try {
threadPool.execute(new RunnableTask2(1));
threadPool.execute(new RunnableTask2(2));
// 线程池无空闲线程，且任务队列已满
threadPool.execute(new RunnableTask2(3));
} catch (RejectedExecutionException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
}
class RunnableTask2 implements Runnable {
private int index; // 任务编号
public RunnableTask2() {
}
public RunnableTask2(int index) {
this.index = index;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "：" + index);
}
}
java执行结果：
自行测试。
4.5. 执行特殊任务代码：
javase/src/code16/test10/Test06.java：
package code16.test10;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
public class Test06 {
public static void main(String[] args) {
// test01();
// test02();
test03();
}
public static void test01() {
List tasks = new ArrayList<>();
for (int i = 1; i < 10; i++) {
tasks.add(new CallableTask3(i));
}
ExecutorService threadPool = Executors.newFixedThreadPool(5);
try {
// List> futures = threadPool.invokeAll(tasks);
// List> futures = threadPool.invokeAll(tasks, 1, TimeUnit.SECONDS); // 超时时间为1秒，超时后，未执行完的任务将被取消
// for (Future future : futures) {
// System.out.println(future.get());
// }
Integer result = threadPool.invokeAny(tasks);
System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
public static void test02() {
// ScheduledExecutorService scheduledThreadPool = new ScheduledThreadPoolExecutor(5);
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5); // 指定核心线程数为5
System.out.println(LocalDateTime.now());
// scheduledThreadPool.schedule(new RunnableTask3(), 3, TimeUnit.SECONDS);
ScheduledFuture future = scheduledThreadPool.schedule(new CallableTask4(), 3, TimeUnit.SECONDS);
try {
System.out.println(future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
scheduledThreadPool.shutdown();
}
// scheduledThreadPool.shutdown();
}
public static void test03() {
RunnableTask4 task = new RunnableTask4();
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
System.out.println(LocalDateTime.now());
// ScheduledFuture future = scheduledThreadPool.scheduleAtFixedRate(task, 1, 1, TimeUnit.SECONDS);// 参数从左至右分别代表：任务、延时时间、固定时间、时间单位
ScheduledFuture future = scheduledThreadPool.scheduleWithFixedDelay(task, 1, 1, TimeUnit.SECONDS);
try {
// 使当前线程休眠5S
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
future.cancel(false);
scheduledThreadPool.shutdown();
}
}
class CallableTask3 implements Callable {
private int index; // 任务编号
public CallableTask3() {
}
public CallableTask3(int index) {
this.index = index;
}
@Override
public Integer call() throws Exception {
Thread.sleep(1000);
return index;
}
}
class RunnableTask3 implements Runnable {
@Override
public void run() {
System.out.println(LocalDateTime.now());
}
}
class CallableTask4 implements Callable {
@Override
public String call() throws Exception {
return LocalDateTime.now().toString();
}
}
class RunnableTask4 implements Runnable {
@Override
public void run() {
System.out.println(LocalDateTime.now());
try {
Thread.sleep(1000); // 模拟任务执行时长
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
java执行结果：
自行测试。
5. 关闭线程池 5.1. 简介通过线程池的shutdown或shutdownNow方法，关闭线程池。
关闭线程池后：
-
shutdown（常用）：
-
不再接收新任务，并拒绝后面的任务。
-
继续执行完任务队列中的任务。
-
-
shutdownNow（一般不用）：
- 不再接收新任务，并拒绝后面的任务。
- 中断所有正在执行的任务（通过interrupt和interrupted方法），并返回任务队列中的任务。
代码：
javase/src/code16/test10/Test07.java：
package code16.test10;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class Test07 {
public static void main(String[] args) {
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
1, 1,
0, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1)
);
try {
threadPool.execute(new RunnableTask5(1));
threadPool.execute(new RunnableTask5(2));
} catch (RejectedExecutionException e) {
e.printStackTrace();
} finally {
// threadPool.shutdown(); // 继续执行完任务队列中的任务
List tasks = threadPool.shutdownNow(); // 尝试中断所有正在执行的任务（通过interrupt和interrupted方法），并返回任务队列中的所有任务
for (Runnable task : tasks) {
task.run();
}
// 关闭线程池后再提交任务
threadPool.execute(new RunnableTask5(3)); // 不再接收新任务，并拒绝后面的任务
}
}
}
class RunnableTask5 implements Runnable {
private int index; // 任务编号
public RunnableTask5() {
}
public RunnableTask5(int index) {
this.index = index;
}
@Override
public void run() {
if (Thread.interrupted()) {
System.out.println(Thread.currentThread().getName() + "被中断");
return;
}
System.out.println(Thread.currentThread().getName() + "：" + index);
}
}
java执行结果：
自行测试。
5.3. shutdown与shutdownNow的区别| shutdown | shutdownNow | |
|---|---|---|
| 立即关闭线程池 | 不是 | 是 |
| 不再接收新任务 | 是 | 是 |
| 继续执行完任务队列中的任务 | 是 | 不是 |
| 返回任务队列中的任务 | 不是 | 是 |
| 线程池状态（后面会讲） | SHUTDOWN | STOP |
- RUNNING：线程池正在运行。可以接收新任务，并且也能处理任务队列中的任务
- SHUTDOWN：线程池准备关闭。不接受新任务，但可以处理任务队列中的任务。
- STOP：线程池准备关闭。不接受新任务，也不处理任务队列中的任务，正在执行的任务也被中断。
- TIDYNG：线程池正在关闭。所有任务已终止，线程池中的线程数量为0。
- TERMINATED：线程池彻底关闭。
ThreadPoolExecutor提供了监控线程池的方法：
| 方法 | 返回值 |
|---|---|
| getActiveCount() | 正在工作的线程数 |
| getPoolSize() | 当前存在的线程数 |
| getLargestPoolSize() | 历史最大的线程数 |
| getTaskCount() | 已提交的任务数 |
| getCompletedTaskCount() | 已完成的任务数 |
| getQueue() | 任务队列 |
代码：
javase/src/code16/test10/MonitorThreadPool.java：
package code16.test10;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class MonitorThreadPool extends ThreadPoolExecutor {
public MonitorThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
monitor();
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
monitor();
}
@Override
protected void terminated() {
monitor();
}
public void monitor() {
System.out.print("正在工作的线程数：" + getActiveCount() + "t");
System.out.print("当前存在的线程数：" + getPoolSize() + "t");
System.out.print("历史存在的最大线程数：" + getLargestPoolSize() + "t");
System.out.print("已提交的任务数：" + getTaskCount() + "t");
System.out.print("已完成的任务数：" + getCompletedTaskCount() + "t");
System.out.println("队列中的任务数：" + getQueue().size());
}
}
javase/src/code16/test10/Test08.java：
package code16.test10;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public class Test08 {
public static void main(String[] args) {
MonitorThreadPool threadPool = new MonitorThreadPool(1, 3, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2));
try {
for (int i = 5; i > 0; i--) {
threadPool.submit(new RunnableTask6(i));
Thread.sleep(500);
}
// 使主线程休眠6秒钟，在关闭线程池之前获取一次线程池情况
Thread.sleep(6000);
threadPool.monitor();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
}
class RunnableTask6 implements Runnable {
private int timeout;
public RunnableTask6(int timeout) {
this.timeout = timeout;
}
@Override
public void run() {
try {
Thread.sleep(timeout * 1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
java执行结果：
