Java callable 深入
java callable/Runnable；
1，相同点，都可以作为线程执行体，给到execute
2， 不同点，callable 可以提供更好的抽象， 与Future 结合使用，可以 获取执行的 结果，抛出的异常，取消执行的任务；
获取执行的结果
Callable 最后会被包装成一个FutureTask，我们通过FutureTask 来获得 结果：
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
@SuppressWarnings("unchecked")
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);
}
我们看到，get 方法是通过识别这个任务的状态码，来决定是 等待，返回，还是抛出异常的；
* 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;
如果是Normal 状态就会返回；
抛出异常
对于状态：
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
我们看到，如下三个状态会抛出异常CancellationException，
对于状态：
private static final int EXCEPTIONAL = 3;
会抛出： ExecutionException，并且有 Throwable x的值，这个值就是 outcome；
我们看一下这个outcome 是在哪里设置的。
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);
}
}
有一个 setException(ex); 方法,完成了 outcome的 赋值。这就是整个异常的处理过程；
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
取消任务
我们先看 cancel 方法：
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
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);
}
}
} finally {
finishCompletion();
}
return true;
}
mayInterruptIfRunning false 则标识不能在运行中打断，则会等待任务完成；
mayInterruptIfRunning true 则是会调用当前线程 的 interrupt 方法。
我们看 get 方法中有一个 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;
}
当我们调用 interupt 之后，
if (Thread.interrupted()) 就会抛出异常； get 方法就会返回。只不过，callable 里面的任务是否能够取消还不一定。
例如，通过下面的代码，我们并不能cancal调 执行的任务：
ExecutorService executor = Executors.newSingleThreadExecutor();
FutureTask<?> futureTask = new FutureTask<String>(new Callable<String>() {
@Override
public String call() throws Exception {
for(int i=0;i<10000;i++){
System.out.println(i);
}
return null;
}
});
executor.execute(futureTask);
System.out.println("futureTask start");
try {
Thread.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
futureTask.cancel(true);
System.out.println("futureTask cancel");
}
futureTask.cancel(true); 会调用 thread.interrupt,我们看看
intterrrupt 方法的注释：只能中断 wait,sleep,join 等方法
* <p> If this thread is blocked in an invocation of the {@link
* Object#wait() wait()}, {@link Object#wait(long) wait(long)}, or {@link
* Object#wait(long, int) wait(long, int)} methods of the {@link Object}
* class, or of the {@link #join()}, {@link #join(long)}, {@link
* #join(long, int)}, {@link #sleep(long)}, or {@link #sleep(long, int)},
* methods of this class, then its interrupt status will be cleared and it
* will receive an {@link InterruptedException}.
可以看到上述代码里并没有这些方法调用，所以无法中断；
正确的做法就是加入一个 是否调用了 interrupted 的判断：
Thread.currentThread().isInterrupted()
public String call() throws Exception {
for(int i=0;i<10000&&!Thread.currentThread().isInterrupted();i++){
System.out.println(i);
}
return null;
}
});
