AsyncTask源码解析

2018-05-16 本文已影响0人一只胖Wa牛

AsyncTask源码解析

最近再刷一些基础的东西，所以就随便记录了一些看源码的心得，目前开发中见到了很多AsyncTask的使用，今天就来简单的分析下8.0中AsyncTask是如何实现的

简单应用

在AsyncTask的注释中有一段这样的示例代码

private class DownloadFilesTask extends AsyncTask(URL, Integer, Long){
      protected Long doInBackground(URL... urls) {
          int count = urls.length;
          long totalSize = 0;
          for (int i = 0; i < count; i++) {
              totalSize += Downloader.downloadFile(urls[i]);
              publishProgress((int) ((i / (float) count) * 100));
              // Escape early if cancel() is called
              if (isCancelled()) break;
          }
          return totalSize;
      }
 
      protected void onProgressUpdate(Integer... progress) {
          setProgressPercent(progress[0]);
      }
 
      protected void onPostExecute(Long result) {
          showDialog("Downloaded " + result + " bytes");
      }
}

使用也非常简单，只需要创建一个DownloadFilesTask对象然后调用execute即可

new DownloadFilesTask().execute(url1, url2, url3);

那么具体内部是实现呢？那么就来跟进下函数的调用

AsyncTask.execute()

frameworks/base/core/java/android/os/AsyncTask.java

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    //保证串行执行任务
    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
    /**
     * An {@link Executor} that executes tasks one at a time in serial
     * order.  This serialization is global to a particular process.
     */
    public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
    ...
    @MainThread
    public final AsyncTask<Params, Progress, Result> execute(Params... params) {
        return executeOnExecutor(sDefaultExecutor, params);
    }
    ...
}

可以看到AsyncTask的函数execute并没有做什么处理，而是进一步调用了函数executeOnExecutor，并且多传入了一个参数sDefaultExecutor，这个sDefaultExecutor是做什么的呢？他其实就是保证线程串行执行任务的线程池，他的实现如下：

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内部持有一个Runnable的ArrayDeque，用来保证mTasks执行的顺序是串行的，最终还是会交由THREAD_POOL_EXECUTOR去执行具体的Runnable,这个THREAD_POOL_EXECUTOR又是个什么东西呢？实现如下

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    // 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
    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;

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

    private static final BlockingQueue<Runnable> sPoolWorkQueue =
            new LinkedBlockingQueue<Runnable>(128);

    /**
     * An {@link Executor} that can be used to execute tasks in parallel.
     */
    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;
    }
    ...
    /**
     * An {@link Executor} that can be used to execute tasks in parallel.
     */
    public static final Executor THREAD_POOL_EXECUTOR;
    ...
}

他其实就是一个Executor，具体实现是ThreadPoolExecutor，CORE_POOL_SIZE、MAXIMUM_POOL_SIZE会根据cpu的不同而数量不同，这一点从声明出可以看出，其他各项参数参考ThreadPoolExecutor构造方法这里不在细述

到此我们可以发现AsyncTask中声明了两个线程池SerialExecutor和THREAD_POOL_EXECUTOR
其中SerialExecutor用来保证Task串行执行
THREAD_POOL_EXECUTOR才是真正维护执行task的线程池
回到一开始分析的地方接着往下看AsyncTask.executeOnExecutor()方法

AsyncTask.executeOnExecutor()

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private final WorkerRunnable<Params, Result> mWorker;
    private final FutureTask<Result> mFuture;
    ...
    @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;
        //这里可以做初始化UI的工作
        onPreExecute();
        //真正执行的逻辑
        mWorker.mParams = params;
        exec.execute(mFuture);

        return this;
    }
    ...
}

可以看到AsyncTask.executeOnExecutor()先对AsyncTask的状态进行了校验，并随后修改mStatus为Status.RUNNING表示AsyncTask进入进行状态，之后调用onPreExecute()此时由于还在UI线程，onPreExecute是个空实现的函数，这个时候子类可以重写该方法做一些初始化UI的操作
紧接着，会对mWorker.mParams赋值为传入的参数parms，之后会调用exec.execute并将mFuture传入
那么mWorker与mFutrue又是什么呢？那么接下来分别查看其实现

WorkerRunnable

mWorker对应的实现为WorkerRunnable，它的定义如下

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

可以看到虽然它名字是WorkerRunnable，但实际上它是一个Callable。那么mWorker又是什么时候初始化呢？答案是构造方法呀

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    public AsyncTask(@Nullable Looper callbackLooper) {
        ...
        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);
                    ...
                } catch (Throwable tr) {
                    ...
                } finally {
                    postResult(result);
                }
                return result;
            }
        };
        ...
    }
    ...
}

AsyncTask总共有三个构造函数，无论那个最终都会调用到以上这个构造函数
可以看到这里创建了一个WorkerRunnable对象并复制给成员变量mWorker，我没看具体的实现可以看到熟悉的doInBackground与postResult，这两个方法我们先不分析，到后面再说
接下来我们再来看FutureTask的实现

FutureTask

package java.until.concurrent

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

public interface RunnableFuture<V> extends Runnable, Future<V> {
    ...
}

可以看到FutureTask实现了RunnableFuture接口，RunnableFuture接口又同时实现了Runnable接口与Future的接口，那么我们就来看下FutureTask的run()方法是怎么实现的

public class FutureTask<V> implements RunnableFuture<V> {
    ...
    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    ...
    public void run() {
        //CAS校验
        ...
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    ...
                }
                if (ran)
                    set(result);
            }
        } finally {
            ...
        }
    }
}

其实就是调用了成员变量的callable的call然后拿到结果
紧接着我们回到AsyncTask来看下mFuture是怎么初始化的

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    public AsyncTask(@Nullable Looper callbackLooper) {
        ...
        mFuture = new FutureTask<Result>(mWorker) {
            @Override
            protected void done() {
                try {
                    postResultIfNotInvoked(get());
                } catch (InterruptedException e) {
                    ...
                } catch (ExecutionException e) {
                    ...
                } catch (CancellationException e) {
                    postResultIfNotInvoked(null);
                }
            }
        };
    }
}

这里将前面分析的初始化过的WorkerRunnable对象mWorker作为参数创建了一个
FutureTask对象并将其赋值给成员变量mFuture
我们回到AsyncTask.executeOnExecutor，前面分析到这个方法中，将参数赋值params给mWorker的成员变量mParams，然后调用SerialExecutor对象exec的execute方法并发mFuture作为参数传入
这样子流程我们就清晰了，exec的execute会调用mFuture的run方法，然后又会调用mWorker的call，所以就会执行到doInBackground，这个时候是在线程池中之行，所以，doInBackground可以做异步操作
那么为啥onPostExecute是在UI线程接受结果呢？,我们先来看AsyncTask.postResult是如何实现的

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private Result postResult(Result result) {
        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }
    ...
}

这里看到postResult就是向AsyncTask持有的Handler发送了一个类型为MESSAGE_POST_RESULT的消息，并且传入的数据类型为AsyncTaskResult

AsyncTaskResult

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private static class AsyncTaskResult<Data> {
        final AsyncTask mTask;
        final Data[] mData;

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

AsyncTaskResult其实就是存储了当前正在执行的AsyncTask与返回结果Result的数据存储类

AsyncTask持有的Handler是怎么来的呢？

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private final Handler mHandler;
    ...
    public AsyncTask() {
        this((Looper) null);
    }
    
    public AsyncTask(@Nullable Looper callbackLooper) {
        mHandler = callbackLooper == null || callbackLooper == Looper.getMainLooper()
            ? getMainHandler()
            : new Handler(callbackLooper);
        ...
    }

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

可以看到在构造函数中，如果说调用了空参构造就会去调用getMainHandler去初始化一个类型为InternalHandler的handler对象并且赋值给成员变量mHandler，前面说的postResult会发送一个类型为MESSAGE_POST_RESULT的消息，其实就是InternalHandler中进行处理的
注意这里传入InternalHandler构造方法的Looper是Looper.getMainLooper()

InternalHandler

那我们就来看下InternalHandler的实现

    ...
}

AsyncTaskResult其实就是存储了当前正在执行的AsyncTask与返回结果Result的数据存储类

AsyncTask持有的Handler是怎么来的呢？

public abstract class AsyncTask<Params, Progress, Result> {
    ...
    private static class InternalHandler extends Handler {
        public InternalHandler(Looper looper) {
            super(looper);
        }

        @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 void finish(Result result) {
        if (isCancelled()) {
            onCancelled(result);
        } else {
            onPostExecute(result);
        }
        mStatus = Status.FINISHED;
    }
    ...
}

InternalHandler的实现也非常简单，
就是解析msg然后通过AsyncTaskResult拿到对应的AsyncTask之后调用finish函数，紧接着finish中就会根据不同的状态去判断如果说没有取消这个task就会回调onPostExecute
至此整个AsyncTask的执行流程就分析结束了，AsyncTask作为Android系统帮开发者封装好的异步加载数据然后回调UI线程的工具类还是设计的满精巧的，但是也有很多不足之处
在Activity中使用容易导致内存泄漏
不能够及时的取消任务
个人感觉还是Rxjava的实现方式更优雅一些

AsyncTask源码解析

AsyncTask源码解析

简单应用

AsyncTask.execute()

AsyncTask.executeOnExecutor()

WorkerRunnable

FutureTask

AsyncTaskResult

AsyncTask持有的Handler是怎么来的呢？

InternalHandler

AsyncTask持有的Handler是怎么来的呢？

猜你喜欢

热点阅读