什么是线程池:
线程池是一种线程的使用模式,预先将线程创建出来采用池化的方式存储起来,使用的时候直接使用即可,避免频繁的线程的创建与销毁带来的性能开销,也可以对线程的数量进行限制管理,避免创建过多的线程导致oom异常。
java中使用多线程的方式
1,继承Thread类,实现Runable/Callable接口,调用start/call方法便可以开启一个线程。
2,就是我们要说的线程池的方式了。
1的方式不用说频繁创建销毁线程带来性能开销,以及线程的数量得不到限制管理实际项目中肯定不能用该方式使用多线程。
java中的线程池Executor框架
他是java对线程池的实现,实现了将线程的执行单元和分开单元分开这种机制.该框架包括三大部分
1,任务单元即实现Runable/Callable接口的任务类(run/call方法中写的是你真正想干的事)。
2,执行单元,缓存在线程池中的线程来执行你的任务.
3,异步计算的结果,当任务是实现callable接口时候,执行方法会将返回结果封装倒Future对象中返回
前置思考:
有一个很经典的面试题问道:线程可以调用多次start方法么?答案是不能,为什么我们可以打开源码
public synchronized void start() {
/**
* This method is not invoked for the main method thread or "system"
* group threads created/set up by the VM. Any new functionality added
* to this method in the future may have to also be added to the VM.
*
* A zero status value corresponds to state "NEW".
*/
/**
他判断了线程的状态状态是0也就是NEW的状态才能执行该方法,
很显然线程调用过一次start方法后状态肯定不会为NEW而是TERMINATED,
所以多次调用会抛出异常。
至于线程的那几个状态在内部枚举类State里面有列举这里不再赘述
*/
if (threadStatus != 0)
throw new IllegalThreadStateException();
/* Notify the group that this thread is about to be started
* so that it can be added to the group's list of threads
* and the group's unstarted count can be decremented. */
group.add(this);
boolean started = false;
try {
start0();
started = true;
} finally {
try {
if (!started) {
group.threadStartFailed(this);
}
} catch (Throwable ignore) {
/* do nothing. If start0 threw a Throwable then
it will be passed up the call stack */
}
}
}好那么我们不禁要思考线程池能够完成对线程的复用,那么他是如何做到让线程执行完任务之后在那里不结束等着下一个任务来了继续去执行的呢？要我们来做的话怎么做呢?我们可以大胆的设想一下是不是像生产者消费者那样的模型就可以呢？。
使用线程池:
线程池的使用并不复杂,有两种方式去使用他:
1，Excutetors工具类(线程池工厂类？)创建一些特定的线程池,这里面是帮你屏蔽了一些参数的设置,直接用他获取线程池对象会有如下问题:他设置的默认参数对你的业务需求来说不见得合理,弄不好就OOM,对你了解线程池的原理也不是好事儿.所以项目中咱们用下一种方式.具体里面有几种线程池读者有兴趣可以自行去研究,项目开发中最好不要使用该种方式创建线程池.
2, ThreadPoolExecutor创建线程池.
/**corePoolSize:核心线程数量
maximumPoolSize:最大线程数量
keepAliveTime:线程多长时间没活干之后销毁(默认只针对非核心线程,但是可以通过allowCoreThreadTimeOut设置核心线程超时销毁)
unit:时间单位
workQueue:缓存任务的阻塞队列(当任务过来时候发现核心线程都在忙着就会先缓存进去该队列)
threadFactory:线程工厂
handler:拒绝策略(当任务无法被执行且不能被缓存时候执行)
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}线程池大致逻辑(摘自百度图片)
线程池中的线程是什么时候创建的：
我们使用线程池是直接调用execute方法执行任务的.
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//这里线程池的设计者巧妙的用1个int类型的变量表示了线程池的状态和当前线程的数量
//二进制的前3位表示线程池状态后29位表示线程的数量
if (workerCountOf(c) < corePoolSize) {
//如果当前的线程数量小于核心线程数量,尝试添加一个核心线程去执行当前任
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
//如果线程池现在是runing的状态,且入队成功
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
//double check线程池状态,如果此时线程池状态不是running移除添加的任务并执行拒绝策略
reject(command);
else if (workerCountOf(recheck) == 0)
//如果此时工作中的线程数量为0添加一个非核心线程
addWorker(null, false);
}
//入队也没吃成功添加非核心线程
else if (!addWorker(command, false))
reject(command);
}addWorker方法:
Worker是线程池的内部类,里面封装了一个线程对象,他本身又实现了runable接口,封装的这个线程就是任务的执行单元.这个线程在实例化的时候传的又是当前的Worker对象.有点绕多品品。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
//真正执行任务的线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//因为线程池采用的是hashset为保证线程安全,加锁
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//将worker添加到线程池中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
//滚动更新池中的最大线程数
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//这里就是执行任务的逻辑了,上面提到这个t是worker里面的一个成员,他的实例化传了worker对象,所以实际上这里执行的逻辑应该是worker实现runable接口后复写的run方法而worker类的run方法又是调用的线程池中的RunWorker方法
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//拿到worker里面封装的task
Runnable task = w.firstTask;
//将worker里面的task清空,这里变量名也取的很好,第一个任务,意思是worker第一个执行的任务肯定是当初实例化他传进去的runable类型的参数,当然这个任务也可能为空,比如线程池创建之后执行预先创建线程的方法prestartAllCoreThreads
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
//如果task不为空或者getTask不为空就去执行task
//我想大家已经猜到了,这个getTask多半就是从阻塞队列中获取任务了
//阻塞队列有什么特点？没有任务他就会阻塞在这里,所以这便可以回答前文的问题,
//他是靠阻塞队列的take方法的阻塞特性让线程挂起从而完成线程的复用的
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.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 {
//这个方法是销毁worker的方法,里面有将worker从池hashset中移除的逻辑,那么他什么时候会走到呢?
//1,上述逻辑发生了异常即completedAbruptly=true.
//2,上述逻辑正常退出,咦,刚不是说上述循环条件会卡在take方法阻塞住么,怎么会正常退出呢？我们看看getTask方法
processWorkerExit(w, completedAbruptly);
}
}
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);
//这里意思是说检查超时的必要条件要么是核心线程也允许超时要么是当前有非核心线程在运行着
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//忽略第一个极端条件超过最大线程数量不看,第一次进来是肯定不会进去这个分支的因为timeout为false.
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
//cas减少一个执行单元的数量,并没有销毁线程池中的线程对象,销毁动作在processWorkerExit方法中即将线程(worker)从池即hashset中移除
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//这里就是阻塞获取队列里面的任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
//获取到了任务
return r;
//超时没有获取到timeout就是true了,再次循环就会到上面减少运行线程数量的分支去了
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
最后的问题:
现在我们大致知道了线程池的原理,可是还有一个很棘手的问题,就是那几个参数应该如何设置才合理?在回答这个问题之前不妨先考虑一下为什么要用多线程,答案相信大家应该都知道为了提高cpu的利用率,我们又知道单核处理器在同一时间只会处理一个任务的,之所以我们可以边看知乎边听歌,得益于cpu的时间片机制在各个(进程)线程之间不断的来回切换运行让我们产生了同时运行的错觉。切线程的上下文切换的开销也不小,因为系统有用户态到内核态的切换。
我们主要看核心线程数量的参数设置
所以既然上下文切换有开销,所以线程并不是越多越好,在选择多线程时候要对具体的任务具体的分析:
CPU密集型任务:即该任务本来就是需要CPU大量参与的计算型任务,CPU的利用率已经很充分了,这个时候你的线程你再弄多的线程也只会增加线程上下文带来的额外开销罢了。所以此时应该理论上设置的线程数量和系统cpu数量相同,但是为了防止一些意外的发生一般设置为cpu数量+1个线程。
IO密集型任务:任务耗时主要体现在线程等待io操作的返回比如网络调用,文件读写之类的,这个时候cpu的利用率并没有得到充分的使用,所以理论上某种程度来说线程数应该是越多越好比如cpu数量的两倍.
关于参数的设置,经验性较强,因为你的服务器上不可能就运行你一个应用,我认为只需明白什么样的任务类型怎么样去设置,在这个大的思想下自己灵活变通就可以。
总结:
线程池的思想可以用大白话来解释就是公司项目组有5个程序员(核心线程数)平时的工作就是解决测试所提交的bug,测试会在一个bug管控平台(阻塞队列)上提交bug报告,闲的时候bug没有很多程序员做在电脑前带薪摸摸鱼等待测试提交新的bug(阻塞获取新的任务),忙的时候五个程序员996都忙不过来啦,bug管控平台都快因为提交bug的测试太多,登录都登录不了啦,这个时候老板说这样吧,我去招几个外包过来,外包过来了也帮着项目组在解决bug,但是这个时候还是有很多测试有bug要提交提交不上去,项目组老大怒了,说你们会不会用老子写的功能,你这提的(ie8不能正常显示)算什么bug，滚蛋(拒绝策略)项目开始闲下来了,bug也没有那么多了外包也就卷铺盖走人了(线程超时销毁).