首先分析下为什么需要使用线程池?
假如不使用线程池,可能会造成线程数量过大,程序崩溃。因为线程也需要占用内存与CPU资源。
如果是你,会如何设计线程池?
最简单的,我们会使用一个容器,存放一定数量的线程,超过数量则不创建。那么,我们来看看jdk是怎么处理的。
线程池构造函数分析
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
1.corePoolSize:核心线程数
2.maximumPoolSize:最大线程数
3.keepAliveTime和unit:超过核心线程数的线程的空闲存活时间,后面是时间的单位
4.workQueue:当前线程数达到核心线程数之后,会通过workQueue接受任务进入等待
5.threadFactory:创建线程的工厂
6.handler:当workQueue满了并且已经达到最大线程数的拒绝策略
异步任务进入线程池执行流程
线程池执行任务核心源码分析
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 获取线程池的状态
int c = ctl.get();
// 判断当前线程数是否小于核心线程数
if (workerCountOf(c) < corePoolSize) {
// 尝试加入工作任务
if (addWorker(command, true))
return;
c = ctl.get();
}
// 进入等待队列
if (isRunning(c) && workQueue.offer(command)) {
// 双重保证该任务是否需要移除或者加入工作任务
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
// 执行拒绝策略
reject(command);
}
线程池如何保证核心线程数不销毁
当线程池执行任务时,会调用getTask获取需要执行的任务,而workQueue.take()就会阻塞住,直到队列中有任务到达。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
// 阻塞等待workQueue进入任务
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
如何使用线程池
1.Executors.newFixedThreadPool (创建固定线程池)
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(1);
for (int i = 0; i < 3; i++) {
executorService.submit(() -> {
try {
Thread.sleep(3000);
System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639356608206
当前线程:pool-1-thread-1,当前时间戳:1639356611215
当前线程:pool-1-thread-1,当前时间戳:1639356614219
构造函数分析:
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
线程池的核心线程数和最大线程数都为指定线程数,无线程增长,内部用LinkedBlockingQueue无界阻塞队列接受任务,可能会造成任务一直在阻塞中,无法执行。
2.Executors.newSingleThreadExecutor (创建单个线程)
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 3; i++) {
executorService.submit(() -> {
try {
Thread.sleep(3000);
System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639367551803
当前线程:pool-1-thread-1,当前时间戳:1639367554805
当前线程:pool-1-thread-1,当前时间戳:1639367557807
构造函数分析:
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
核心线程数和最大线程数都为1,无线程增长。内部使用LinkedBlockingQueue无界阻塞队列接受任务,可能会造成任务一直在阻塞中,无法执行。
3.Executors.newCachedThreadPool (创建可缓存使用的线程池)
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 3; i++) {
executorService.submit(() -> {
try {
Thread.sleep(3000);
System.out.println("当前线程:" + Thread.currentThread().getId() + ",当前时间戳:" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
构造函数分析:
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
核心线程数为0,最大线程数为最大整数,存活时间60秒,该线程池可能会一直创建线程,导致内存溢出。
4.newSingleThreadScheduledExecutor (创建单个定时任务线程池)
public static void main(String[] args) {
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
// 首次0延时执行,之后每三秒执行一次
executorService.scheduleAtFixedRate(() -> {
System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
}, 0, 3, TimeUnit.SECONDS);
}
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639369721662
当前线程:pool-1-thread-1,当前时间戳:1639369724667
当前线程:pool-1-thread-1,当前时间戳:1639369727670
当前线程:pool-1-thread-1,当前时间戳:1639369730669
构造函数分析
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
底层通过DelegatedScheduledExecutorService实现,核心线程数为1,最大线程数为最大整数,通过DelayedWorkQueue延时队列实现定时调度功能。
线程池有哪些拒绝策略
1.ThreadPoolExecutor.AbortPolicy (抛出拒绝异常)
public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(1, 3,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < 8; i++) {
executorService.submit(() -> {
System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
});
}
}
结果打印:
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-1,当前时间戳:1639399896121
当前线程:pool-1-thread-3,当前时间戳:1639399896121
当前线程:pool-1-thread-2,当前时间戳:1639399896121
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@58372a00 rejected from java.util.concurrent.ThreadPoolExecutor@4dd8dc3[Running, pool size = 3, active threads = 3, queued tasks = 2, completed tasks = 0]
当达到最大线程数并且队列满了之后,就会调用拒绝策略,抛出异常,符合预期。
2.ThreadPoolExecutor.CallerRunsPolicy(使用调用者线程执行)
public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(1, 3,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.CallerRunsPolicy());
for (int i = 0; i < 8; i++) {
executorService.submit(() -> {
System.out.println("当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
});
}
}
结果打印:
当前线程:main,当前时间戳:1639400060907
当前线程:pool-1-thread-3,当前时间戳:1639400060907
当前线程:pool-1-thread-1,当前时间戳:1639400060907
当前线程:pool-1-thread-2,当前时间戳:1639400060907
当前线程:pool-1-thread-3,当前时间戳:1639400060908
当前线程:pool-1-thread-1,当前时间戳:1639400060908
当前线程:main,当前时间戳:1639400060907
当前线程:pool-1-thread-2,当前时间戳:1639400060908
当达到最大线程数并且队列满了之后,就会使用调用者线程main执行。
3.ThreadPoolExecutor.DiscardPolicy (直接丢弃任务)
public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(1, 3,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.DiscardPolicy());
for (int i = 0; i < 8; i++) {
int no = i;
executorService.submit(() -> {
System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
});
}
}
结果打印:
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400376316
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400376316
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400376316
当前循环:2当前线程:pool-1-thread-2,当前时间戳:1639400376316
当前循环:1当前线程:pool-1-thread-1,当前时间戳:1639400376316
当达到最大线程数并且队列满了之后,直接丢弃之后的任务,靠前的任务能保证执行。
4.ThreadPoolExecutor.DiscardOldestPolicy (丢弃最早的任务)
public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(1, 3,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
new ThreadPoolExecutor.DiscardOldestPolicy());
for (int i = 0; i < 8; i++) {
int no = i;
executorService.submit(() -> {
System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
});
}
}
结果打印:
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400217828
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400217828
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400217828
当前循环:7当前线程:pool-1-thread-2,当前时间戳:1639400217828
当前循环:6当前线程:pool-1-thread-1,当前时间戳:1639400217828
当达到最大线程数并且队列满了之后,直接丢弃最早的任务,靠后的任务能保证执行。
5.自定义拒绝策略
public static void main(String[] args) {
ExecutorService executorService = new ThreadPoolExecutor(1, 3,
10, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(2),
(r, executor) -> {
System.out.println("执行自定义拒绝策略");
});
for (int i = 0; i < 8; i++) {
int no = i;
executorService.submit(() -> {
System.out.println("当前循环:" + no + "当前线程:" + Thread.currentThread().getName() + ",当前时间戳:" + System.currentTimeMillis());
});
}
}
结果打印:
执行自定义拒绝策略
执行自定义拒绝策略
执行自定义拒绝策略
当前循环:0当前线程:pool-1-thread-1,当前时间戳:1639400497694
当前循环:4当前线程:pool-1-thread-3,当前时间戳:1639400497694
当前循环:3当前线程:pool-1-thread-2,当前时间戳:1639400497694
当前循环:1当前线程:pool-1-thread-1,当前时间戳:1639400497695
当前循环:2当前线程:pool-1-thread-3,当前时间戳:1639400497695