AsyncTask 源码分析

1. AsyncTask 是 Handler 的一层封装，内部持有 InternalHandler
2. 内部维持一个线程池，并将该线程池转换为串行池
3. 使用 FutureTask 执行
4. 由于 FutureTask 的特点，使的 AsyncTask 的子线程
   1. 可以被取消
   2. 可以获得执行结果
   3. 但是 只能执行一次，原因：FutureTask 一旦计算完成，就不能再重新开始或取消计算。
5. AsyncTask 的线程池可以被外部使用
6. 如果使用 execute(Runnable runnable) 方法，可以发送一个非 FetureTask 的 task
7. 可以切换 线程池的 执行策略 setDefaultExecutor(Executor exec)

public abstract class AsyncTask<Params, Progress, Result> {
    private static final String LOG_TAG = "AsyncTask";

    /*线程池的参数初始化*/

    // We want at least 2 threads and at most 4 threads in the core pool,
    // preferring to have 1 less than the CPU count to avoid saturating
    // the CPU with background work
    // 我们希望核心池中至少有2个线程和最多4个线程，最好是比CPU数少1个线程，以避免使CPU背景饱和
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    private static final int KEEP_ALIVE_SECONDS = 30;// 超出的最长等待时间(s)

    // 线程创建工厂
    private static final ThreadFactory sThreadFactory = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);
        public Thread newThread(Runnable r) {
            return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
        }
    };

    // 此阻塞队列仅保持由 execute 方法提交的 Runnable 任务。
    private static final BlockingQueue<Runnable> sPoolWorkQueue =
            new LinkedBlockingQueue<Runnable>(128);

    public static final Executor THREAD_POOL_EXECUTOR;// 并行化线程池引用

    // 使用 静态代码块 初始化 线程池
    static {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
                sPoolWorkQueue, sThreadFactory);
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        THREAD_POOL_EXECUTOR = threadPoolExecutor;
    }

    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
    public static final Executor SERIAL_EXECUTOR = new SerialExecutor();

    // 将并行化线程池 转化为 串行化,实现机制类似递归调用
    private static class SerialExecutor implements Executor {
        final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
        Runnable mActive;

        public synchronized void execute(final Runnable r) {
            // 向双端队列中添加 Runnable()
            mTasks.offer(new Runnable() { 
                public void run() {
                    try {
                        r.run();// 编写 run 方法，此时并未执行
                    } finally {
                        scheduleNext();// 执行完一次，寻找下一次，并执行，直到双端队列内无数据
                    }
                }
            });
            if (mActive == null) { // 如果是第一次执行execute() 此时 mActive = null,执行
                scheduleNext();
            }
        }
        // 并行化线程池 执行 双端队列中的 Runnable()
        protected synchronized void scheduleNext() {
            if ( (mActive = mTasks.poll())  != null) { // 取出双端队列的末尾 Runnable() 
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }            
    }

    // 设置默认的 线程池 ， 可以抹去 串行化 
    public static void setDefaultExecutor(Executor exec) {
        sDefaultExecutor = exec;
    }


    // 内部Handler 的标记位
    private static final int MESSAGE_POST_RESULT = 0x1;
    private static final int MESSAGE_POST_PROGRESS = 0x2;

    // 内部Handler
    private static InternalHandler sHandler;
    // 执行Handler
    private final Handler mHandler;

    private final WorkerRunnable<Params, Result> mWorker;
    private final FutureTask<Result> mFuture;

    private volatile Status mStatus = Status.PENDING;

    private final AtomicBoolean mCancelled = new AtomicBoolean();
    private final AtomicBoolean mTaskInvoked = new AtomicBoolean();

    // 执行状态
    public enum Status {
        PENDING,
        RUNNING,
        FINISHED,
    }

    public final Status getStatus() {
        return mStatus;
    }

    /*
    构造方法 
    new()
    */
    public AsyncTask() {
        this((Looper) null);
    }

    public AsyncTask(@Nullable Handler handler) {
        this(handler != null ? handler.getLooper() : null);
    }

    public AsyncTask(@Nullable Looper callbackLooper) { 
        mHandler = 
                callbackLooper == null || 
                callbackLooper == Looper.getMainLooper() 
                ? getMainHandler() // looper 为空，looper 为主looper
                : new Handler(callbackLooper); // 如果传入 非主looper
        /*
        使用 FutureTask + Callable 获取执行结果
        */
        mWorker = new WorkerRunnable<Params, Result>() {
            public Result call() throws Exception {
                mTaskInvoked.set(true);
                Result result = null;
                try {
                    // 设置线程 优先级
                    Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
                    // 子线线程的执行方法
                    result = doInBackground(mParams);
                    /*
                    将当前线程中挂起的所有Binder命令刷新到内核驱动程序。
                    在执行一个可能长时间阻塞的操作之前调用这个函数可能很有用，
                    以确保任何悬而未决的对象引用已经被释放，
                    以防止进程持续到超出需要的对象。
                    */
                    // 确保及时释放资源
                    Binder.flushPendingCommands();
                } catch (Throwable tr) {
                    mCancelled.set(true);// 关闭
                    throw tr;
                } finally {
                    postResult(result);// post结果
                }
                return result;// 返回执行的结果
            }
        };

        /*
        可取消的异步计算。
        利用 开始和取消计算 的方法,查询计算是否完成的方法和获取计算结果的方法，
        此类提供了对 Future 的基本实现。
        仅在计算完成时才能获取结果；如果计算尚未完成，则阻塞 get 方法。
        一旦计算完成，就不能再重新开始或取消计算。
        */
        mFuture = new FutureTask<Result>(mWorker) {
            /*
            当此任务转换到状态 isDone（不管是正常地还是通过取消）时，调用受保护的方法。
            默认实现不执行任何操作。
            子类可以重写此方法，以调用完成回调或执行簿记。
            注意，可以查询此方法的实现内的状态，从而确定是否已取消了此任务。
            */
            @Override
            protected void done() {
                try {
                    postResultIfNotInvoked(get());
                } catch (InterruptedException e) {
                    android.util.Log.w(LOG_TAG, e);
                } catch (ExecutionException e) {
                    throw new RuntimeException("An error occurred while executing doInBackground()",
                            e.getCause());
                } catch (CancellationException e) {
                    postResultIfNotInvoked(null);
                }
            }
        };
    }

    private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
        Params[] mParams;
    }

    /*
    mFuture.get() 会抛出如下异常
    CancellationException - 如果计算被取消
    InterruptedException - 如果当前的线程在等待时被中断
    ExecutionException - 如果计算抛出异常
    TimeoutException - 如果等待超时
    */
    public final Result get() throws InterruptedException, ExecutionException {
        return mFuture.get();
    }

    public final Result get(long timeout, TimeUnit unit) throws InterruptedException,
            ExecutionException, TimeoutException {
        return mFuture.get(timeout, unit);
    }

    private void postResultIfNotInvoked(Result result) {
        final boolean wasTaskInvoked = mTaskInvoked.get();
        if (!wasTaskInvoked) {// false
            postResult(result);
        }
    }

    // 发送 子线成 执行的 结果
    private Result postResult(Result result) {
        @SuppressWarnings("unchecked")
        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }


    @WorkerThread
    protected abstract Result doInBackground(Params... params);

    @MainThread
    protected void onPreExecute() {
    }

    @SuppressWarnings({"UnusedDeclaration"})
    @MainThread
    protected void onPostExecute(Result result) {
    }

    @SuppressWarnings({"UnusedDeclaration"})
    @MainThread
    protected void onProgressUpdate(Progress... values) {
    }

    @SuppressWarnings({"UnusedParameters"})
    @MainThread
    protected void onCancelled(Result result) {
        onCancelled();
    }

    @MainThread
    protected void onCancelled() {
    }

    public final boolean isCancelled() {
        return mCancelled.get();
    }

    // 使用 FutureTask 的特性 停止 子线程 的执行
    public final boolean cancel(boolean mayInterruptIfRunning) {
        mCancelled.set(true);
        return mFuture.cancel(mayInterruptIfRunning);
    }


    /*
    execute 方法
    */
    @MainThread
    public final AsyncTask<Params, Progress, Result> execute(Params... params) {
        return executeOnExecutor(sDefaultExecutor, params);
    }

    // 该重载 未执行 FutureTask
    @MainThread
    public static void execute(Runnable runnable) {
        sDefaultExecutor.execute(runnable);
    }

    @MainThread
    public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
                                                                       Params... params) {
        if (mStatus != Status.PENDING) { // 非等待
            switch (mStatus) {
                case RUNNING:// 该 task 正在执行
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task is already running.");
                case FINISHED:// 该 task 只能执行一次
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task has already been executed "
                            + "(a task can be executed only once)");
            }
        }

        mStatus = Status.RUNNING;// 正在执行

        onPreExecute(); // pre

        mWorker.mParams = params;
        exec.execute(mFuture); // 执行 预知好的 FutureTask

        return this;
    }



    @WorkerThread
    protected final void publishProgress(Progress... values) {
        if (!isCancelled()) {
            getHandler().obtainMessage(MESSAGE_POST_PROGRESS,
                    new AsyncTaskResult<Progress>(this, values)).sendToTarget();
        }
    }

    private void finish(Result result) {
        if (isCancelled()) { // cancel
            onCancelled(result);
        } else { // finish
            onPostExecute(result);
        }
        mStatus = Status.FINISHED;// 已完成
    }

    private Handler getHandler() {
        return mHandler;
    }

    /*
    handler
    */
     private static Handler getMainHandler() {
        synchronized (AsyncTask.class) {
            if (sHandler == null) {
                sHandler = new InternalHandler(Looper.getMainLooper());
            }
            return sHandler;
        }
    }

    private static class InternalHandler extends Handler {
        public InternalHandler(Looper looper) {
            super(looper);
        }

        @SuppressWarnings({"unchecked", "RawUseOfParameterizedType"})
        @Override
        public void handleMessage(Message msg) {
            AsyncTaskResult<?> result = (AsyncTaskResult<?>) msg.obj;
            switch (msg.what) {
                case MESSAGE_POST_RESULT:
                    // There is only one result
                    result.mTask.finish(result.mData[0]); // mTask -> AsyncTask
                    break;
                case MESSAGE_POST_PROGRESS:
                    result.mTask.onProgressUpdate(result.mData);
                    break;
            }
        }
    }

    @SuppressWarnings({"RawUseOfParameterizedType"})
    private static class AsyncTaskResult<Data> {
        final AsyncTask mTask;
        final Data[] mData;

        AsyncTaskResult(AsyncTask task, Data... data) {
            mTask = task;
            mData = data;
        }
    }
}