学习线程池 掌握治理线程的法宝
1.为何须要线程池
在当今计算机的CPU计算速度很是快的状况下，为了可以充分利用CPU性能提升程序运行效率咱们在程序中使用了线程。可是在高并发状况下会频繁的建立和销毁线程，这样就变相的阻碍了程序的执行速度，因此为了管理线程资源和减小线程建立以及销毁的性能消耗就引入了线程池。数据库
2.什么场景下适合使用线程池
- 当服务器接收到大量任务时，若是使用线程池能够大量减小线程的建立与销毁次数，从而提高程序执行效率
- 在实际开发中，若是须要建立5个以上的线程，那么就可使用线程池来管理
3.线程池参数介绍以及特色
3.1 corePoolSize和maxPoolSize
corePoolSize：线程池在建立完时，里面并无线程，只有当任务到来时再去建立线程。缓存
maxPoolSize：线程池可能会在核心线程数的基础上额外增长一些线程，可是线程数量的上限是maxPoolSize。好比第一天执行的任务很是多，次日执行的任务很是少，可是有了maxPoolSize参数，就能够加强任务处理的灵活性。bash
3.2 添加线程的规则
- 当线程数量小于
corePoolSize即便线程没有在执行任务，也会建立新的线程。 - 若是线程数量等于（或大于）
corePoolSize，但小于maxPoolSize则将任务放入队列。 - 若是队列已满，而且线程数小于
maxPoolSize，则建立新的线程运行任务。 - 若是队列已满，而且线程数大于或等于
maxPoolSize，则拒绝该任务。
执行流程：服务器
3.3 增减线程的特色
- 将
corePoolSize和maxPoolSize设置为相同的值，那么就会建立固定大小的线程池。 - 线程池但愿保持更少的线程数，而且只有在负载变得很大时才会增长它。
- 若是将线程池的
maxPoolSize参数设置为很大的值，例如Integer.MAX_VALUE，能够容许线程池容纳任意数量的并发任务。 - 只有在队列满了的时候才会去建立大于
corePoolSize的线程，因此若是使用了无界队列（如：LinkedBlockingQueue）就不会建立到超过corePoolSize的线程数。
3.4 keepAliveTime
若是线程池当前的线程数大于corePoolSize,那么若是多余的线程的空闲时间大于keepAliveTime，它们就会被终止。网络
keepAliveTime参数的使用能够减小线程数过多冗余时的资源消耗。并发
3.5 threadFactory
新的线程由ThreadFactory建立，默认使用Executors.defaultThreadFactory()，建立出来的线程都在同一个线程组，拥有一样的NORM_PRIORITY优先级而且都不是守护线程。若是本身指定ThreadFactory，那么就能够改变线程名、线程组、优先级、是不是守护线程等。一般状况下直接使用defaultThreadFactory就行。ide
3.6 workQueue
- 直接交接（
SynchronousQueue）：任务很少时，只须要用队列进行简单的任务中转，这种队列没法存储任务，在使用这种队列时，须要将maxPoolSize设置的大一点。 - 无界队列（
LinkedBlockingQueue）：若是使用无界队列看成workQueue，将maxQueue设置的多大都没有用，使用无界队列的优势是能够防止流量突增，缺点是若是处理任务的速度跟不上提交任务的速度，这样就会致使无界队列中的任务愈来愈多，从而致使OOM异常。 - 有界队列（
ArrayBlockingQueue）：使用有界队列能够设置队列大小，让线程池的maxPoolSize有意义。
4.线程池应该手动建立仍是自动建立
手动建立更好，由于这样可让咱们更加了解线程池的运行规则，避免资源耗尽的风险。函数
4.1 直接调用JDK封装好的线程池会带来的问题
- newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
复制代码
newFixedThreadPool线程池经过传入相同的corePoolSize和maxPoolSize能够保证线程数量固定，0L的keepAliveTime表示时刻被销毁，workQueue使用的是无界队列。这样潜在的问题就是当处理任务的速度赶不上任务提交的速度的时候，就可能会让大量任务堆积在workQueue中，从而引起OOM异常。高并发
4.2 演示newFixedThreadPool内存溢出问题
/**
* 演示newFixedThreadPool线程池OOM问题
*/
public class FixedThreadPoolOOM {
private static ExecutorService executorService = Executors.newFixedThreadPool(1);
public static void main(String[] args) {
for (int i = 0; i < Integer.MAX_VALUE; i++) {
executorService.execute(new SubThread());
}
}
}
class SubThread implements Runnable {
@Override
public void run() {
try {
//延长任务时间
Thread.sleep(1000000000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
复制代码
更改JVM参数工具
运行结果
4.3 newSingleThreadExecutor
使用线程池打印线程名
public class SingleThreadExecutor {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 1000; i++) {
executorService.execute(new Task());
}
}
}
复制代码
newSingleThreadExecutor源码
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
复制代码
从源码能够看出newSingleThreadExecutor和newFixedThreadPool基本相似，不一样的只是corePoolSize和maxPoolSize的值，因此newSingleThreadExecutor也存在内存溢出问题。
4.4 newCachedThreadPool
newCachedThreadPool也被称为可缓存线程池，它是一个
无界线程池，具备
自动回收多余线程的功能。
public class CachedThreadPool {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 1000; i++) {
executorService.execute(new Task());
}
}
}
复制代码
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
复制代码
newCachedThreadPool的maxPoolSize设置的值为Integer.MAX_VALUE，因此可能会致使线程被无限建立，最终致使OOM异常。
4.5 newScheduledThreadPool
该线程池支持周期性任务的执行
public class ScheduledThreadPoolTest {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService =
Executors.newScheduledThreadPool(10);
// scheduledExecutorService.schedule(new Task(), 5, TimeUnit.SECONDS);
scheduledExecutorService.scheduleAtFixedRate(new Task(), 1, 3, TimeUnit.SECONDS);
}
}
复制代码
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
复制代码
4.6 正确的建立线程池的方法
根据业务场景不一样，本身设置线程池参数，例如内存有多大，本身取线程名子等。
4.7 线程池里的线程数量设置多少比较合适？
- CPU密集型（加密、计算hash等）：最佳线程数设置为CPU核心数的1——2倍。
- 耗时I/O型（读写数据库、文件、网络读写等）：最佳线程数通常会大于CPU核心数不少倍，以JVM监控显示繁忙状况为依据，保证线程空闲能够衔接上。参考Brain Goezt推荐的计算方法：
线程数=CPU核心数 × (1+平均等待时间/平均工做时间)
5.对比线程池的特色
- FixedThreadPool：经过手动传入
corePoolSize和maxPoolSize，以固定的线程数来执行任务 - SingleThreadExecutor：
corePoolSize和maxPoolSize默认都是1，全程只以1条线程执行任务 - CachedThreadPool：它没有须要维护的核心线程数，每当须要线程的时候就进行建立，由于它的线程存活时间是60秒，因此它也凭借着这个参数实现了自动回收的功能。
- ScheduledThreadPool：这个线程池能够执行定时任务，
corePoolSize是经过手动传入的，它的maxPoolSize为Integer.MAX_VALUE，而且具备自动回收线程的功能。
5.1 为何FixedThreadPool和SingleThreadExecutor的Queue是LinkedBlockingQueue？
由于这两个线程池的核心线程数和最大线程数都是相同的，也就没法预估任务量，因此须要在自身进行改进，就使用了无界队列。
5.2 为何CachedThreadPool使用的Queue是SynchronousQueue？
由于缓存线程池的最大线程数是“无上限”的，每当任务来的时候直接建立线程进行执行就行了，因此不须要使用队列来存储任务。这样避免了使用队列进行任务的中转，提升了执行效率。
5.3 为何ScheduledThreadPool使用延迟队列DelayedWorkQueue？
由于ScheduledThreadPool是延迟任务线程池，因此使用延迟队列有利于对执行任务的时间作延迟。
5.4 JDK1.8中加入的workStealingPool
workStealingPool适用于执行产生子任务的环境，例如进行二叉树的遍历。workStealingPool具备窃取能力。- 使用时最好不要加锁，并且不保证执行顺序。
6.中止线程池的正确方法
- shutdown：调用了
shutdown()方法不必定会当即中止，这个方法仅仅是初始整个关闭过程。由于线程池中的线程有可能正在运行，而且队列中也有待处理的任务，不可能说停就停。因此每当调用该方法时，线程池会把正在执行的任务和队列中等待的任务都执行完毕再关闭，而且在此期间若是接收到新的任务会被拒绝。
/**
* 演示关闭线程池
*/
public class ShutDown {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 0; i < 1000; i++) {
executorService.execute(new ShutDownTask());
}
Thread.sleep(1500);
executorService.shutdown();
//再次提交任务
executorService.execute(new ShutDownTask());
}
}
class ShutDownTask implements Runnable {
@Override
public void run() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName());
}
}
复制代码
- isShutdown：能够用于判断线程池是否被shutdown了
/**
* 演示关闭线程池
*/
public class ShutDown {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 0; i < 1000; i++) {
executorService.execute(new ShutDownTask());
}
Thread.sleep(1500);
System.out.println(executorService.isShutdown());
executorService.shutdown();
System.out.println(executorService.isShutdown());
//再次提交任务
// executorService.execute(new ShutDownTask());
}
}
class ShutDownTask implements Runnable {
@Override
public void run() {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName());
}
}
复制代码
- isTerminated：能够判断线程是否被彻底终止了
/**
* 演示关闭线程池
*/
public class ShutDown {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 0; i < 1000; i++) {
executorService.execute(new ShutDownTask());
}
Thread.sleep(1500);
System.out.println(executorService.isShutdown());
executorService.shutdown();
System.out.println(executorService.isShutdown());
System.out.println(executorService.isTerminated());
//再次提交任务
// executorService.execute(new ShutDownTask());
}
}
复制代码
isTerminated方法的地方休眠10s
- awaitTermination：传入等待时间，等待时间达到时判断是否中止了，主要用于检测。
//在3s后判断线程池是否被终止，返回boolean值
System.out.println(executorService.awaitTermination(3L, TimeUnit.SECONDS));
复制代码
- shutdownNow：调用了这个方法时，线程池会当即终止，并返回没有被处理完的任务。若是须要继续执行这里的任务能够再次让线程池执行这些返回的任务。
7.任务太多，怎么拒绝？
7.1 拒绝的时机
- 当
Executor关闭时，新提交的任务会被拒绝。 - 以及
Executor对最大线程数和工做队列容量使用有限边界而且已经饱和时。
7.2 拒绝策略
- AbortPolicy（中断策略）：直接抛出异常进行拒绝
- DiscardPolicy（丢弃策略）：不会获得通知，默默的抛弃掉任务
- DiscardOldestPolicy（丢弃最老的）：因为队列中存储了不少任务，这个策略会丢弃在队列中存在时间最久的任务。
- CallerRunsPolicy：好比主线程给线程池提交任务，可是线程池已经满了，在这种策略下会让提交任务的线程去执行。
总结：第四种拒绝策略相对于前三种更加“机智”一些，能够避免前面三种策略产生的损失。在第四种策略下能够下降提交的速度，达到负反馈的效果。
8.使用钩子为线程池加点料（可用于日志记录）
/**
* 演示每一个任务执行的先后放钩子函数
*/
public class PauseableThreadPool extends ThreadPoolExecutor {
private boolean isPaused;
private final ReentrantLock lock = new ReentrantLock();
private Condition unPaused = lock.newCondition();
public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
}
public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
super.beforeExecute(t, r);
lock.lock();
try {
while (isPaused) {
unPaused.await();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
private void pause() {
lock.lock();
try {
isPaused = true;
} finally {
lock.unlock();
}
}
public void resume() {
lock.lock();
try {
isPaused = false;
//唤醒所有
unPaused.signalAll();
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
PauseableThreadPool pauseableThreadPool = new PauseableThreadPool(10, 20, 10L,
TimeUnit.SECONDS, new LinkedBlockingQueue<>());
Runnable runnable = new Runnable() {
@Override
public void run() {
System.out.println("我被执行");
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
for (int i = 0; i < 10000; i++) {
pauseableThreadPool.execute(runnable);
}
Thread.sleep(1500);
pauseableThreadPool.pause();
System.out.println("线程池被暂停了");
Thread.sleep(1500);
pauseableThreadPool.resume();
System.out.println("线程池被恢复了");
}
}
复制代码
9.线程池实现原理
9.1 线程池组成部分
- 线程池管理器
- 工做线程
- 任务队列
- 任务
9.2 Executor家族
- Executor：它是一个顶层接口，其余接口以及类都i继承或实现于它，包含如下方法：
- void execute(Runnable command);
- ExecutorService：它继承于
Executor，是Executor的子接口，在接口内部增长了一些新的方法，例如第6小节讲到的几个方法 - Executors：这个类是一个工具类，里面包含一些建立线程池的方法
9.3 线程池实现任务复用的原理
利用相同线程执行不一样任务
- 源码分析
public void execute(Runnable command) {
// 判断任务是否为空，为空就抛出异常
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// 若是当前线程数小于核心线程数，就增长Worker
if (workerCountOf(c) < corePoolSize) {
// command就是任务，点击addWorker方法
// 第二个参数用于判断当前线程数是否小于核心线程数
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);
}
复制代码
由于要查看的是Worker因此进入到addWorker()方法后点击Worker类查看runWorker()方法
w = new Worker(firstTask);
复制代码
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
复制代码
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 获取到任务
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 只要任务不为空或者可以获取到任务就执行下面的方法
while (task != null || (task = getTask()) != null) {
w.lock();
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
// task是一个Runnable类型，调用run()方法就是运行线程
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
复制代码
总结：核心原理就是获取到task,若是task不为空就调用run()方法，这样就实现了线程的复用，达到让相同的线程执行不一样任务的目的。
10.线程池状态
- RUNNING：接受新任务并处理排队任务
- SHUTDOWN：不接受新的任务可是处理排队任务
- STOP：不接受新的任务，也不处理排队的任务，并中断正在执行的任务，就是调用
shutdownNow()带来的效果 - TIDYING：中文意思是整洁，意思就是说任务都已经终止，
workerCount为零时，线程会转换到TIDYING状态，并将运行terminate()钩子方法 - TERMINATED：
terminate()运行完成
11.使用线程池的注意点
- 避免任务的堆积（堆积容易产生内存溢出）
- 避免线程数过多增长（缓存线程池会致使线程数过分增长）
- 排查线程泄漏（线程已经执行完毕却没法被回收）
相关文章
- 1. 线程池 掌握治理线程的法宝
- 2. 线程池学习
- 3. 多线程-JUC学习-线程池-线程池源码分析
- 4. 线程池的学习
- 5. java线程池的学习
- 6. 多线程学习--线程池
- 7. java线程池的原理学习（二）
- 8. java线程池的原理学习（三）
- 9. java线程池的原理学习
- 10. Java线程池二：线程池原理
- 更多相关文章...
- • C# 多线程 - C#教程
- • SVG 渐变 - 线性 - SVG 教程
- • 适用于PHP初学者的学习线路和建议
- • Kotlin学习（一）基本语法
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 线程池 掌握治理线程的法宝
- 2. 线程池学习
- 3. 多线程-JUC学习-线程池-线程池源码分析
- 4. 线程池的学习
- 5. java线程池的学习
- 6. 多线程学习--线程池
- 7. java线程池的原理学习（二）
- 8. java线程池的原理学习（三）
- 9. java线程池的原理学习
- 10. Java线程池二：线程池原理