本文分析AsyncTask，主要通过源码走一下流程。
先给出API中的一个示例：
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");
}
}
new DownloadFilesTask().execute(url1, url2, url3);
Task实例必须在UI线程创建，并在UI线程调用execute方法。
从HONEYCOMB开始，tasks在一个单线程中顺序执行，如果想要并行执行，可以调用executeOnExecutor(java.util.concurrent.Executor, Object[])}。AsyncTask定义了一个线程池：
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 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;
从代码中可以看到，线程池的CorePoolSize大小为2~4。
通常我们是new一个AsyncTask的实例，然后调用它的execute方法。
private final WorkerRunnable<Params, Result> mWorker;
private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}
private final AtomicBoolean mTaskInvoked = new AtomicBoolean();
private final FutureTask<Result> mFuture;
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 {
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);
}
}
};
}
@MainThread
public final AsyncTask<Params, Progress, Result> execute(Params... params) {
return executeOnExecutor(sDefaultExecutor, params);//调用默认的Executor去执行任务
}
private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
@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)");//这里可以看到一个task只能被执行一次
}
}
mStatus = Status.RUNNING;
onPreExecute();//这是一个空方法，可以被子类复写，实现自定义的逻辑
mWorker.mParams = params;//将参数传递给mWorker
exec.execute(mFuture);//调用Executor的execute方法，上面看到传过来的是默认的Executor sDefaultExecutor
return this;
}
private static class SerialExecutor implements Executor {
final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
Runnable mActive;
public synchronized void execute(final Runnable r) {//这里传进来的的是mFuture
mTasks.offer(new Runnable() {
public void run() {
try {
r.run();
} finally {
scheduleNext();
}
}
});//new 一个Runnable并加入队列mTasks
if (mActive == null) {//第一次这里是null
scheduleNext();
}
}
//,第二个AsyncTask调用execute时，把任务加入到mTasks队列，等待上个任务执行完成后，再从队列中poll出来，执行，所以默认情况下，多个AsyncTask是顺序执行的
protected synchronized void scheduleNext() {
if ((mActive = mTasks.poll()) != null) {//从队列mTasks中获取刚才加入队列的Runnable并赋值给mActive
THREAD_POOL_EXECUTOR.execute(mActive);//调用线程池执行这个Runnable
}
}
}
ThreadPoolExecutor.execute方法比较复杂，暂时跳过，我们只要知道
try {
r.run();
} finally {
scheduleNext();
}
这一段代码将会在线程池的某个线程中去执行，这里的r就是mFutrue。
public class FutureTask<V> implements RunnableFuture<V>
public interface RunnableFuture<V> extends Runnable, Future<V>
FutureTask实现了RunnableFuture，而RunnableFuture就是Runnable和Futrue的组合（这是多重继承？）。
下面看一下FutrueTask的run方法
//忽略了部分代码
try {
Callable<V> c = callable;//这里的callable就是mWorker
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();//mWorker.call()
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
private void finishCompletion() {
...... //忽略部分代码
done(); //这里回调了done方法
callable = null; // to reduce footprint
}
所以，mFutrue的run方法简单来说就是调用了mWorker的call方法，如果成功，则再调用done方法。
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);//这里就回调了doInBackground方法，mParams是在executeOnExecutor方法中赋值的。
Binder.flushPendingCommands();
} catch (Throwable tr) {
mCancelled.set(true);
throw tr;
} finally {
postResult(result);
}
return result;
}
};
private Result postResult(Result result) {
@SuppressWarnings("unchecked")
Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
new AsyncTaskResult<Result>(this, result));
message.sendToTarget();
return result;
}
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]);//这里调用了AsyncTask的finish方法，并将doInBackground返回的Result传递给finish方法
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;
}
}
private void finish(Result result) {
if (isCancelled()) {//如果已经请求取消，
onCancelled(result);
} else {
onPostExecute(result);//否则，回调onPostExecute
}
mStatus = Status.FINISHED;
}
然后，继续mFutrue的run方法，会回调它的done方法。
mFuture = new FutureTask<Result>(mWorker) {
@Override
protected void done() {
try {
postResultIfNotInvoked(get());//这里调用Future的get方法，与set对应，拿到Result（细节较复杂）
} 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) {//如果mWorker的的call方法没有执行
postResult(result);
}
}
最后scheduleNext(),如果有其他任务在队列中，就poll出任务并执行。
简单总结一下，在主线程调用execute会回调onPreExecute，然后在线程池中通过mWorker回调doInBackground，
最后通过handler发送消息给主线程，在主线程回调onPostExecute。
@WorkerThread
protected final void publishProgress(Progress... values) {
if (!isCancelled()) {
getHandler().obtainMessage(MESSAGE_POST_PROGRESS,
new AsyncTaskResult<Progress>(this, values)).sendToTarget();
}
}
在doInBackground中调用publishProgress，实际也是通过Handler将消息发送到主线程，在主线程回调onProgressUpdate
case MESSAGE_POST_PROGRESS:
result.mTask.onProgressUpdate(result.mData);
break;
AsyncTask提供了cancel方法
public final boolean cancel(boolean mayInterruptIfRunning) {
mCancelled.set(true);//设置为true， isCancelled将返回true
return mFuture.cancel(mayInterruptIfRunning);
}
FutrueTask的cancel方法,部分代码
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();//调用Thread的interrupt方法
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
如果需要让多个Task并行执行，可以调用executeOnExecutor，并将THREAD_POOL_EXECUTOR作为参数传入。
关于线程池相关的内容，将在后续整理。
网友评论