线程池
线程池的创建代码
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler);
参数的含义
corePoolSize —— 保留在池中的线程数，即使它们处于空闲状态也不会销毁，除非设置了allowCoreThreadTimeOut
maximumPoolSize —— 池中允许的最大线程数
keepAliveTime —— 当线程数大于核心数时，多余的空闲线程在终止前等待新任务的最长时间。
unit —— keepAliveTime参数的时间单位
workQueue —— 在执行任务之前用于保存任务的队列。该队列将仅保存由execute方法提交的Runnable任务。
threadFactory —— 执行者创建新线程时使用的工厂
handler —— 由于达到线程边界和队列容量而阻塞执行时使用的处理程序
任务队列
BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>();
BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(num, true);
BlockingQueue<Runnable> blockingQueue = new SynchronousQueue<>(true);
- LinkedBlockingQueue 推荐用于无界队列，此时容量大小为 Integer.MAX_VALUE。也可指定容量，用作有界队列。
- ArrayBlockingQueue 如果为true则在插入或移除时阻塞的线程的队列访问将按 FIFO 顺序处理；如果为false则未指定访问顺序。用于有界队列。
- SynchronousQueue 提交的任务不会被保存，总是会马上提交执行。如果为true，则等待线程以 FIFO 顺序竞争访问；否则顺序未指定。无参创建为 false。
任务过多时的拒绝策略
- AbortPolicy 任务不能执行或放入队列，抛出异常
- CallerRunsPolicy 线程池未关闭时，直接在execute方法的调用线程中运行被拒绝的任务
- DiscardOldestPolicy 线程池未关闭时，丢弃队列头部的未处理请求，然后提交任务运行
- DiscardPolicy 直接丢弃任务，不做任何处理
可自定义拒绝策略
如下保证任务不会丢失，但可能无法满足业务场景。
// 线程池未关闭时，在execute方法的调用线程中线程阻塞直到任务放入队列，否则在execute方法的调用线程中运行任务
RejectedExecutionHandler handler = (runnable, executor) -> {
if (!executor.isShutdown()) {
try {
executor.getQueue().put(runnable);
} catch (InterruptedException e) {
e.printStackTrace();
runnable.run();
}
} else {
runnable.run();
}
};
java线程池ThreadPoolExecutor类使用详解
线程池提交任务的流程
分 3 个步骤进行：
- 如果正在运行的线程少于 corePoolSize，请尝试使用给定命令启动一个新线程作为其第一个任务。对 addWorker 的调用以原子方式检查 runState 和 workerCount，因此通过返回 false 来防止错误警报，这些错误警报会在不应添加线程时添加线程。
- 如果一个任务可以成功放入队列，那么我们仍然需要仔细检查我们是否应该添加一个线程（因为自从上次检查后现有的线程已经终止）或者池在进入这个方法后关闭了。因此，我们重新检查状态，如果有必要，如果停止，则回滚排队，或者如果没有，则启动一个新线程。
- 如果我们不能排队任务，那么我们尝试添加一个新线程。如果它失败了，我们知道我们已经关闭或饱和，因此拒绝该任务。
JVM 异常处理
未捕获的异常会由 Thread#dispatchUncaughtException 处理
先获取 Thread#UncaughtExceptionHandler，若线程设置了 handler 则调用设置的 handler 处理
否则调用 线程的 ThreadGroup 对象 进行异常处理
ThreadGroup 中 UncaughtExceptionHandler 会找寻顶层父级，父级设置 handler 调用父级处理，否则进行统一的异常打印
详见 Thread#dispatchUncaughtException 方法，未处理的异常 JVM 会自动此方法执行异常处理逻辑。
测试代码源码如下
import java.util.concurrent.*;
public class PoolDemo {
// java线程池ThreadPoolExecutor类使用详解 https://www.cnblogs.com/dafanjoy/p/9729358.html
public static void main(String[] args) {
/*
* 未捕获的异常会由 Thread#dispatchUncaughtException 处理
* 先获取 Thread#UncaughtExceptionHandler，若线程设置了 handler 则调用设置的 handler 处理
* 否则调用 线程的 ThreadGroup 对象 进行异常处理
* ThreadGroup 中 UncaughtExceptionHandler 会找寻顶层父级，父级设置 handler 调用父级处理，否则进行统一的异常打印
*/
/*
Thread.currentThread().setUncaughtExceptionHandler(((t, e) -> {
System.err.println("出错啦");
e.printStackTrace();
}));
*/
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// ThreadFactory threadFactory = Executors.privilegedThreadFactory();
// BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<>();
BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(1, true);
// BlockingQueue<Runnable> blockingQueue = new SynchronousQueue<>(true);
// 线程池未关闭时，在execute方法的调用线程中线程阻塞直到任务放入队列，否则在execute方法的调用线程中运行任务
RejectedExecutionHandler handler = (runnable, executor) -> {
if (!executor.isShutdown()) {
try {
executor.getQueue().put(runnable);
} catch (InterruptedException e) {
e.printStackTrace();
runnable.run();
}
} else {
runnable.run();
}
};
// 任务不能执行或放入队列，抛出异常
// handler = new ThreadPoolExecutor.AbortPolicy();
// 线程池未关闭时，直接在execute方法的调用线程中运行被拒绝的任务
// handler = new ThreadPoolExecutor.CallerRunsPolicy();
// 线程池未关闭时，丢弃队列头部的未处理请求，然后提交任务运行
// handler = new ThreadPoolExecutor.DiscardOldestPolicy();
// 直接丢弃任务，不做任何处理
// handler = new ThreadPoolExecutor.DiscardPolicy();
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
1, 1, 30L, TimeUnit.SECONDS,
blockingQueue, threadFactory, handler);
threadPoolExecutor.execute(() -> {
System.out.printf("%s execute0%n", Thread.currentThread().getName());
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("submit0");
for (int i = 0; i < 2; i++) {
final int j = i + 1;
System.out.printf("submit%d%n", j);
// try {
threadPoolExecutor.execute(() -> System.out.printf("%s execute%d%n", Thread.currentThread().getName(), j));
// } catch (Exception e) {
// System.err.print("Exception in thread \"" + Thread.currentThread().getName() + "\" ");
// e.printStackTrace(System.err);
// }
}
System.out.println("关闭线程池");
threadPoolExecutor.shutdown();
}
}