今天我们了解下java中提供的并发框架Executor。
首先了解下Executors中的常用的几个方法
1.创建一个固定线程池
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<runnable>());
}
/<runnable>2.创建只有一个线程的线程池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<runnable>()));
}
/<runnable>3.不限制最大线程数
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<runnable>());
}
/<runnable>4.定时器，延时执行的线程池
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
以上四个方法中有哪些相同点和不同点？
要想了解这些，需要先了解一下ThreadPoolExecutor的构造方法都需要哪些
public ThreadPoolExecutor(int corePoolSize,//核心线程数量
int maximumPoolSize, //最在在线程数量
long keepAliveTime,//超时时间，超出线程数量以外的线程的空余线程的存活时间
TimeUnit unit,//存活时间的单位
BlockingQueue<runnable> workQueue,//阻塞队列
RejectedExecutionHandler handler) {//reject操作
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
/<runnable>
我可以理解Executors中的这四个方法其实就是ThreadPoolExecutor构造方法的变种。
下面我们分析ThreadPoolExecutor的实现方式
execute()方法
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();
}
//线程在运行并行，offer:核心线程数已经满了
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);
}
addWorker()方法
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;
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) {//若添加成功，则启动线程
t.start();//这里的start是指的Worker的线程
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
在这里说明下Worker中ThreadPoolExecutor中的内部类实现 AQS
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{...}
start()方法后执行Worker中的run()方法
public void run() {
runWorker(this);
}
runWorker()
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.run();//这里不是启动一个线程,则是运行一个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);
}
}
线程池的主要目的是什么?
是通过Worker中的for循环不停的执行我们的任务,从而达到线程复用的目的.
还有四种拒绝策略
AbortPolicy 处理失败后直接抛异常
CallerRunsPolicy 调用者所在的线程执行任务
DiscardOldestPolicy 丢弃最靠前的任务,并执行当前任务
DiscardPolicy 直接丢弃
若不使用以上四种策略,则可以自己通过实现RejectedExecutionHandler 接口的方式。
今天的分析已经讲完了， 讨论一个线上问题，通过并发框架Executors默认提供的方法，会达到一个瓶颈造成OOM，请问是哪些方法会造成？
了解的同学请留言回答，在回答的过程中分析线上应用。
----------
分享到:
閱讀更多 零售雲技術內參 的文章
