java.util.concurrent.ExecutorService与Executors例子的简单剖析
java.util.concurrent.ExecutorService与Executors例子的简单剖析
?
对于多线程有了一点了解之后，那么来看看java.lang.concurrent包下面的一些东西。在此之前，我们运行一个线程都是显式调用了Thread的start()方法。我们用concurrent下面的类来实现一下线程的运行，而且这将成为以后常用的方法或者实现思路。?
?
? ? ? ? 看一个简单的例子：?
?
Java代码 ?
public class CacheThreadPool { ?
? ? public static void main(String[] args) { ?
? ? ? ? ExecutorService exec=Executors.newCachedThreadPool(); ?
? ? ? ? for(int i=0;i<5;i++) ?
? ? ? ? ? ? exec.execute(new LiftOff()); ?
? ? ? ? exec.shutdown();//并不是终止线程的运行，而是禁止在这个Executor中添加新的任务 ?
? ? } ?
} ?
?
? ? ? ? 这个例子其实很容易看懂，ExecutorService中有一个execute方法，这个方法的参数是Runnable类型。也就是说，将一个实现了Runnable类型的类的实例作为参数传入execute方法并执行，那么线程就相应的执行了。?
?
? ? ? ? 一、ExecutorService?
? ? ? ? 先看看ExecutorService，这是一个接口，简单的列一下这个接口：?
?
Java代码 ?
public interface ExecutorService extends Executor { ?
??
? ? void shutdown(); ?
??
? ? List<Runnable> shutdownNow(); ?
??
? ? boolean isShutdown(); ?
??
? ? boolean isTerminated(); ?
??
? ? boolean awaitTermination(long timeout, TimeUnit unit) ?
??
? ? <T> Future<T> submit(Callable<T> task); ?
??
? ? <T> Future<T> submit(Runnable task, T result); ?
??
? ? Future<?> submit(Runnable task); ?
??
? ? <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) ?
??
? ? <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) ?
??
? ? <T> T invokeAny(Collection<? extends Callable<T>> tasks) ?
??
? ? <T> T invokeAny(Collection<? extends Callable<T>> tasks, ?
? ? ? ? ? ? ? ? ? ? long timeout, TimeUnit unit) ?
} ?
?
? ? ? ? ExecuteService继承了Executor，Executor也是一个接口，里面只有一个方法：?
?
Java代码 ?
void execute(Runnable command) ?
?
?
? ? ? ? 二、Executors?
? ? ? ? Executors是一个类，直接援引JDK文档的说明来说一下这个类的作用：?
? ? ? ?
?
? ? ? ? Factory and utility methods for Executor, ExecutorService, ScheduledExecutorService, ThreadFactory, and Callable classes defined in this package. This class supports the following kinds of methods:?
?
?
? ? ? ?
Methods that create and return an ExecutorService set up with commonly useful configuration settings.
? ? ? ?
Methods that create and return a ScheduledExecutorService set up with commonly useful configuration settings.
? ? ? ?
Methods that create and return a "wrapped" ExecutorService, that disables reconfiguration by making implementation-specific methods inaccessible.
? ? ? ?
Methods that create and return a ThreadFactory that sets newly created threads to a known state.
? ? ? ?
Methods that create and return a Callable out of other closure-like forms, so they can be used in execution methods requiring Callable.
? ? ? ?
?
? ? ? ? 在上面的例子中，我们用到了newCachedThreadPool()方法。看一下这个方法：?
?
Java代码 ?
public static ExecutorService newCachedThreadPool() { ?
? ? ? ? return new ThreadPoolExecutor(0, Integer.MAX_VALUE, ?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 60L, TimeUnit.SECONDS, ?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? new SynchronousQueue<Runnable>()); ?
? ? } ?
?
? ? ? ? 在源码中我们可以知道两点，1、这个方法返回类型是ExecutorService；2、此方法返回值实际是另一个类的实例。看一下这个类的信息：?
?
Java代码 ?
public class ThreadPoolExecutor extends AbstractExecutorService { ?
? ? .......... ?
? ? private final BlockingQueue<Runnable> workQueue;//这个变量在下面会提到 ?
? ? .......... ?
} ?
?
? ? ? ? ThreadPoolExecutor继承了AbstractExecutorService,而AbstractExecutorService又实现了ExecutorService接口。所以，根据多态，ThreadPoolExecutor可以看作是ExecutorService类型。?
?
? ? ? ? 线程执行的最关键的一步是执行了executor方法，根据java的动态绑定，实际执行的是ThreadPoolExecutor所实现的executor方法。看看源码：?
?
Java代码 ?
public class ThreadPoolExecutor extends AbstractExecutorService { ?
? ? .......... ?
? ? public void execute(Runnable command) { ?
? ? ? ? if (command == null) ?
? ? ? ? ? ? throw new NullPointerException(); ?
? ? ? ? if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { ?
? ? ? ? ? ? if (runState == RUNNING && workQueue.offer(command)) { ?
? ? ? ? ? ? ? ? if (runState != RUNNING || poolSize == 0) ?
? ? ? ? ? ? ? ? ? ? ensureQueuedTaskHandled(command); ?
? ? ? ? ? ? } ?
? ? ? ? ? ? else if (!addIfUnderMaximumPoolSize(command)) ?
? ? ? ? ? ? ? ? reject(command); // is shutdown or saturated ?
? ? ? ? } ?
? ? } ?
? ? .......... ?
} ?
?
? ? ? ? 根据程序正常执行的路线来看，这个方法中比较重要的两个地方分别是：?
? ? ? ? 1、workQueue.offer(command)?
? ? ? ? workQueue在上面提到过，是BlockingQueue<Runnable>类型的变量，这条语句就是将Runnable类型的实例加入到队列中。?
? ? ? ? 2、ensureQueuedTaskHandled(command)?
? ? ? ? 这个是线程执行的关键语句。看看它的源码：?
?
Java代码 ?
public class ThreadPoolExecutor extends AbstractExecutorService { ?
? ? .......... ?
? ? private void ensureQueuedTaskHandled(Runnable command) { ?
? ? ? ? final ReentrantLock mainLock = this.mainLock; ?
? ? ? ? mainLock.lock(); ?
? ? ? ? boolean reject = false; ?
? ? ? ? Thread t = null; ?
? ? ? ? try { ?
? ? ? ? ? ? int state = runState; ?
? ? ? ? ? ? if (state != RUNNING && workQueue.remove(command)) ?
? ? ? ? ? ? ? ? reject = true; ?
? ? ? ? ? ? else if (state < STOP && ?
? ? ? ? ? ? ? ? ? ? ?poolSize < Math.max(corePoolSize, 1) && ?
? ? ? ? ? ? ? ? ? ? ?!workQueue.isEmpty()) ?
? ? ? ? ? ? ? ? t = addThread(null); ?
? ? ? ? } finally { ?
? ? ? ? ? ? mainLock.unlock(); ?
? ? ? ? } ?
? ? ? ? if (reject) ?
? ? ? ? ? ? reject(command); ?
? ? ? ? else if (t != null) ?
? ? ? ? ? ? t.start(); ?
? ? } ?
? ? .......... ?
} ?
?
? ? ? ? 在这里我们就可以看到最终执行了t.start()方法来运行线程。在这之前的重点是t=addThread(null)方法，看看addThread方法的源码：?
?
Java代码 ?
public class ThreadPoolExecutor extends AbstractExecutorService { ?
? ? .......... ?
? ? private Thread addThread(Runnable firstTask) { ?
? ? ? ? Worker w = new Worker(firstTask); ?
? ? ? ? Thread t = threadFactory.newThread(w); ?
? ? ? ? if (t != null) { ?
? ? ? ? ? ? w.thread = t; ?
? ? ? ? ? ? workers.add(w); ?
? ? ? ? ? ? int nt = ++poolSize; ?
? ? ? ? ? ? if (nt > largestPoolSize) ?
? ? ? ? ? ? ? ? largestPoolSize = nt; ?
? ? ? ? } ?
? ? ? ? return t; ?
? ? } ?
? ? .......... ?
} ?
?
? ? ? ? 这里两个重点，很明显：?
? ? ? ? 1、Worker w = new Worker(firstTask)?
? ? ? ? 2、Thread t = threadFactory.newThread(w)?
? ? ? ? 先看Worker是个什么结构：?
?
Java代码 ?
public class ThreadPoolExecutor extends AbstractExecutorService { ?
? ? .......... ?
? ? private final class Worker implements Runnable { ?
? ? ? ? .......... ?
? ? ? ? Worker(Runnable firstTask) { ?
? ? ? ? ? ? this.firstTask = firstTask; ?
? ? ? ? } ?
??
? ? ? ? private Runnable firstTask; ?
? ? ? ? .......... ?
??
? ? ? ? public void run() { ?
? ? ? ? ? ? try { ?
? ? ? ? ? ? ? ? Runnable task = firstTask; ?
? ? ? ? ? ? ? ? firstTask = null; ?
? ? ? ? ? ? ? ? while (task != null || (task = getTask()) != null) { ?
? ? ? ? ? ? ? ? ? ? runTask(task); ?
? ? ? ? ? ? ? ? ? ? task = null; ?
? ? ? ? ? ? ? ? } ?
? ? ? ? ? ? } finally { ?
? ? ? ? ? ? ? ? workerDone(this); ?
? ? ? ? ? ? } ?
? ? ? ? } ?
? ? } ?
??
? ? Runnable getTask() { ?
? ? ? ? for (;;) { ?
? ? ? ? ? ? try { ?
? ? ? ? ? ? ? ? int state = runState; ?
? ? ? ? ? ? ? ? if (state > SHUTDOWN) ?
? ? ? ? ? ? ? ? ? ? return null; ?
? ? ? ? ? ? ? ? Runnable r; ?
? ? ? ? ? ? ? ? if (state == SHUTDOWN) ?// Help drain queue ?
? ? ? ? ? ? ? ? ? ? r = workQueue.poll(); ?
? ? ? ? ? ? ? ? else if (poolSize > corePoolSize || allowCoreThreadTimeOut) ?
? ? ? ? ? ? ? ? ? ? r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); ?
? ? ? ? ? ? ? ? else ?
? ? ? ? ? ? ? ? ? ? r = workQueue.take(); ?
? ? ? ? ? ? ? ? if (r != null) ?
? ? ? ? ? ? ? ? ? ? return r; ?
? ? ? ? ? ? ? ? if (workerCanExit()) { ?
? ? ? ? ? ? ? ? ? ? if (runState >= SHUTDOWN) // Wake up others ?
? ? ? ? ? ? ? ? ? ? ? ? interruptIdleWorkers(); ?
? ? ? ? ? ? ? ? ? ? return null; ?
? ? ? ? ? ? ? ? } ?
? ? ? ? ? ? ? ? // Else retry ?
? ? ? ? ? ? } catch (InterruptedException ie) { ?
? ? ? ? ? ? ? ? // On interruption, re-check runState ?
? ? ? ? ? ? } ?
? ? ? ? } ?
? ? } ?
? ? } ?
? ? .......... ?
} ?
?
? ? ? ? Worker是一个内部类。根据之前可以知道，传入addThread的参数是null，也就是说Work中firstTask为null。
? ? ? ? 在看看newThread是一个什么方法：?
?
Java代码 ?
public class Executors { ?
? ? .......... ?
? ? static class DefaultThreadFactory implements ThreadFactory { ?
? ? ? ? .......... ?
? ? ? ? public Thread newThread(Runnable r) { ?
? ? ? ? ? ? Thread t = new Thread(group, r, ?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? namePrefix + threadNumber.getAndIncrement(), ?
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 0); ?
? ? ? ? ? ? if (t.isDaemon()) ?
? ? ? ? ? ? ? ? t.setDaemon(false); ?
? ? ? ? ? ? if (t.getPriority() != Thread.NORM_PRIORITY) ?
? ? ? ? ? ? ? ? t.setPriority(Thread.NORM_PRIORITY); ?
? ? ? ? ? ? return t; ?
? ? ? ? } ?
? ? ? ? .......... ?
? ? } ?
? ? .......... ?
} ?
?
? ? ? ? 通过源码可以得知threadFactory的实际类型是DefaultThreadFactory，而DefaultThreadFactory是Executors的一个嵌套内部类。?
?
? ? ? ? 之前我们提到了t.start()这个方法执行了线程。那么现在从头顺一下，看看到底是执行了谁的run方法。首先知道，t=addThread(null)，而addThread内部执行了下面三步，Worker w = new Worker(null);Thread t = threadFactory.newThread(w);return t;这里两个t是一致的。?
? ? ? ? 从这里可以看出，t.start()实际上执行的是Worker内部的run方法。run()内部会在if条件里面使用“短路”：判断firstTask是否为null，若不是null则直接执行firstTask的run方法；如果是null，则调用getTask()方法来获取Runnable类型实例。从哪里获取呢？workQueue！在execute方法中，执行ensureQueuedTaskHandled(command)之前就已经把Runnable类型实例放入到workQueue中了，所以这里可以从workQueue中获取到。
