你真的了解AsyncTask吗？AsyncTask源码分析

转载请注明出处：http://blog.csdn.net/yianemail/article/details/51611326

1，概述
Android UI是线程不安全的，如果想要在子线程很好的访问ui，就要借助Android中的异步消息处理机制
http://blog.csdn.net/yianemail/article/details/50233373
通过Thread 执行耗时操作，通常利用Handler 发送消息给ui线程。
这种方式代码相对臃肿，并且不能对多任务执行很好的控制。

为了简化操作，Android 1.5提供了更加轻量级的AsyncTask异步工具类，使创建异步任务变得异常简单。

2，AsyncTask 的使用

看一个最典型的AsyncTask 简单使用场景

package com.listenread.luhuanju.helloword;import android.os.AsyncTask;import android.support.v7.app.AppCompatActivity;import android.os.Bundle;public class MainActivity extends AppCompatActivity {    @Override    protected void onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentView(R.layout.activity_main);        //AsyncTask 开始执行异步任务        new MyAsyncTask().execute();    }    class MyAsyncTask extends AsyncTask<Void, Void, String> {        @Override        protected String doInBackground(Void... params) {            //Commen is request resurce from http and return to onPostExecute            //模拟数据返回            return "mStr";        }        @Override        protected void onPostExecute(String s) {            super.onPostExecute(s);            //获取数据        }    }}

3，源码解析

既然代码入口

new MyAsyncTask().execute();

我们就从它的构造函数分析，

  /** * Creates a new asynchronous task. This constructor must be invoked on the UI thread. * 方法注释就说明了必须要在ui thread 进行AsyncTask调用 */    public AsyncTask() {        mWorker = new WorkerRunnable<Params, Result>() {            public Result call() throws Exception {                mTaskInvoked.set(true);                Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);                //noinspection unchecked                Result result = doInBackground(mParams);                Binder.flushPendingCommands();                return postResult(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);                }            }        };

来分析一下构造函数实例化的两个变量
mWorker 以及 mFuture

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

可以看到mWorker 是实现了参数是泛型的 Callable接口，看下Callable接口的声明

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

只有一个返回类型为泛型的 Call() 方法。

这里介绍一下Callable 这个接口，在java中，实现多线程的方法除了Runnable之外，还有Callable,Future 以及FutureTask(没错，就是一会要分析的这货)，与Runnable不同的是，这几个类型都只能在线程池当中实现。Callable 与 Runnable 最关键的异点是Callable 可以有返回值。然而Runnable 是拿不到执行的返回值的。
这也就再一次解释了

public interface Callable<V> {    V call() throws Exception; //返回值泛型v}

对比Runnable的接口声明

public interface Runnable {    public void run(); //无返回值}

再来看一下mFuture 这货，可以看到把mWorker当作参数

public class FutureTask<V> implements RunnableFuture<V> {

可以看到FutureTask 实现了RunnableFuture< V>接口

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

而RunnableFuture 继承了Runnable 以及Future
终于到了Future了，继续看Future 的声明

public interface Future<V> { /** * ...忽略方法的注释声明 */    boolean cancel(boolean mayInterruptIfRunning);    boolean isCancelled();    boolean isDone();    V get() throws InterruptedException, ExecutionException;    V get(long timeout, TimeUnit unit)        throws InterruptedException, ExecutionException, TimeoutException;}

看到这我们大概明白了AsyncTask的一大优势所在，没错，就是可控。
总结一下FutureTask的作用，FutureTask作为一个可管理的异步任务，使得在线程池中执行的异步任务可以被更精准的控制。

分析完构造函数中两个实例化的变量，在继续看execute()方法

    public final AsyncTask<Params, Progress, Result> execute(Params... params) {        return executeOnExecutor(sDefaultExecutor, params);    }

execute()继续调用了executeOnExecutor(sDefaultExecutor, params) ，并且传递两个参数sDefaultExecutor以及params

我们看下sDefaultExecutor

    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;

   public static final Executor SERIAL_EXECUTOR = new SerialExecutor();

所以我们只需要看下SerialExecutor 的实现即可；

  private static class SerialExecutor implements Executor {        final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();        Runnable mActive;        public synchronized void execute(final Runnable r) {            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就是一个线程池， AsyncTask 的内部类，实现了Executor 接口

public interface Executor {    /**     * Executes the given command at some time in the future.  The command     * may execute in a new thread, in a pooled thread, or in the calling     * thread, at the discretion of the {@code Executor} implementation.     *     * @param command the runnable task     * @throws RejectedExecutionException if this task cannot be     * accepted for execution     * @throws NullPointerException if command is null     */    void execute(Runnable command);}

我们看下这句方法的注释就能明白execute()的作用就是在线程池中分配线程执行异步任务。那么execute()方法在AsyncTask 到底执行什么任务呢？没错，就是

 public synchronized void execute(final Runnable r) {            mTasks.offer(new Runnable() {                public void run() {                    try {                        r.run();                    } finally {                        scheduleNext();                    }                }            });            if (mActive == null) {                scheduleNext();            }        }

我们发现了 mTasks.offer 这个方法，看下mTasks

final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();

简单说一下ArrayDeque

ArrayDeque是一个双端队列，它具有：

数组双端队列没有容量限制，使他们增长为必要支持使用。

不支持多线程并发访问。

它们要比堆栈Stack和LinkedList快。

看下它的offer（）

   /**     * Inserts the specified element at the end of this deque.     * 这句注释的意思就是说添加特定的元素（就是Runnable）到队尾     * <p>This method is equivalent to {@link #offerLast}.     *     * @param e the element to add     * @return <tt>true</tt> (as specified by {@link Queue#offer})     * @throws NullPointerException if the specified element is null     */    public boolean offer(E e) { return offerLast(e);    }

调用的 offerLast(e)

    public boolean offerLast(E e) {        addLast(e);        return true;    }

继续addLast（e）

    public void addLast(E e) {        if (e == null)            throw new NullPointerException("e == null");        elements[tail] = e;        if ( (tail = (tail + 1) & (elements.length - 1)) == head)            doubleCapacity();    }

 /** * Double the capacity of this deque. Call only when full, i.e., * when head and tail have wrapped around to become equal. */    private void doubleCapacity() {        // assert head == tail;        int p = head;        int n = elements.length;        int r = n - p; // number of elements to the right of p        int newCapacity = n << 1;        if (newCapacity < 0)            throw new IllegalStateException("Sorry, deque too big");        Object[] a = new Object[newCapacity];        System.arraycopy(elements, p, a, 0, r);        System.arraycopy(elements, 0, a, r, p);        elements = a;        head = 0;        tail = n;    }

简单分析一下addLast（e）的作用吧，如果传入参数为空，抛出异常
如果此时队列已满，就执行doubleCapacity（）；而这个方法就是把ArrayDeque扩容的，通过

       ...        int r = n - p; // number of elements to the right of p        int newCapacity = n << 1;//二进制，左移        ...        Object[] a = new Object[newCapacity];        System.arraycopy(elements, p, a, 0, r);        System.arraycopy(elements, 0, a, r, p);       ...

我们可知，满了就要扩容一倍的

说了这么多，我们重新回顾一下SerialExecutor

private static class SerialExecutor implements Executor {        final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();        Runnable mActive;        public synchronized void execute(final Runnable r) {            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);            }        }    }

分析一下它的执行过程
第一次运行的时候，毫无疑问，局部变量mActive为null，执行scheduleNext（）方法

  protected synchronized void scheduleNext() {            if ((mActive = mTasks.poll()) != null) {                THREAD_POOL_EXECUTOR.execute(mActive);            }        }

又看到了 THREAD_POOL_EXECUTOR

    public static final Executor THREAD_POOL_EXECUTOR            = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE,                    TimeUnit.SECONDS, sPoolWorkQueue, sThreadFactory);

THREAD_POOL_EXECUTOR 也是一个线程池，通过构造函数初始化了
核心线程数，超时时长，所容纳的最大线程数，等一系类配置。在我的sdk版本好像对之前4.x做了优化，并没有固定写死，因为是这样的

  private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();    private static final int CORE_POOL_SIZE = CPU_COUNT + 1;    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;    private static final int KEEP_ALIVE = 1;

sDefaultExecutor 线程池内部维护一个任务队列（ArrayDeque）,
execute()方法，将Runnable放入队尾，第一次执行走scheduleNext()，此时 mTasks.poll() 取出队首任务，不为空则传入THREAD_POOL_EXECUTOR进行执行。

4，执行过程回顾
再来看一下 AsyncTask.execute()的执行过程

 @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();  //熟悉这个onPreExecute（）方法吗？        mWorker.mParams = params;//外部参数赋值给mWorker局部变量        //最主要的逻辑就在这个方法了,提交mFuture执行，封装了worker,        exec.execute(mFuture);        return this;    }

mFuture 是需要mWorker 作为参数，而mWorker的执行

  mWorker = new WorkerRunnable<Params, Result>() {            public Result call() throws Exception {                mTaskInvoked.set(true);                 //提升线程的优先级               Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);                //noinspection unchecked                //看到了吗？这里的doInBackground 执行的就是你外部调用的 doInBackground（）的逻辑。                Result result = doInBackground(mParams);                Binder.flushPendingCommands();                return postResult(result);            }        };

拿到结果调用postResult(result)

   private Result postResult(Result result) {        @SuppressWarnings("unchecked")        //getHandler（）实际上就是 super(Looper.getMainLooper());，也就是Main 中的looper        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,                new AsyncTaskResult<Result>(this, result));        message.sendToTarget();        return result;    }

AsyncTaskResult就是一个简单的携带参数的对象。
看到Messager 肯定会有Handler，于是找到

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

result.mTask.finish(result.mData[0]) 其实就是调用了AsyncTask类中的finish方法

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

调用了cancel()则执行onCancelled回调；正常执行的情况下调用我们的onPostExecute(result);最后将状态置为FINISHED。

再回到mFuture

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

执行 postResultIfNotInvoked(get());

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

根据 wasTaskInvoked状态判断是否执行postResult(result);
在mWorker 中我们已经知道 wasTaskInvoked.set(true)
所以该方法不会执行。

至此，AsyncTask源码分析完毕，相信大家对AsyncTask有了更深的理解~~~

更多相关文章

随机推荐