Future 和Future task

原文链接

public class FutureAndCallableExample {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		ExecutorService executorService = Executors.newSingleThreadExecutor();

		Callable<String> callable = () -> {
			// 做一些操作
			Thread.sleep(2000);
			return "Hello, Martian!";
		};

		Future<String> future = executorService.submit(callable);

		String result = future.get();

		System.out.println(result);

		// 关闭所有tasks，防止同一个thread group里面的task导致main thread被block住无法退出。
		executorService.shutdown();
	}
}

关于futureTask的核心代码

Callable<V> c = callable;
V result;
result = c.call();
set(result);

在executor service里面submit一个任务的全流程：

• 我们通过Executors得到一个ExecutorService
• 我们创建一个Callable的任务
• 我们把callable的任务submit进ExecutorService
• ExecutorService把我们的callable任务封装成FutureTask
• ExecutorService帮我们管理运行task所需要的thread，并把thread和task都交给worker去执行。
• worker负责在thread里面执行task。
• task的类型是FutureTask，在内部执行我们的callable task，并把结果放在内部的outcome变量里。

future.get

public V get() throws InterruptedException, ExecutionException {
	int s = state;
	if (s <= COMPLETING)
		s = awaitDone(false, 0L);
	return report(s);
}

关于awaitDone方法

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);
	}
}