标签：null html func iss 精准 comment signal 自己 ++
Eexecutor接口：包含Eexecutor、ExecutorService、ScheduledExecutorService
ThreadPool线程池：包含ThreadPoolExecutor、ScheduledThreadPoolExecutor
Fork/Join框架：JDK1.7新增
类之间的关系如下：
Executor框架将线程的创建与执行解耦，可以异步调用，让任务相互独立，用阻塞队列管理任务，直接在当前线程中消费队列，可以减少线程之间进行资源竞争，也可以减少线程的创建和系统的开销，要廉价多了。这种设计就是经典并发模式Active Object Models（也称Actor Models）的实现，如下图：
可以参考execute方法，就按照上面的模式来的。另外Executors工厂类创建了不同的连接池，为任务的执行分配了不同执行策略。
concurrent包里面主要包含ThreadPoolExecutor、ScheduledThreadPoolExecutor两种线程池，Executors类提供了很多创建线程池的方法
，newFixedThreadPool创建定长的线程池、newWorkStealingPool创建ForkJoinPool（jdk1.8新增）、newCachedThreadPool创建缓存线程池（可以回收空闲的线程）、newSingleThreadExecutor创建当个线程池（保证FIFOLIFO优先级），newScheduledThreadPool创建定时器线程池，如果有特殊处理的，也可以根据自己的需求来创建连接池，如tomcat也是基于ThreadPoolExecutor实现了自己的连接池。
public (int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
从ThreadPoolExecutor最简单的构造函数可以看出，线程池总是依赖阻塞队列而工作的，newFixedThreadPool与newSingleThreadExecutor使用的是LinkedBlockingQueue无界队列，newCachedThreadPool使用SynchronousQueue，ThreadPoolExecutor还定义了一个Worker执行任务线程，除此之外，还有个非常重要的变量ctl（线程池控制状态）由执行器状态和工作线程的数量组成，在控制执行的时候都是围绕这个变量来判断。
参考前面的Active Object Models图，可以将线程池执行任务主要分为3个步骤：
1、如果任务少于线程池大小时，就作为firstTask分别创建工作线程执行任务
2、如果第n个任务超出了线程池大小，就加入到阻塞队列，并从新检测执行器状态状态以及工作线程数量（有可能线程发生RuntimeException挂掉，最后可工作线程数量变为0），如果工作线程挂完了就重新启动一个线程（解决线程泄露问题）。阻塞队列的任务执行就在第一步所创建的线程执行，参考代码：
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();
}
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);
}
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 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);
大专栏 concurrent包分析之Executor框架 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 {
processWorkerExit(w, completedAbruptly);
}
}
3、如果任务入队列失败了，就重新开启一个线程去执行，如果还是失败了，就可以确定是执行器关闭了或者线程池已经达到饱和状态了。
从上面的执行步骤来就可以看出线程池依赖于阻塞队列，然后基于生产者消费者模式实现的，execute()方法一直添加任务（生产者），当任务数量超出线程池的最大长度就添加到阻塞队列等待排队执行，而第一次创建的所有工作线程（最大数量为线程池的最大长度）就会一直判断线程池里面是否有任务执行，如果有就执行任务（消费者）。这样做就可以重用线程了，不用每次去创建线程，性能肯定比一个任务一个线程好多了。
ScheduledThreadPoolExecutor由ScheduledFutureTask、DelayedWorkQueue组成，实现任务执行还是用的父类ThreadPoolExecutor的工作线程。jdk1.5之前使用定时任务都用的Timer，但是与ScheduledThreadPoolExecutor有着明显区别：
还是从一个简单的例子看ScheduledThreadPoolExecutor是如何工作的。addWork参考领导/跟随者模式
@Test
public void testScheduleAtFixedRate() throws Exception {
ScheduledThreadPoolExecutor p = new ScheduledThreadPoolExecutor(1);
CountDownLatch latch = new CountDownLatch(3);
p.scheduleAtFixedRate(new Task(latch), 0, 1000, TimeUnit.MILLISECONDS);
latch.await();
}
线程池都是基于生产者消费者模式的，所以ScheduledThreadPoolExecutor也不例外，执行任务的时候会一直像队列里面添加任务：
private void delayedExecute(RunnableScheduledFuture<?> task) {
if (isShutdown())
reject(task);
else {
//添加到队列，让工作线程去消费
super.getQueue().add(task);
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
//启动线程
ensurePrestart();
}
}
ScheduledThreadPoolExecutor第一次执行用的父类ThreadPoolExecutor的addWorker方法添加工作线程并启动它，然后每个工作线程会检查队列里面是否有消费的线程，参考ThreadPoolExecutor的runWorker方法，与ThreadPoolExecutor有点不同的是，ScheduledThreadPoolExecutor使用的是延时阻塞队列DelayedWorkQueue。每次消费就进行take操作：
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0];
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
//返回队列中的执行任务之前，要先执行finally模块中的唤醒操作
return finishPoll(first);
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
//领导线程初始化为null，当前线程为线程池中的线程
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
//当前线程等待延迟时间到期
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
参考资料：
标签：null html func iss 精准 comment signal 自己 ++
原文地址：https://www.cnblogs.com/lijianming180/p/12230943.html
