并发编程16-Callable,FutureTask
1.概述
- 通过实现Callback接口，并用Future可以来接收多线程的执行结果
- Future表示一个可能还没有完成的异步任务的结果，针对这个结果可以添加Callback以便在任务执行成功或失败后作出相应的操作。
2.示例
2.1 核心代码
Callable<Integer> call = new Callable<Integer>();
FutureTask<Integer> task = new FutureTask<>(call);
task.get();
2.2 简要示例
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
public class FutureThread {
/**
* Callalbe和Runnable的区别
* Runnable run方法是被线程调用的，在run方法是异步执行的
* Callable的call方法，不是异步执行的，是由Future的run方法调用的
*/
public static void main(String[] args) throws Exception {
Callable<Integer> call = new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("正在计算结果...");
Thread.sleep(3000);
return 1;
}
};
FutureTask<Integer> task = new FutureTask<>(call);
Thread thread = new Thread(task);
thread.start();
// do something
System.out.println(" 干点别的...");
Integer result = task.get();
System.out.println("拿到的结果为：" + result);
}
}
3. Callable与Runnable
3.1 源码分析
- Runnable 的run方法为void
public interface Runnable {
public abstract void run();
}
- Callable 泛型接口，call()函数返回的类型就是传递进来的V类型。
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
3.2 接口使用
Callable一般配合ExecutorService来使用的，在ExecutorService接口中声明了若干个submit方法的重载版本
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
3.3 Callable和Runnable的区别
接口可以继承多个接口. interface RunnableFuture就是这样.
Callable和Runnable的区别
Runnable是线程调用的,再ruan方法中是异步执行的,
Callable的call方法，不是异步执行的，是由Future的run方法调用的(耗时操作)
4 Future及相关类关系
4.1 Future及相关类图
4.2 Future相关类介绍
- RunnableFuture
这个接口同时继承Future接口和Runnable接口，在成功执行run（）方法后，可以通过Future访问执行结果。这个接口都实现类是FutureTask,一个可取消的异步计算，这个类提供了Future的基本实。如果计算没有完成，get方法会阻塞，一旦计算完成，这个计算将不能被重启和取消，除非调用runAndReset方法。
- FutureTask
能用来包装一个Callable或Runnable对象，因为它实现了Runnable接口，而且它能被传递到Executor进行执行。为了提供单例类，这个类在创建自定义的工作类时提供了protected构造函数。
-
SchedualFuture
这个接口表示一个延时的行为可以被取消。通常一个安排好的future是定时任务SchedualedExecutorService的结果 -
CompleteFuture
一个Future类是显示的完成，而且能被用作一个完成等级，通过它的完成触发支持的依赖函数和行为。当两个或多个线程要执行完成或取消操作时，只有一个能够成功。 -
ForkJoinTask
基于任务的抽象类，可以通过ForkJoinPool来执行。一个ForkJoinTask是类似于线程实体，但是相对于线程实体是轻量级的。大量的任务和子任务会被ForkJoinPool池中的真实线程挂起来，以某些使用限制为代价。
5.Future
5.1 Future概述
Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。
也就是说Future提供了三种功能：
1）判断任务是否完成；
2）能够中断任务；
3）能够获取任务执行结果。
因为Future只是一个接口，所以是无法直接用来创建对象使用的，因此就有了下面的FutureTask。
5.2 Future源码
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;
}
- cancel
cancel方法用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。 -
- 参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返回false；如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。
- isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
- isDone方法表示任务是否已经完成，若任务完成，则返回true；
- get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
- get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null。
5.FutureTask
5.1 FutureTask主要方法
- FutureTask(Callable callable)
- run()
- V get()
- FutureTask源码
public class FutureTask<V> implements RunnableFuture<V>
//FutureTask类实现了RunnableFuture接口，我们看一下RunnableFuture接口的实现：
public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}
public FutureTask(Callable<V> callable) {
}
public FutureTask(Runnable runnable, V result) {
}
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 {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
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);
}
}
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);
}
可以看出RunnableFuture继承了Runnable接口和Future接口，而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。
事实上，FutureTask是Future接口的一个唯一实现类。
5.2.自己实现Future及设计模式
SelfFutureThread
6.源码综合分析
- Callable
@FunctionalInterface
public interface Callable<V> {
V call() throws Exception;
}
- RunnableFuture
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
- Thread
package java.lang;
public class Thread implements Runnable {
public Thread(Runnable target) {
init(null, target, "Thread-" + nextThreadNum(), 0);
}
}
- FutureTask
//FutureTask继承RunnableFuture,RunnableFuture继承Runnable和Future
public class FutureTask<V> implements RunnableFuture<V> {
/*
* 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;//已中断
private Callable<V> callable;
private Object outcome; // non-volatile, protected by state reads/writes
private volatile Thread runner;
private volatile WaitNode waiters;
//构造方法1-传Callable
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
//构造方法2-传Runnable和返回结果-(最终把runnable和返回值转换为callable)
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
public void run() {
/*
//新的刚运行线程就不是new就return
//如果是new: runnerOffset对应本类FutureTask的runner属性,将本线程设置为本类runner(比较前的期望值为null).如果设置为失败也返回.
关于compareAndSwapObject的补充:
[这个方法有四个参数，其中第一个参数为需要改变的对象，第二个为偏移量(即之前求出来的valueOffset的值)，第三个参数为期待的值，第四个为更新后的值。
整个方法的作用即为若调用该方法时，value的值与expect这个值相等，那么则将value修改为update这个值，并返回一个true，
如果调用该方法时，value的值与expect这个值不相等，那么不做任何操作，并范围一个false。]
*/
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
//如果state为new,则执行callable的call方法
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
//发生异常时还要设置异常
setException(ex);
}
//如果正常执行没有发生异常
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,获取值的时候,待处理的可能还没开始进行,也可能进行完了.
public V get() throws InterruptedException, ExecutionException {
int s = state;
//没完成状态,等待
if (s <= COMPLETING)
s = awaitDone(false, 0L);
//s大于COMPLETING,准备返回结果
return report(s);
}
//获取值2
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();
//s大于COMPLETING,准备返回结果
return report(s);
}
//等待线程运行结果
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;
}
//如果执行线程为正在执行中,则当前循环线程放弃cpu执行权等一会,
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//第一次进来,q为空,创建等待队列
else if (q == null)
q = new WaitNode();
//如果q不等于空,则走这里在waiter队列后加一个节点.
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q); //如果设置成功,则queued变为true
else if (timed) {
//如果超时了
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
//没有超时,则等待
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this); //线程等待
}
}
//get调用report返回运行结果值,从outcome中取回返回值
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);
}
//设置异常
protected void setException(Throwable t) {
//将状态从new设置为COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//异常放到outcome
outcome = t;
//再将状态设置为EXCEPTIONAL
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
//设置正常
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v; //设置返回值
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
//叫醒等待的线程节点
private void finishCompletion() {
// assert state > COMPLETING;
//等待等列不为空
for (WaitNode q; (q = waiters) != null;) {
//将等待标志位设置为空(下一个等待可以加入了.)
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
//循环就能将所有的q唤醒了
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,准备将next设置为q,则next没有了.断掉q的next
q = next;//next就成为q了
}
break;
}
}
done();
callable = null; // to reduce footprint
}
//又一个内部类,递归结构(next又是一个WaitNode)
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
}
- Executors
package java.util.concurrent;
public class Executors {
//将runnable转换为callable,FutureTask的构造方法2调用
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
//实现Callable,有call方法.也就是一个callable
//call执行的内容就是runnable的run方法,而返回值就是resutl.
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;
}
}
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
}
7.更多示例
7.1 使用Callable+Future获取执行结果
- ExecutorTask核心代码
ExecutorService executor = Executors.newCachedThreadPool();
MyTask task = new MyTask();
Future<Integer> result = executor.submit(task);
executor.shutdown();
- ExecutorTask
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class ExecutorTask {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
MyTask task = new MyTask();
Future<Integer> result = executor.submit(task);
executor.shutdown();
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
System.out.println("主线程在执行任务");
try {
System.out.println("task运行结果"+result.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
System.out.println("所有任务执行完毕");
}
}
import java.util.concurrent.Callable;
class MyTask implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("子线程在进行计算");
Thread.sleep(3000);
int sum = 0;
for(int i=0;i<100;i++)
sum += i;
return sum;
}
}