标签:cut lin 带来 class long type www private 标准
参考:
https://blog.csdn.net/u014730165/article/details/84065996
https://www.jianshu.com/p/69a6ae850736
从上面章节的学习,我们了解到,每个执行的任务是实现Runnable接口。但是实现Runnable接口的弊端是没有返回值或者异常的抛出。如果现在又搞需求是,要求任务执行完成,返回任务执行的结果,那么Runnable就无法实现。在线程池中,提供了另外的一个提交方式submit提交,允许用户提交带返回结果的任务。这章节我们主要深入分析带返回结果的任务是如果执行的原理和方法。
从类继承图可以看出,ExecutorService 提供了 submit的三个方法的接口,在AbstractExecutorService 抽象类中进行了实现。我们看下AbstractExecutorService抽象类中的实现方法是:
public abstract class AbstractExecutorService implements ExecutorService {
// 传入实现Runnable的task任务
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 封装成 RunnableFuture<Void>,返回结果为Void类型
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
// 传入实现Runnable实现的task,同时传入可封装结果的result
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 封装成 RunnableFuture<T>,返回结果为泛型T
// 源码详情分析,请参考:4.1.1、submit(Runnable task, T result) 源码分析
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
// 传入实现Callable<T> task 任务,附带返回结果类型
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
// 封装当前任务,生产 RunnableFuture<T> ftask
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
}
从submit的方法参数可以发现,主要涉及的有以下的几个类和接口:Future,Callable,RunnableFuture。核心的封装方法为newTaskFor。下面我们依次分析这些类接口的源码实现。
@FunctionalInterface
public interface Callable<V> {
// call 接口,返回值为泛型,可以抛出异常
V call() throws Exception;
}
从类继承图可以发现,最终会把一个任务封装成FutureTask任务。
public interface Future<V> {
// 参数当可能运行的任务
boolean cancel(boolean mayInterruptIfRunning);
// 判断任务是否被删除
boolean isCancelled();
// 当前任务是否已经执行完成
boolean isDone();
// 等待计算完成,然后获取结果,会抛出异常
V get() throws InterruptedException, ExecutionException;
// 在指定时间内返回结果,否则抛出异常
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
继承 Runnable和Future接口,重写Runnable接口的run方法
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
public class FutureTask<V> implements RunnableFuture<V> {
// 传入实现Callable接口的任务
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
// 初始化当前task的状态
this.state = NEW; // ensure visibility of callable
}
// 传入实现Runnable接口的任务,同时传入封装结果集的result
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
// 初始化当前task的状态
this.state = NEW; // ensure visibility of callable
}
}
/*
* FutureTask中定义了一个state变量,用于记录任务执行的相关状态 ,状态的变化过程如下
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
private volatile int state;
// 开始状态
private static final int NEW = 0;
// 正在运行状态
private static final int COMPLETING = 1;
// 正常运行完结束状态
private static final int NORMAL = 2;
// 异常状态
private static final int EXCEPTIONAL = 3;
// 取消状态
private static final int CANCELLED = 4;
// 中断
private static final int INTERRUPTING = 5;
// 中断结束状态
private static final int INTERRUPTED = 6;
// 需要执行的任务,提示:如果提交的是Runnable对象,会先转换为Callable对象,这是构造方法参数
private Callable<V> callable;
// 任务运行的结果
private Object outcome;
// 执行此任务的线程
private volatile Thread runner;
// 等待该FutureTask的线程链表,对于同一个FutureTask,如果多个线程调用了get方法,对应的线程都会加入到waiters链表中
// 同时当FutureTask执行完成后,也会告知所有waiters中的线程
private volatile WaitNode waiters;
入口方法submit有三种实现方式,其他的两种实现方式,上面已经说过,不做分析。这里我们分析下比较特殊的实现方式:submit(Runnable task, T result)
传入Runnable task作为参数:
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 初始化FutureTask
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
初始化FutureTask通过调用: newTaskFor(Runnable runnable, T value)
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
// 通过构造函数实例化FutureTask实例
return new FutureTask<T>(runnable, value);
}
FutureTask构造函数:
public FutureTask(Runnable runnable, V result) {
// 通过调用Executors.callable的方法进行实例化
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
通过调用Executors的callable方法进行实例化,返回Callable实例
java.util.concurrent.Executors#callable(java.lang.Runnable, T)
这里使用了大家耳熟能详的适配器模式对Runnable接口进行适配成目标接口Callable的实例
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
java.util.concurrent.Executors.RunnableAdapter
RunnableAdapter是Executors的内部类,把Runnable任务通过适配器包装厂目标的Callable接口
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
通过上面分析,可以看出,当传入的是实现Runnable的接口实例的时候,会通过适配器进行适配,封装成目标接口Callable。
execute(Runnable command) 方法在我们上章节,线程池源码分析已经深入分析。这里主要通过线程池去执行task.run() 里面的核心代码,所以,我们execute方法的分析我们这里不做分析,详情看上一章节线程池源码分析。Java多线程之ThreadPoolExecutor实现原理和源码分析(五)
前面我们已经分析了FutureTask的实例化过程。其核心执行的逻辑是被封装在java.util.concurrent.FutureTask#run方法中。下面我们深入的分析下run方法的实现逻辑。
public void run() {
// 如果当前线程为 NEW的时候,通过CAS算法,将当前的成员变量 private volatile Thread runner; 初始化。
// 设置为当前执行线程,对应线程池中的一个线程
// 获取当前线程的主要用于get方法中唤醒操作。获取执行结果
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 调用用户自动源码,复制给返回结果 result
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 如果抛出异常,将当前的异常值赋值给outcome
// 源码详情请参考:5.1.2、 setException(Throwable t)
setException(ex);
}
// 如果执行成功,将临时变量result赋值给成员变量result
if (ran)
// set方法源码分析请参考5.1.1 java.util.concurrent.FutureTask#set
set(result);
}
} finally {
// 执行完成之后,将当前线程引用设置为null
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void set(V v) {
// 设置当前线程状态为执行中
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 赋值执行结果
outcome = v;
// 设置当前线程的状态为正常结束
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
// 唤醒阻塞在get() 方法上面的线程,详情源码请参考:5.1.3、 finishCompletion()
finishCompletion();
}
}
protected void setException(Throwable t) {
// 如果抛出异常,设置到当前任务的执行状态为正在运行
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 将异常值赋值给outcome
outcome = t;
// 将当前线程的最终状态设置为执行异常
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
// 完成任务之后,做清理,详细源码分析,参考: 5.1.3、 finishCompletion()
finishCompletion();
}
}
private void finishCompletion() {
// assert state > COMPLETING;
// 在多线程环境下,当有多个线程等待获取当前任务的结果的时候,都调用get()方法获取结果
// 这些等待结果的线程都会被存放到WaitNode waiters 链表中。
// 当执行完成后,都会唤醒当前阻塞的线程,获取当前的执行结果
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
// 唤醒指定的线程
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
// 设置当前的callable为null
callable = null; // to reduce footprint
}
当线程去异步去执行任务的时候,当我们需要获取执行结果。那么我们就得调用get() 方法获取线程执行的结果。下面我们深入分析下get() 方法。
public V get() throws InterruptedException, ExecutionException {
int s = state;
// 判断当前任务是否执行完成
if (s <= COMPLETING)
// 如果没有,将当前方法进行阻塞。传入参数 。flase表示一直阻塞。0
s = awaitDone(false, 0L);
// 6.3 java.util.concurrent.FutureTask#report 源码分析
return report(s);
}
timed:表示是否阻塞
nanos: 表示阻塞的时间
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 判断需要阻塞的时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
// 死循环,一直判断当前的任务状态
// 如果当前任务是第一次调用get方法,那么会把当前任务封装成一个WaitNode 节点,添加到queue中。
for (;;) {
// 如果当前线程已经被中断,直接移除等待,抛出中断异常
// 具体的中断方式,可以通过java.util.concurrent.FutureTask#cancel(boolean mayInterruptIfRunning) 进行设置
if (Thread.interrupted()) {
// 队列中移除当前
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
// 如果当前线程已经运行完成,直接返回结果
if (s > COMPLETING) {
if (q != null)
// 置空当前的线程引用
q.thread = null;
return s;
}
// 如果当前线程还在执行,则礼让出获取状态
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
// 第一次调用get()方法,会创建一个WaitNode 节点
// 第一次for循环,将当前节点q初始化
else if (q == null)
q = new WaitNode();
// 第二次for循环,将当前的q节点追加到waiters节点
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
// 第三次for循环,判断是否有时间限制。在指定的时间内获取结果,如果没有获取结果已出q从waiters中。
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
// 从waiters中已出,返回状态state,传送给 java.util.concurrent.FutureTask#report
removeWaiter(q);
return state;
}
// 阻塞指定的时间,自动唤醒
LockSupport.parkNanos(this, nanos);
}
else
// 没有时间限制,一直阻塞当前线程
// 唤醒之后,再次进入循环,直到满足条件退出
LockSupport.park(this);
}
}
方法参数s,代表当前任务的状态
private V report(int s) throws ExecutionException {
// 赋值输出结果
Object x = outcome;
// 任务正常结束,返回指定的返回值
if (s == NORMAL)
return (V)x;
// 如果任务是被删除的话,直接抛出异常
if (s >= CANCELLED)
throw new CancellationException();
// 都不满足抛出异常
throw new ExecutionException((Throwable)x);
}
至此,Future的内部实现原理和源码分析完成,如有错误之处,欢迎指正!!
前阵子在用C++ 98(是比较落后了,嗯,C++11原生支持Future)开发的时候,对脱离业务的公共逻辑抽象出来了一个简单的任务执行框架,里面主要是线程池和一些同步异步的任务。在开发异步任务的时候,为了实现类似java Future模式的能力,对实现方式考量了好久,最终使用了信号量这么重的东西来实现了Future的能力,同时也不禁对java的Future实现产生兴趣,java的Future是怎么实现的呢?
java Future相关的代码基本都在java.util.concurrent
的包里面,Future.java
是一个接口,定义了最基本的一些任务操作和状态判断。
// Future.java
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit);
我们从FutureTask.java
去了解Future
的内部机制,FutureTask
对Future
有比较接地气的实现,其他的实现或多或少都加入了新的一些特性,对了解原理没太大帮助。
FutureTask.java
是对Futre
和Runnable
最简单的实现,所以FutureTask
是一个可执行的异步对象。
FutureTask
定义了7种状态,这7种状态里面包含了简单的状态机,使用了一个用volatile
修饰的int
来记录状态。如下:
/** Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
状态的变换基本在FutureTask.java
的所有函数中都有体现,我们来看看几个典型的状态变换:
构造函数
构造函数完成时,将状态置为NEW
取消操作
取消操作对任务状态进行判断。
如果任务正在执行但没有完成时,发出中断,并将任务状态置为中断中状态,并在执行线程完成后,置为中断完成状态NEW -> INTERRUPTING -> INTERRUPTED
当任务还没有执行,则直接置为取消状态 NEW -> CANCELLED
执行操作
执行分执行成功和执行异常两种,状态转换路径是这两个。
NEW -> COMPLETING -> NORMAL
和NEW -> COMPLETING -> EXCEPTIONAL
FutureTask
因为封装了Runnable
的接口,实现了run
函数,所以可以直接执行,直接执行使用主线程;FutureTask
的另外一种执行方式是提交到线程池去执行,由线程池去分配执行线程;只有提交到线程池去执行才能体现异步的特性。不过我们不关注执行的方式,我们关注执行的逻辑。
FutureTask
的构造函数传入了Callable
或Runnable
对象,也即是需要执行的业务逻辑,他是业务逻辑的基本表现形式,保存在类属性callable
,在run
函数里面,调用callalbe.call()
来执行业务逻辑。run
函数主要完成以下几个操作。
NEW
,如果不是,说明任务被执行过,或者已被取消,直接返回。NEW
,接着会通过unsafe
类把任务执行线程保存在runner
字段中,如果保存失败,则直接返回。set
保存结果,不成功setException
保存异常信息,任务的状态在这里改变。以下是run
函数的逻辑。
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread())) // 1, 2
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call(); // 3
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex); // 4
}
if (ran)
set(result); // 4
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
由于在任务执行的过程中,可能被取消,所以在finally
块里,会任务的根据状态来做一些善后的工作。
任务的取消比较简单,我们知道,在执行的时候,执行任务的线程会保存在runner
属性中,所以对于正在执行的任务,取消的本质就是将执行的线程取出来,向该线程发出interupt
信号。但对于一个较为完备的取消动作,cancel
做了一下几个动作。
判断任务当前执行状态,如果任务状态不为NEW
,则说明任务或者已经执行完成,或者执行异常,不能被取消,直接返回false
表示执行失败。
判断需要中断任务执行线程,则
NEW
转化到INTERRUPTING
。这是个中间状态。INTERRUPTED
。如果不需要中断任务执行线程,直接把任务状态从NEW
转化为CANCELLED
。如果转化失败则返回false
表示取消失败。
调用finishCompletion
这个函数将阻塞在等待这个任务完成的线程唤醒,具体操作是LockSupport.unpark(t)
,这些线程都是在awaitDone
函数内的LockSupport.park(this)
中阻塞的,关于awaitDone
函数的作用后文还会继续介绍。finishCompletion
除了在这里有调用以外,在set
和setException
中也有调用。
以下是finishCompletion
的实现
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
以下是cancel
函数的实现。
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) // 1, 2
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // 3
}
}
} finally {
finishCompletion(); // 4
}
return true;
}
异步任务提交到线程池去执行后,由于无法预知什么时候结束,所以必须得提供接口来获取任务执行的结果,在FutureTask
中,get
函数用来获取任务执行的结果。该函数有了设置超时和不超时的两种实现。
除去超时的差异,get
操作对任务的状态进行判断,当状态还没有完成的时候,调用awaitDone
函数来等待完成,我们在上面提到过这个函数,对于未完成的任务awaitDone
阻塞,而finishCompletion
唤醒阻塞。
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
我们来详细看看awaitDone函数做了什么操作,它是如何阻塞的,它是怎么设置超时和完成返回的。
awaitDone
主题是一个死循环,轮询判断任务的状态。
当执行的线程被中断时,调用removeWaiter
移除等待节点WaitNode
,抛出中断异常
当状态为已经完成,直接返回
当状态为完成中,通过Thread.yield()
让出CPU时间
如果当前线程还没有创建WaitNode
等待节点保存到等待队列里面去,则新建一个等待节点,插入到等待链表,表明当前线程也准备进入等待该任务完成的队列中去。
最后是进入阻塞的动作,通过LockSupport.park
,如果设置了超时的时间,则将时间作为参数传递到park
中去。
综上,awaitDone
函数除了对状态的判断以外,核心就是LockSupport
的阻塞等待完成的操作了。我们后面还会探讨一下LockSupport
。
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
当awaitDone
的阻塞完成以后,就会将结果返回,将结果返回是通过report
函数来实现的,返回的是执行完成的结果或者是执行中获得的异常信息。
异步执行任务,在获取任务状态时,阻塞是必然的,在开头引子我简单的提到了在自己实现的那个简陋的框架内,使用了信号量的方式来设置异步任务的阻塞和执行状态。在这里是通过LockSupport
来实现的,阻塞(park)和唤醒(unpark)。
public static void unpark(Thread thread) {
if (thread != null)
UNSAFE.unpark(thread);
}
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
可以看到LockSupport
则是通过UNSAFE
的同名函数来实现的。java不能直接访问操作系统底层,而是通过本地方法来访问。Unsafe类提供了硬件级别的原子操作。
Unsafe类是在sun.misc包下,不属于java标准。但是很多Java的基础类库,包括一些被广泛使用的高性能开发库都是基于Unsafe类开发的,比如Netty、Cassandra、Hadoop、Kafka等。Unsafe类在提升Java运行效率,增强Java语言底层操作能力方面起了很大的作用。Unsafe类使Java拥有了像C语言的指针一样操作内存空间的能力,同时也带来了指针的问题。过度的使用Unsafe类会使得出错的几率变大,因此Java官方并不建议使用的,官方文档也几乎没有。
所以在这里java也是使用了C级别的线程同步机制来完成这些操作的,在这就不再展开了。
为保持文章内容的连贯性,部分内容参考自:
标签:cut lin 带来 class long type www private 标准
原文地址:https://www.cnblogs.com/xuwc/p/14090786.html