Executor框架分析实践1——入门
Executor框架——基础知识
Executor框架是Java用来管理多线程执行task的一套框架，本篇先介绍些此框架涉及到的几个重要的类。
介绍一些基类
-
Callable——有返回值的task
-
Runnable——无返回值的task
-
RunnableFuture<T>——通过Future来获取Runnable的执行结果
-
Executor类——task执行类
Executor类定义如下：
public interface Executor{
void executor(Runnable command);
}
Excutor类只有一个接口executor（Runnable command），用于将任务分解为单独task，并没有规定每个task必须运行在新线程中，也可以运行在调用线程。
异步运行的例子：
class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) {
new Thread(r).start();
}
另外，很多Executor要求task的实施必须是在某种特定的条件，比如，特定的时间，如何执行，下面这个例子要求task顺序执行。
class SerialExecutor implements Executor {
final Queue<Runnable> tasks = new ArrayDeque<>();
final Executor executor;
Runnable active;
SerialExecutor(Executor executor) {
this.executor = executor;
}
public synchronized void execute(final Runnable r) {
tasks.add(new Runnable() {
public void run() {
try {
r.run();
} finally {
scheduleNext();
}
}
});
if (active == null) {
scheduleNext();
}
}
protected synchronized void scheduleNext() {
if ((active = tasks.poll()) != null) {
executor.execute(active);
}
}
}}
-
ExecutorService类——关闭Task，Future
ExecutorService类提供了terminate方法，可以终止一个task。也提供了submit方法，能够返回一个feature用于跟踪task执行的过程
* ExecutorService可以被关闭，一旦进入shut Down，禁止task再被加入进来，会报RejectedExecutionException。一旦调用shutDown，对于已提交的（submitted）的task允许执行完成，但禁止添加新task；shutDownNow，阻止已经进入等待状态的task，也会尝试阻止当前正在执行的task。一旦进入shutDown状态，不会有active的task，也不会有task等待执行，更不能添加新的task。
* 一个没有用的Executor应该会被及时释放，资源可以循环利用；
* submit方法可以看成execute函数的升级版，返回了Feature类，用于获取runnable执行的状态，包括是否取消，是否已完成，完成后获取result
* invokeAny和invokeAll函数用于执行大部分的execute任务，执行一个task或执行一个collection，等待一个或所有task执行完成；
官方demo：
ExecutorService pool = Executors.newFixedThreadPool(poolSize);
try {
for (;;) {
pool.execute(runnable);
}
} catch (IOException ex) {
pool.shutdown();
}
another 官方demo，先shutdown，禁止增加task，等待60s后查看是否terminate，如果没有调用shutdownNow立即关闭：
void shutdownAndAwaitTermination(ExecutorService pool) {
pool.shutdown(); // Disable new tasks from being submitted
try { // Wait a while for existing tasks to terminate
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
pool.shutdownNow(); // Cancel currently executing tasks
// Wait a while for tasks to respond to being cancelled
if (!pool.awaitTermination(60, TimeUnit.SECONDS))
System.err.println("Pool did not terminate");
}
} catch (InterruptedException ie) {
// (Re-)Cancel if current thread also interrupted
pool.shutdownNow();
// Preserve interrupt status
Thread.currentThread().interrupt();
}
}}
-
Future类——查看task执行状态
- cancle——尝试取消task，如果task已经执行完成或者已经被取消，或者因为某些原因不能被取消，返回false；如果task没有开始，cancle之后task不会再被执行；如果task已经开始了，那么mayInterruptIfRunning参数会决定是否通过打断线程来终止task；
- isCancle
- isDone
- get（）阻塞式获取，等待知道task执行完成，然后获取他的result。还有另外一个method，限制了等待的时间 V get(long timeout,TimeUnit unit)
-
FutureTask类——Future类的基本实现类implementation
- 前面都是interface，此类为具体实现，包括cancle，get，run等等；
- task的几个状态
- NEW -> COMPLETING -> NORMAL
- NEW -> COMPLETING -> EXCEPTIONAL
- NEW -> CANCELLED
- NEW -> INTERRUPTING -> INTERRUPTED
-
Executors类——executorServices的工厂类
- 负责生产ExecutorService、Executor、ScheduledExecutorService、ThreadFactory，callable
- 利用ThreadPoolExecutor类来生成各式各样的ThreadPool
- 未完，待续
-
ThreadPoolExecutor类——利用线程池中的thread来execute每个被submitted的task
- Inter的高三位用来存放runState，剩下的用来存放worker的个数，所以runState的类型是int
-
Worker 内部类，保存了运行task的thread的中断控制，扩展了AbstractQueuedSynchronizer，简化了获取lock和释放lock的操作。内部参数：Thread(控制task运行的线程),Runnable（第一个task）
- execute函数通过addWorker来实现
- 内部参数：
- ThreadFactory（用来产生线程），
- workQueue（runnable的queue），
- worker（task通过worker的run来运行，控制中断）
- workers（worker的HashSet，用来管理worker），
- corePoolSize(最小保持alive的线程数)，
-
Easy Demo
未完待续。。。
