面试系列之AsyncTask源码深入解读

前言

源码

梳理

前言

安卓中经常会涉及到多线程问题，一般多用Handler+Thread的组合来解决。源码中对其进行了精妙的封装，于是有了AsyncTask。相对于Handler+Runnable来讲，AsyncTask开启的异步，过程可控、结果可知，使用的线程池技术并发性能好，封装优势明显，使用起来也比较简单，但在多个线程共同控制一个UI时，就表现的繁琐不便。而Handler+Thread则相对灵活，执行效率较高。具体的比较，可参看这篇博文http://www.open-open.com/lib/view/open1413796240122.html

源码

AsyncTask的源码并不多，这里去掉所有的注释，逐个来看一下。

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

//cpu个数，创建线程池一贯的做法
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;

//long型，表示，如果线程的数量多于核心池大小，在新任务来临之前，线程池允许线程的空闲时间，超过这个时间，如果仍在空闲，线程即被终止
private static final int KEEP_ALIVE = 1;

//线程工厂
private static final ThreadFactory sThreadFactory = new ThreadFactory() {
//原子操作，避免i++或者++i的线程不安全操作。
private final AtomicInteger mCount = new AtomicInteger(1);
//返回一个线程，并命名
public Thread newThread(Runnable r) {
return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
}
};
//拥塞队列，长度128,暂存由execute方法提交的、等待执行的线程
private static final BlockingQueue<Runnable> sPoolWorkQueue =
new LinkedBlockingQueue<Runnable>(128);

//初始化线程池
public static final Executor THREAD_POOL_EXECUTOR
= new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE,
TimeUnit.SECONDS, sPoolWorkQueue, sThreadFactory);
//实现Executor，加同步锁，让线程有序执行，one by one
public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
//message标志位，结果
private static final int MESSAGE_POST_RESULT = 0x1;
//message标志位，进度
private static final int MESSAGE_POST_PROGRESS = 0x2;
//sDefaultExecutor指向SERIAL_EXECUTOR
private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
//处理进程消息
private static InternalHandler sHandler;
//Work线程，实现Callable接口，即给线程加上返回值
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 {
//双端队列，存储线程
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);
}
}
}

//枚举类，线程的状态
public enum Status {
//待执行
PENDING,
//正在执行
RUNNING,
//已完成
FINISHED,
}
//返回处理消息的Handler
private static Handler getHandler() {
synchronized (AsyncTask.class) {
if (sHandler == null) {
sHandler = new InternalHandler();
}
return sHandler;
}
}
//设置默认的Executor,可自定义Executor工厂
public static void setDefaultExecutor(Executor exec) {
sDefaultExecutor = exec;
}
//构造方法
public AsyncTask() {
//创建work抽象类，实质是Callback接口
mWorker = new WorkerRunnable<Params,
Result>() {
public Result call() throws Exception {
mTaskInvoked.set(true);
//设置线程优先级                 android.os.Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
//noinspection unchecked
//真正的耗时处理
Result result = doInBackground(mParams);
//释放线程阻塞而占用的对象
Binder.flushPendingCommands();
//返回处理结果
return postResult(result);
}
};
//WorkerRunnable做参数，创建FutureTask线程
mFuture = new FutureTask<Result>(mWorker) {
@Override
protected void done() {
try {
//FutureTask过程可控，这里对是否Invoke做以判断
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 void postResultIfNotInvoked(Result result) {
final boolean wasTaskInvoked = mTaskInvoked.get();
if (!wasTaskInvoked) {
postResult(result);
}
}

//handler发送消息，处理结果
private Result postResult(Result result) {
@SuppressWarnings("unchecked")
Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
new AsyncTaskResult<Result>(this, result));
message.sendToTarget();
return result;
}

//返回当前线程状态
public final Status getStatus() {
return mStatus;
}

//后台的耗时操作
@WorkerThread
protected abstract Result doInBackground(Params... params);

//UI线程，预处理，doInBackground()之前执行
@MainThread
protected void onPreExecute() {
}

//UI线程，处理线程执行结果
@SuppressWarnings({"UnusedDeclaration"})
@MainThread
protected void onPostExecute(Result result) {
}

//更新进度，常与publishProgerss()配合使用
@SuppressWarnings({"UnusedDeclaration"})
@MainThread
protected void onProgressUpdate(Progress... values) {
}

//cancel()调用之后的回调
@MainThread
protected void onCancelled() {
}

//如果在任务正常结束之前取消任务成功则返回true，否则返回false
public final boolean isCancelled() {
return mCancelled.get();
}

//尝试取消线程，如果已经开始执行、或者已被取消、或者因某些原因不能取消，则会返回false。
public final boolean cancel(boolean mayInterruptIfRunning) {
mCancelled.set(true);
return mFuture.cancel(mayInterruptIfRunning);
}

//等待计算结束并返回结果
public final Result get() throws InterruptedException, ExecutionException {
return mFuture.get();
}
//等待计算结束并返回结果,最长等待时间为：timeOut
public final Result get(long timeout, TimeUnit unit) throws InterruptedException,
ExecutionException, TimeoutException {
return mFuture.get(timeout, unit);
}

//方法比较有意思，执行线程，并将自己返回回来。早期的版本是由
//一个单一的后台线程来调度队列中的任务，后来才改成线程池。
@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;

//也由此可见AsyncTask中最先执行的方法是onPreExecute
onPreExecute();

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

return this;
}

//静态方法，AsyncTask执行一个线程，我个人用的比较少。
@MainThread
public static void execute(Runnable runnable) {
sDefaultExecutor.execute(runnable);
}

//可在doInBackground()中调用，传入进度参数
@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()) {
onCancelled(result);
} else {
onPostExecute(result);
}
mStatus = Status.FINISHED;
}

//定义的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;
}
}
}

//静态内部类WorkerRunnable，主要实现Callable接口，可返回线程的执行结果
private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}

//静态内部类，AsyncTask就不用说了，Data[]主要来记录执行进度的
@SuppressWarnings({"RawUseOfParameterizedType"})
private static class AsyncTaskResult<Data> {
final AsyncTask mTask;
final Data[] mData;

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

}

源码已经给出了比较详细的注释，可参考api自行解读。

梳理

从上面的源码不难看出，AsyncTask源码结果主要分为两部分：一部分负责与UI交互；一部分负责线程的调度。

与UI交互自然用到了Handler，事实上在源码中很多地方都能看到它的身影。而在AsyncTask里面定义了两种消息类型MESSAGE_POST_PROGRESS和MESSAGE_POST_RESULT，意思很明显，一种处理进度，一种处理结果。这一部分没有太多可述之处。

线程调度是整个AsyncTask的关键。

我们已经看到，除了最早期的版本（SDK=4）是使用单个线程来处理任务之外，之后的多任务调度，AsyncTask全部委托给线程池来处理。依赖于SDK版本的不同，内部线程池的演化又经历这样两个重要变更：

sdk 11前，使用内部的线程池，多线程并发执行。线程池大小等于5，最大达128；

sdk 11<=current version <= sdk 18时 线程池大小等于5，最大达128，使用默认的serial线程池，执行完一个线程，再顺序执行下一个线程；

sdk 19后线程池大小调整为更合理的 cpu count + 1，最大值为cpu count * 2 + 1，依然默认serial线程池；

也就是说目前的AsyncTask里面存在两种线程池，源码中已经很明显地写出来了，THREAD_POOL_EXECUTOR 和 SERIAL_EXECUTOR。默认使用的是SERIAL_EXECUTOR，注释中已经提到，它是one by one的执行方式，只有等当前线程执行完毕，线程池才会invoke下一个线程，否则只有等待。所以如果当你在自己应用中发现：调用AsyncTask来处理多个异步任务时，自己的doInBackground()方法，在execute()之后，等了半天竟然没有执行！你不要怪他，那是因为人家本来就是这样的啦。

有同学可能要问，这样做，岂不是没有达到并发的目的？事实确实这样，SERIAL_EXECUTOR是google在安卓3.0的AsyncTask引入的重要更新。在此以前，线程最大容纳量是5，就是说当有6个线程并发执行时，只有5个在执行，另外一个在阻塞或者等待状态。在此3.0以后，引入SERIAL_EXECUTOR来控制拥塞队列（即sPoolWorkQueue），实现默认按先入先出的顺序，串行执行。为了避免并行变成串行，google同时也开放了setDefaultExecutor()方法，使用者可通过它定义自己的线程池，来实现个性化的并发控制。就是说你想定义多大容量都可以，只要开心就好。

另外THREAD_POOL_EXECUTOR也说一下。

google最初对它的设定容量是5，后来与时俱进，改成动态设定的了。

使用AsyncTask时，如果你觉得自己定义线程池的开销太大或者太麻烦，就考虑THREAD_POOL_EXECUTOR吧，根据你cpu的个数来动态设定线程池的大小，合理地规划cpu资源。虽然这种方式不能无限并发，但至少不用等到一个结束了才会执行另一个，关键是用起来简单方便。

总结

面试中经常会问到，AsyncTask能不能在子线程中执行？

源码中我们能看到，AsyncTask中有大量的run on main thread注释，比如onPreExecute( )，onProgressUpdate（), onCancelled( ), execute( )等方法，毫无疑问必须在UI线程中触发，但是像这种轻便型的开启异步的方法static void execute（Runnable runnable），就未必一定要在main thread中执行了。

虽然相比于Thread+Handler的异步处理方式，AsyncTask经过谷歌的完美封装，使用起来有着无与伦比的优势。但有了这些分析，以后在代码中使用AsyncTask时就要慎重一下了：我这里可能都有几个并发任务，使用它会不会造成阻塞等待？

这里给出两点个人建议：

如果涉及与UI交互，优先考虑AsyncTask，并发数量较大，最好定义自己的线程池。否则请使用Thread来处理；

如果要求异步任务马上执行，可考虑Thread，或者自定义线程池，值得注意的是维护一个线程池的开销，特别是池的大小不加限制的话；

优化程序结构，尽量少用异步。

个人理解，仅供参考，欢迎留言拍砖，转载请注明出处。