Callable原理分析

前言

今天我们来分析实现多线程的一种方式,实现Callable接口。这种方式有种特殊的地方,就是可以拿到线程返回值。具体怎么实现的呢?我们来研究下。

栗子

我们先举个简单的栗子来看下Callable接口吧。
要首先明白,线程池提交实现Callable接口的线程后会返回一个Future对象,这个对象里包含程序的运行结果。

1
2
3
4
5
6
7
8
9
10
11
12
public class CallableTest {
public static void main(String[] args) throws ExecutionException, InterruptedException,TimeoutException{
//创建一个线程池
ExecutorService executor = Executors.newCachedThreadPool();
Future<String> future = executor.submit(()-> {
//System.out.println("CallableTest");
TimeUnit.SECONDS.sleep(5);
return "CallableTest";
});
System.out.println(future.get());
}
}

可以看到我们可以通过future.get()拿到结果”CallableTest”。
我们也可以设置指定时间后拿到结果,如指定6s后拿到结果。

1
System.out.println(future.get(4,TimeUnit.SECONDS));

可以看到也拿到了返回结果,如果我们设置4s拿到结果,小于程序运行时间5s,可以看到它抛出了超时异常。java.util.concurrent.TimeoutException。

原理

是不是很神奇?

我们来研究下Callable接口获取返回值的原理。

我们先来看看ExecutorService的submit方法,它接受一个Callable对象,返回一个Future对象。

1
<T> Future<T> submit(Callable<T> task);

及它的实现。AbstractExecutorService的submit方法。

1
2
3
4
5
6
7
8
9
10
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
void execute(Runnable command);

ThreadPoolExecutor的execute是对其实现。

upload successful

可以看到创建了一个FutureTask对象并执行。

FutureTask对象实现了Runable接口,我们来看下它。

upload successful

看一下它的run方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public void run() {
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 {
//拿到结果设置ran为true
result = c.call();
ran = true;
} catch (Throwable ex) {
//异常设置结果为空ran为false并设置异常
result = null;
ran = false;
setException(ex);
}
//ran为true时放入结果
if (ran)
set(result);
}
} 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);
}
}
1
2
3
4
5
6
7
public V get() throws InterruptedException, ExecutionException {
int s = state;
//会一直挂起知道处理业务的线程完成,唤醒等待线程
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}

我们调用get方法时,他先查询线程状态,如果未完成,就调用awaitDone方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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);
}
}

带有超时时间的get,到达时间后,会判断线程状态,如果未完成,抛出超时异常。

1
2
3
4
5
6
7
8
9
10
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}

总结

因此,带有返回值得异步线程基本上可以这样理解。

由于有返回值,如果未设置等待时间,会等线程执行完成后返回,基本类似同步。其原理是线程运行后,如果未完成,会放入等待队列。直到线程状态变化(完成或者异常等)。如果设置了等待时间,则到时间后无论线程状态是否完成都会返回线程状态。




-------------文章结束啦 ~\(≧▽≦)/~ 感谢您的阅读-------------

您的支持就是我创作的动力!

欢迎关注我的其它发布渠道