美文网首页
Java callable 深入

Java callable 深入

作者: NazgulSun | 来源:发表于2021-06-10 11:01 被阅读0次

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

相关文章

网友评论

      本文标题:Java callable 深入

      本文链接:https://www.haomeiwen.com/subject/neeueltx.html