如何设计线程池

所以这里尼恩给大家做一下系统化、体系化的线程池梳理使得大家可以充分展示一下大家雄厚的

“技术肌肉”让面试官爱到

《尼恩Java面试宝典》V62版本供后面的小伙伴参考提升大家的

PDF

多线程编程是在开发过程中非常基础且非常重要的一个环节基本上任何一家软件公司或者项目中都会使用多线程。

主要有三个原因

降低资源的消耗。

降低线程创建和销毁的资源消耗。

提高响应速度线程的创建时间为T1执行时间T2销毁时间T3免去T1和T3的时间提高线程的可管理性

总之线程池是一种常用的并发编程工具它可以帮助我们更好地管理和复用线程资源提高程序的性能和稳定性。

在Java中我们可以使用java.util.concurrent包中提供的ThreadPoolExecutor类来创建和使用线程池。

ThreadPoolExecutor

的底层原理和源码大家要做到非常深入的掌握。

大家一定要深入看ThreadPoolExecutor线程池源码了解其执行过程。

另外看懂线程池执行流程和源码设计有助于提升我们多线程编程技术和解决工作中遇到的问题。

很多小伙伴给尼恩反馈

可以手撸一个简单版的线程池加强一下对执行流程的理解。

然后再深入源码去历险记。

或者说如果我们想要更好地理解线程池的工作原理那么自己动手实现一个简单的线程池是一个很好的选择。

我们将从最基本的功能开始逐步添加更多的功能和优化最终实现一个完整的线程池。

线程池是一个典型的生产者-消费者模型。

下图所示为线程池的实现原理

调用方不断向线程池中提交任务

生产者线程池中有一组线程不断地从队列中取任务消费者线程池管理一个任务队列对

异步任务进行缓冲

如果是无界的调用方不断往队列中方任务可能导致内存耗尽。

如果是有界的当队列满了之后调用方如何处理线程池中的线程个数是固定的还是动态变化的每次提交新任务是放入队列还是开新线程当没有任务的时候线程是睡眠一小段时间还是进入阻塞如果进入阻塞如何唤醒

针对问题4有3种做法

当队列为空时线程池中的线程只能睡眠一会儿然后醒来去看队列中有没有新任务到来如此不断轮询。

不使用阻塞队列但在队列外部线程池内部实现了阻塞/唤醒机制使用阻塞队列

很显然做法3最完善既避免了线程池内部自己实现阻塞/唤醒机制的麻烦也避免了做法1的睡眠/轮询带来的资源消耗和延迟。

现在来带大家手写一个简单的线程池让大家更加理解线程池的工作原理

手写一个简单线程池

首先我们需要定义一个线程池接口用来表示线程池应该具备哪些功能。

添加任务并执行关闭线程池强制关闭线程池

因此我们可以定义一个ThreadPool接口它包含三个方法execute、shutdown和shutdownNow。

import

execute方法用来添加任务并执行shutdown方法用来关闭线程池它会等待已经提交到线程池中的任务都执行完毕后再关闭shutdownNow方法用来强制关闭线程池它会立即停止所有正在执行或等待执行的任务并返回未执行的任务列表。

接下来我们需要实现一个简单的线程池类它实现了ThreadPool接口并提供了基本的功能。

为了简化代码先实现一个v1版本这是一个基础版本一个简单的实现示例。

在v1版本中我们先不考虑拒绝策略、自动调节线程资源等高级功能。

首先定义一个工作线程类

WorkerThread(BlockingQueueRunnable

taskQueue)

(!Thread.currentThread().isInterrupted()

{try

从任务队列中取出一个任务如果队列为空则阻塞等待Runnable

task

WorkerThread(taskQueue);workerThread.start();threads.add(workerThread);}}//

实现execute方法用于将任务加入到任务队列并通知工作线程来执行Overridepublic

void

IllegalStateException(ThreadPool

shutdown);}taskQueue.offer(task);}//

关闭线程池,

中断线程thread.interrupt();}}Overridepublic

ListRunnable

ArrayList();taskQueue.drainTo(remainingTasks);//

中断所有线程for

}这个版本的线程池实现了基本的添加任务并执行、关闭线程池和强制关闭线程池等功能。

它在构造方法中接收一个初始化线程池大小参数用于初始化任务队列和工作线程集合并创建一定数量的工作线程。

但是它存在一个问题任务队列没有指定容量大小是个无界队列其次只指定了初始的线程池大小应该要提供根据不同的应用场景来调整线程池的大小参数以提高性能和资源利用率。

因此线程池实现类需要实现自定义初始大小、最大大小以及核心大小的功能。

初始大小是指线程池初始化时创建的工作线程数量最大大小是指线程池能够容纳的最多的工作线程数量核心大小是指线程池在没有任务时保持存活的工作线程数量。

这三个参数需要在基本的线程池实现类中定义为成员变量并在构造方法中传入并赋值。

同时还需要在execute方法中根据这三个参数来动态地调整工作线程的数量例如

当活跃的工作线程数量小于核心大小时尝试创建并启动一个新的工作线程来执行任务当活跃的工作线程数量大于等于核心大小时将任务加入到任务队列等待空闲的工作线程来执行当任务队列已满时尝试创建并启动一个新的工作线程来执行任务当活跃的工作线程数量达到最大大小时无法再创建新的工作线程。

我们还需要在构造方法里提供一个参数queueSize用于限制队列大小。

下面我们就对类进行改造

java.util.concurrent.BlockingQueue;

import

java.util.concurrent.LinkedBlockingQueue;

import

java.util.concurrent.TimeUnit;public

class

LinkedBlockingQueue(queueSize);threads

new

WorkerThread();workerThread.start(taskQueue);threads.add(workerThread);}}Overridepublic

void

IllegalStateException(ThreadPool

shutdown);}//

IllegalStateException(执行任务失败);}}}//

void

WorkerThread();workerThread.start(taskQueue);threads.add(workerThread);//

任务放入队列taskQueue.offer(task);}省略其它代码}这一步我们在

SimpleThreadPool里新增了initialSizemaxSize

coreSize

三个变量在构造方法里传入对应三个参数同时在execute方法里当有任务进入时先判断当前线程池数量是否满足不同条件进而执行不同的处理逻辑。

第四步设计饱和拒绝策略

这个功能是为了处理当任务队列已满且无法再创建新的工作线程时也是就线程池的工作量饱和时如何处理被拒绝的任务。

直接抛出异常忽略任务阻塞当前线程等等

定义一个拒绝策略接口声明一个方法用于处理被拒绝的任务。

然后需要在基本的线程池实现类中定义一个拒绝策略成员变量并在构造方法中传入并赋值。

最后在execute方法中在无法创建新的工作线程时调用拒绝策略来处理该任务。

我们首先定义了一个RejectedExecutionHandler接口用来表示拒绝策略。

用户可以根据需要实现这个接口并在构造线程池时传入自己的拒绝策略。

public

DEFAULT_REJECT_HANDLER);}public

SimpleThreadPool(int

LinkedBlockingQueue(queueSize);threads

new

ArrayList(initialSize);this.rejectHandler

rejectHandler;//

WorkerThread(taskQueue);workerThread.start();threads.add(workerThread);}}Overridepublic

void

IllegalStateException(ThreadPool

shutdown);}//

{addWorkerThread(task);System.out.printf(创建新的线程:

thread

{addWorkerThread(task);System.out.printf(创建新的线程:

thread

{//使用拒绝策略rejectHandler.rejectedExecution(task,

this);}}}//

}这个版本的线程池在构造方法中新增了一个handler参数用来表示拒绝策略。

当任务队列已满时它会调用handler的rejectedExecution方法来处理被拒绝的任务。

第五步性能优化之自动调节线程资源

线程资源功能是为了让线程池可以根据任务的变化动态地增加或减少工作线程的数量以提高性能和资源利用率。

为了实现这个功能需要在基本的线程池实现类中定义一个空闲时长成员变量并在构造方法中传入并赋值。

同时还需要在工作线程类中定义一个空闲开始时间成员变量并在run方法中更新它。

空闲开始时间是指当工作线程从任务队列中取出一个任务后上一次取出任务的时间。

如果当前时间减去空闲开始时间大于空闲时长那么工作线程就会自动退出。

ok那么我们继续改进线程池,

DEFAULT_REJECT_HANDLER);}public

SimpleThreadPool(int

LinkedBlockingQueue(queueSize);threads

new

ArrayList(initialSize);this.rejectHandler

rejectHandler;this.keepAliveTime

keepAliveTime;//

threads);workerThread.start();threads.add(workerThread);}}//

省略其它代码

(!Thread.currentThread().isInterrupted()

{try

{task.run();System.out.printf(WorkerThread

%d,

Thread.currentThread().getId(),

task.toString());lastActiveTime

else

从线程池中移除threads.remove(this);System.out.printf(WorkerThread

exit

Thread.currentThread().getId());break;}}

catch

从线程池中移除threads.remove(this);e.printStackTrace();//

}在WorkerThread类run方法里采用taskQueue.poll方法指定等待时长这里是线程退出的关键。

如果超时未获取到任务则表明当前线程长时间未处理任务可以正常退出并从线程池里移除该线程。

拒绝策略相关类

}为了通过输出日志清晰的展现线程池中任务的运行流程新增了RunnableWrapper用于记录taskId方便日志监控。

public

(!Thread.currentThread().isInterrupted()

{try

TimeUnit.MILLISECONDS);RunnableWrapper

wrapper

{System.out.printf(WorkerThread

%s,

Thread.currentThread().getName(),

wrapper.getTaskId());task.run();lastActiveTime

else

从线程池中移除threads.remove(this);System.out.printf(WorkerThread

exit

Thread.currentThread().getName());break;}}

catch

从线程池中移除System.out.printf(WorkerThread

occur

Thread.currentThread().getName());threads.remove(this);e.printStackTrace();//

}简单线程池实现类

DEFAULT_REJECT_HANDLER);}public

SimpleThreadPool(int

LinkedBlockingQueue(queueSize);threads

new

ArrayList(initialSize);this.rejectHandler

rejectHandler;this.keepAliveTime

keepAliveTime;//

threads);workerThread.start();threads.add(workerThread);}}Overridepublic

void

IllegalStateException(ThreadPool

wrapper

{addWorkerThread(task);System.out.printf(小于核心线程数创建新的线程:

thread

taskQueue.remainingCapacity());}

else

{addWorkerThread(task);System.out.printf(队列已满,

thread

taskQueue.remainingCapacity());}

else

{rejectHandler.rejectedExecution(task,

this);}}

taskQueue.remainingCapacity());}}//

void

threads);workerThread.start();threads.add(workerThread);//

任务放入队列try

taskQueue.remainingCapacity());//

修改状态isShutdown

中断线程thread.interrupt();}}Overridepublic

ListRunnable

taskQueue.remainingCapacity());//

修改状态isShutdown

ArrayList();taskQueue.drainTo(remainingTasks);//

中断所有线程for

最后我们编写了一个测试用例SimpleThreadPoolTest

定义一个测试用例类

RunnableWrapper(i));}Thread.sleep(10_000);threadPool.shutdown();}

}这个测试用例创建了一个拥有1个初始线程、最多4个线程、核心线程数为2、队列长度为3空闲线程保留时间为2000毫秒的线程池。

然后测试用例向线程池中提交了10个简单的任务每个任务都会打印一条消息然后睡眠100毫秒再打印一条消息。

初始化线程池:

3从输出中可以看出线程池中最多有4个线程在同时运行。

当有空闲线程时新提交的任务会被立即执行否则新提交的任务会被添加到任务队列中等待执行。

ok到了这里大家能够帮助大家更好地理解上文中实现的简单线程池。

实操总结

我们从最基本的功能开始逐步添加了拒绝策略、自动调节线程资源等高级功能并对线程池进行了优化。

当然这只是一个简单的示例实际应用中的线程池可能会更加复杂和强大。

上的实操与Java标准库中提供的线程池相比仍然存在一些不足之处。

没有提供足够的构造参数和方法让用户可以更好地控制和监控线程池的行为。

没有提供足够多的拒绝策略让用户可以根据不同的场景选择不同的拒绝策略。

没有提供定时任务和周期性任务的执行功能。

没有提供足够完善的错误处理机制。

Java标准库中提供的线程池实现如ThreadPoolExecutor类则在这些方面都做得更好。

它提供了丰富的构造参数和方法让用户可以更好地控制和监控线程池的行为它提供了多种拒绝策略让用户可以根据不同的场景选择不同的拒绝策略它还提供了ScheduledThreadPoolExecutor类用来执行定时任务和周期性任务它还提供了完善的错误处理机制可以帮助用户更好地处理异常情况。

如果要深入了解Java标准库中提供的线程池实现可以参考清华大学出版社所出版的尼恩

Java高并发三部曲之二

https://www.cnblogs.com/daoqidelv/p/7043696.html

清华大学出版社《Java高并发核心编程

《峰值21WQps、亿级DAU小游戏《羊了个羊》是怎么架构的》

《2个大厂

《一文搞定SpringBoot、SLF4j、Log4j、Logback、Netty之间混乱关系史上最全》

linux