java线程池
2018-10-28 本文已影响0人
CoderZzbJohn
1.线程池活跃线程。
2.阻塞队列。
3.增加的线程。
4.拒绝策略。
1.ctl。线程池中用一个AtomicInteger表示线程池状态和活动线程数。高三位表示线程池状态。后28位表示线程个数。
线程状态：
1. runing：-1。正在运行中
2. shutdown 0 。 线程池已关闭，已加入线程池中的任务会继续进行。
3. stop 1。 线程池中任务立即停止。
4. tidying
5 terminated
1.线程池刚初始化，线程数< corePoolSize时。这时候来了任务，直接初始化线程，将任务分配给该线程。
2.当线程数=corePoolSize时。这时候来了任务，加入到队列，由线程自取。线程一直在消费任务
3.当队列已经满了且线程数<maxnumPoolSize，则创建新的线程，并将任务直接分配给该线程。
4.线程数=maxPoolSize，将任务加入队列。若队列已满，执行拒绝策略。
int c = ctl.get();
// 判断如果当前线程数小于corePoolSize
if (workerCountOf(c) < corePoolSize) {
// true：corePoolSize false：maxnumPoolSize
// 增加一个线程处理当前任务
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);
}
//新创建线程处理任务，线程个数不能超过maxnumPoolSize，否则拒绝任务
else if (!addWorker(command, false))
reject(command);
//增加工作线程
// core true：与corePoolSize相比 false：与maxnumPoolSize相比较
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) {
// 整个类的全局锁。关闭线程池，增加Worker等都需要获取锁。
final ReentrantLock mainLock = this.mainLock;
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();
workers.add(w);
int s = workers.size();
//记录线程池中线程数最多的个数
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 添加线程成功。调用线程run方法
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//直接处理firstTask。若firstTask为null，到队列中取任务。
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
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 {
// 若task为null，关闭线程。
processWorkerExit(w, completedAbruptly);
}
}
该方法一共三种情况。
1.阻塞直到任务返回。corePoolSize内的线程默认不会回收。但是可以设置。
2.超时返回。超过keepAliveTime没有任务，返回。
3.返回值为null时，关闭线程。 返回null情况：
3.1.线程池线程个数>maxnumPoolSize,开发者调用了setMaxnumPoolSize。
3.2.线程池状态为stop。或者状态为shutdown并且任务队列为空。
/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait, and if the queue is
* non-empty, this worker is not the last thread in the pool.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
//阻塞，直到获取到任务。 BlockQueue，阻塞队列。
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())) {
// 线程数超过核心线程数，且已超时，timeOut==true
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
拒绝策略：
//用当前线程来执行任务
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
// 只要队列已满。抛异常。Default
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
//不做任何操作
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
// 将队列里等待时间最长的去除，换成当前任务。
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
Executors:
// 线程数固定在 nThreads。 无界队列。这n个线程永远不会被销毁，除非线程池销毁了。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
// 单个线程。无界队列。线程创建了之后不会被销毁。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
// 只要有任务，没有空闲线程。就创建线程。创建的线程，如果60s内没有新任务，则销毁线程。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
Executor:
ExecutorService:
AbstractExecutorService:
FutureTask:
线程池corePoolSize数量设置建议：
1.CPU密集型应用
CPU密集的意思是任务需要进行大量复杂的运算，几乎没有阻塞，需要CPU长时间高速运行。
一般公式：corePoolSize=CPU核数+1个线程。JVM可运行的CPU核数可以通过Runtime.getRuntime().availableProcessors()查看。
2.IO密集型应用
IO密集型任务会涉及到很多的磁盘读写或网络传输，线程花费更多的时间在IO阻塞上，而不是CPU运算。一般的业务应用都属于IO密集型。
参考公式：最佳线程数=CPU数/(1-阻塞系数); 阻塞系数=线程等待时间/(线程等待时间+CPU处理时间) 。
IO密集型任务的CPU处理时间往往远小于线程等待时间，所以阻塞系数一般认为在0.8-0.9之间，以4核单槽CPU为例，corePoolSize可设置为 4/(1-0.9)=40。当然具体的设置还是要根据机器实际运行中的各项指标而定。
上述设置建议是希望一个线程池能最大的发挥CPU资源，实际上是以机器为维度的，如果应用需要几个线程池同时执行，建议值就对应这些线程池的核心线程总数。另外，实际设置的线程数只需要满足使用即可（例如设10、20个，任务的性能和运行时间就能被接受了），不用都设置的要占满CPU资源一样。
