FUTURES 异步使用

Runnable方式

Runnable方式，通过实现Runnable，完成多线程

val runnable = Runnable {
    Log.d("MainActivity",Thread.currentThread().name)
}
val thread = Thread(runnable)
thread.run()

Futures方式

Callable接口和Runnable接口类似，只不过有返回值，可以抛出受检异常。

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

Future接口

image.png

get() - 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成
get(Long timeout , TimeUnit unit) - 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的timeout时间，该方法将返回null
isDone() - 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回true
isCanceller() 如果任务完成前被取消，则返回true
cancel(boolean mayInterruptRunning) - 取消任务，cancel(false) 等待任务执行完毕然后回收资源，cancel(true) 尝试直接终中断任务

RunnableFuture继承Future

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

FutureTask

UML图

futuretask.png

第一个接收一个Callable参数。
第二个接收一个Runnable和一个任务成功执行完毕所需要的返回值作为参数。可以看到实际把Runnable包装成了Callable。

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        此处传递callable其实调用call()方法的时候，内部调用runnable的run方法
        this.state = NEW;       // ensure visibility of callable
    }

新建一个FutureTask

val futureTask = FutureTask(Callable<String> {
            Log.d("MainActivity",Thread.currentThread().name+"====test")
            Thread.sleep(10000)
            "test"
        })
val futureThread = Thread(futureTask)
futureThread.start()
//阻塞线程,此处会阻塞UI线程
//futureTask会保存结果
futureTask.get()
Log.d("MainActivity",Thread.currentThread().name+"===")

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);
                    //保存结果
            }

set()方法

set方法保存结果，同时唤醒当前线程

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //保存执行结果
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        //唤醒当前线程。
        finishCompletion();
    }
}

/**
 * Removes and signals all waiting threads, invokes done(), and
 * nulls out callable.
 */
private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            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
                q = next;
            }
            break;
        }
    }
    done();
    callable = null;        // to reduce footprint
}

get方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        //阻塞当前线程
        s = awaitDone(false, 0L);
    //当线程从阻塞状态被唤醒，将结果返回。
    return report(s);
}