AsyncTask源码分析

突然发现自己开发其实很少用到AsyncTask⁄(⁄ ⁄•⁄ω⁄•⁄ ⁄)在这里补上对它的源码分析

初见

在Android启蒙导师黑马程序员的视频中曾经听到过一句话，AsyncTask：轻量级的线程工具，方便快捷的实现工作线程到主线程的通信。
在翻看AsyncTask源码时一部分注释如下

/**
 * <p>AsyncTask enables proper and easy use of the UI thread. This class allows you
 * to perform background operations and publish results on the UI thread without
 * having to manipulate threads and/or handlers.</p>
 *
 * <p>AsyncTask is designed to be a helper class around {@link Thread} and {@link Handler}
 * and does not constitute a generic threading framework. AsyncTasks should ideally be
 * used for short operations (a few seconds at the most.) 

大意呢就是AsyncTask可以允许你在工作线程中发布结果到UI线程，但是它适合短时间操作最多几秒钟，嗨哟，这么傲娇，走了走了，还是滚去用RxJava算了，HandlerThread也行。
AsyncTask：别看注释吓人，我这样都没被淘汰，亲儿子啊~~
那我们再留下看看他还有什么话要说(;¬_¬)

使用

可以看到AsyncTask需要传入3个泛型参数
public abstract class AsyncTask<Params, Progress, Result>
依次代表传入工作线程的参数，进度回调，回调给UI线程的结果，艾玛，都有进度回调，这么一看好像很NB的样子。
我们随便写个AsyncTask看看

TestTask task = new TestTask();
task.execute(1, 2, 3, 4, 5, 6);
/**
     * 测试
     */
    class TestTask extends AsyncTask<Integer, Integer, String> {
        private static final String TAG = "TestTask";

        @Override
        protected void onPreExecute() {
            //预处理(UI线程)
        }

        @Override
        protected String doInBackground(Integer... params) {
            //必须实现，异步耗时任务(工作线程)
            StringBuilder sb = new StringBuilder();
            for (int i : params) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                sb.append(i);
            }
            Log.d(TAG, "doInBackground: " + sb.toString());
            return sb.toString();
        }

        @Override
        protected void onProgressUpdate(Integer... values) {
            //进度回调（UI线程）
            Log.d(TAG, "onProgressUpdate: " + values[0]);
        }

        @Override
        protected void onPostExecute(String s) {
            //回调给UI线程(UI线程)
            Log.d(TAG, "onPostExecute: " + s);
        }
    }

过了一会果然输出了2行日志，进度什么的还是要自己来搞
doInBackground: 123456
onPostExecute: 123456

分析

那么我们来看看execute这个方法是在怎么启动的

//UI线程限定
 @MainThread
    public final AsyncTask<Params, Progress, Result> execute(Params... params) {
        return executeOnExecutor(sDefaultExecutor, params);
    }

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

        mStatus = Status.RUNNING;

        onPreExecute();

        mWorker.mParams = params;
        exec.execute(mFuture);

        return this;
    }

看到这里，我们有了大概的思路，execute方法内部调用了默认的执行方法来执行，相对状态进行校验过后，将参数赋值后，执行方法，我们接下来先看看参数是在怎么赋值的，
mWorker.mParams = params;

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

可以发现mWorker其实就就是简简单单一个实现Callback接口的静态内部类，里面存放参数，那么exec.execute(mFuture);呢，execute传递进去了一个默认的执行规则

public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
//execute 默认
private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;

其中Executor是一个接口

public interface Executor {
    void execute(Runnable command);
}

SerialExecutor，翻译出来叫做串行执行，我们来瞧瞧TA的实现

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();
                    } finally {
                        //不考虑执行结果如何都会执行下一个任务
                        scheduleNext();
                    }
                }
            });
            if (mActive == null) {
                //第一次从这里开始执行任务
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext() {
            //从队列取出一个任务
            if ((mActive = mTasks.poll()) != null) {
                //真正开始执行
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }
    }

看到这里，敢情SerialExecutor就是THREAD_POOL_EXECUTOR的封装版，保证里面的runnable都是串行执行，名至实归，看到这里，笔者突然发现小小的AsyncTask中蕴含了不少的东西，那么我们完整的看下AsyncTask的构造器和成员变量

深入

//API25
public abstract class AsyncTask<Params, Progress, Result> {
    //手机CPU数目
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    //根据CPU来配置核心线程数目，每个编译版本可能不一样
    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;
    //空线程30s保活
    private static final int KEEP_ALIVE_SECONDS = 30;
    //配置线程池时，给每个线程名字命名
    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());
        }
    };
    //阻塞队列，限长128
    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;
    }

    public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
    //消息的what
    private static final int MESSAGE_POST_RESULT = 0x1;
    private static final int MESSAGE_POST_PROGRESS = 0x2;

    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
    //回调的Handler
    private static InternalHandler sHandler;
    //实现Callback，存放参数
    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();

    private static class SerialExecutor implements Executor {
        //上文已有，省略。。。
    }

    public enum Status {
        //等待执行
        PENDING,
        //执行中
        RUNNING,
        //结束
        FINISHED,
    }


    public AsyncTask() {
        mWorker = new WorkerRunnable<Params, Result>() {
            public Result call() throws Exception {
                //任务执行
                mTaskInvoked.set(true);
                Result result = null;
                try {
                    //设置线程的优先级
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
                    //noinspection unchecked
                    //执行方法
                    result = doInBackground(mParams);
                    Binder.flushPendingCommands();
                } catch (Throwable tr) {
                    mCancelled.set(true);
                    throw tr;
                } finally {
                    //结果post
                    postResult(result);
                }
                return result;
            }
        };
        //相当于修饰者模式
        mFuture = new FutureTask<Result>(mWorker) {
            @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);
                }
            }
        };
    }

每处地方笔者几乎都打上了注释，毕竟我就是那么的贴心，恩(°∀°)ﾉ，其中有几处细节
AsyncTask的线程池配置根据手机配置来的，缓冲队列默认是128，也就是说如果有一只4核的手机，那么根据如上API25的配置

• CPU_COUNT=4
• CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4)) = 4
• MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1 = 7

线程池中最大容纳7个线程，同时处理4个工作线程，满出来的放到BlockingQueue中，也就是128的数量限制，默认还是串行执行，如果工作线程过于密集，不仅会阻塞后续的线程，还有可能抛出RejectedExecutionException异常，被线程池拒绝，至此，前文的AsyncTask的注释提示适合短时间的任务也就揭开了面纱。
当后台任务完成会调用这个方法返回结果

 private Result postResult(Result result) {
        @SuppressWarnings("unchecked")
        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }

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

        @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]);
                    break;
                case MESSAGE_POST_PROGRESS:
                    result.mTask.onProgressUpdate(result.mData);
                    break;
            }
        }
    }

private static class AsyncTaskResult<Data> {
        final AsyncTask mTask;
        final Data[] mData;

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

通过InternalHandler来分发消息，然后对应执行mTask的相关回调方法

总结

综上所诉，AsyncTask驱动的其实是mFuture，本质也是一个Runnable；进度回调由我们自己调用publishProgress，工作线程限定

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

一遍源码看下来，AsyncTask其实是将一个异步任务分配到线程池中来执行，完成后通过Handler来分发的一个封装的库，由于默认的执行还是串行的，虽然支持多线程并行，但是不太适合高频的网络请求，适用于一些轻量的IO操作。
Ps：由于使用过程中带有一些坑，就偷懒不放调用流程图了

参考：
AsyncTask和AsyncTaskCompat源码解析
Android AsyncTask 源码解析

凡劳苦担重担的人可以到我这里来，我就使你们得安息。 (马太福音 11:28 和合本)