Android源码系列(15) -- AsyncTask

Posted by phantomVK on October 21, 2018

一、类签名

1.1 作用

AsyncTask 令主线程的正确使用变得简单。无需维护线程或 Handler ,即能让任务在后台线程运算,并把结果提交到主线程。

public abstract class AsyncTask<Params, Progress, Result>

AsyncTask 设计为围绕着 ThreadHandler,且无需构造普通线程框架的帮助类。适合执行(最多运算几秒的)短任务。如果任务导致线程长时间执行,强烈建议用由 java.util.concurrent 包下 ExecutorThreadPoolExecutorFutureTask 提供的APIs。

1.2 组成

工作任务通过后台线程执行,结果最后发布到主线程。

异步任务构成:

  • 3个类型: ParamsProgressResult
  • 4个步骤: onPreExecutedoInBackgroundonProgressUpdateonPostExecute

AsyncTask由子类继承并重写方法 doInBackground(),通常也重写另一个方法 onPostExecute()

用法示例:任务执行参数为URL,进度值类型为Integer,执行结果类型为Long

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]);
            // 把下载进度传递到主线程更新UI
            publishProgress((int) ((i / (float) count) * 100));
            // 通过isCancelled()判断已调用cancel(),尽快跳出本方法
            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);

1.3 3个参数类型

有以下三个被异步任务使用的参数:

  • Params: 任务执行所需参数的类型;
  • Progress: 任务在后台计算时进度单元的类型,如Integer;
  • Result: 后台计算结果返回类型;

三个参数不需全部用上,不需要的用 Void 代替。例如:

private class MyTask extends AsyncTask<Void, Void, Void> { ... }

1.4 4个方法

分别是 onPreExecutedoInBackgroundonProgressUpdateonPostExecute

AsyncTask_Execution

  1. onPreExecute 在任务执行前于主线程调用,起配置任务的作用:如在界面上弹出进度条;
  2. 随后在后台线程调用 doInBackground。本步骤负责执行时间较长的计算任务,参数在此步骤传递到异步任务。计算结果也在这里返给上游。在子线程计算过程中,可通过 publishProgress 传送进度到主线程;
  3. 子线程执行 publishProgress 触发主线程调用 onProgressUpdate,向界面传送进度;
  4. 后台线程执行完毕,计算结果作为参数在主线程传给方法 onPostExecute

1.5 取消任务

任何时候都可通过 cancel(boolean) 取消任务,方法会继续调起 isCancelled() 并返回 true

调用 cancel(boolean) 后,doInBackground(Object[]) 返回后的下一个执行方法是 onCancelled(Object),而不是 onPostExecute(Object) 。(参考小节1.4示意图)

为保证任务能及时取消,需周期性地在 doInBackground(Object[]) 中检查 isCancelled() 方法的返回值。(参考小节1.2示例代码)

1.6 线程规则

为保证类正常运行,有些线程规则需要遵守:

  • AsyncTask 必须在主线程载入。VERSION_CODES.JELLY_BEAN 中此过程自动完成;
  • 任务实例必须在主线程中创建;
  • execute 方法必须在主线程调用;
  • 不得手动调用 onPreExecute()onPostExecute()doInBackground()onProgressUpdate()
  • 每个任务仅能执行一次,任务重复启动会抛出异常;

1.7 内存可观察能力

AsyncTask 保证所有回调通过以下安全、不需显式同步的方式调用:

  • 构造方法onPreExecute 设置成员变量,在 doInBackground 引用;
  • doInBackground 设置成员变量,在 onProgressUpdateonPostExecute 引用;

1.8 执行的顺序

历史实现:

  • 首次发布的 AsyncTasks 类,任务在后台线程中串行执行;

  • VERSION_CODES.DONUT 开始改为线程池,并在多线程并行执行;

  • 为避免并行计算导致错误,从 VERSION_CODES.HONEYCOMB 始任务回到单线程执行;

需要并行执行任务可以通过 executeOnExecutor(java.util.concurrent.Executor, Object[]) 达到使用 THREAD_POOL_EXECUTOR 的目的。并行线程池无法约束任务完成的先后顺序,所以任务之间不能有依赖关系。

1.9 关于系统版本

本文源码来自 Android 28 。但不同历史版本源码实现方法差别非常大,也会出现不同结果。所以在实际运行过程中,务必关注运行时系统版本。

二、常量

2.1 并行线程池

获取设备处理器核心数

private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

核心线程数 最少2个线程、最多4个线程。遵循此前提下,AsyncTask 倾向核心线程数比实际核心数少1个,避免完全占用处理器而影其他任务执行

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); // 原子整形,从1开始递增

    public Thread newThread(Runnable r) {
        return new Thread(r, "AsyncTask #" + mCount.getAndIncrement()); // 设置线程名称
    }
};

缓存任务的阻塞队列,长度128

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

并行任务Executor

public static final Executor THREAD_POOL_EXECUTOR;

初始化并行线程池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;
}

2.2 串行线程池

同一进程共用一个Executor顺序执行任务

public static final Executor SERIAL_EXECUTOR = new SerialExecutor();

2.3 消息类型

消息类型为任务执行结果

private static final int MESSAGE_POST_RESULT = 0x1;

消息类型为主线程进度通知

private static final int MESSAGE_POST_PROGRESS = 0x2;

三、数据成员

// 顺序任务执行的Executor
private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;

// 保存InternalHandler,内部使用主线程Looper或自定义Looper
private static InternalHandler sHandler;

private final WorkerRunnable<Params, Result> mWorker;
private final FutureTask<Result> mFuture;

// 任务状态,任务构建后默认处于Status.PENDING
private volatile Status mStatus = Status.PENDING;

// 任务是否已被取消,注意类型是AtomicBoolean
private final AtomicBoolean mCancelled = new AtomicBoolean();

// 任务是否已被触发,注意类型是AtomicBoolean
private final AtomicBoolean mTaskInvoked = new AtomicBoolean();

private final Handler mHandler;

设置 sHandler

// 获取主线程Handler
private static Handler getMainHandler() {
    // AsyncTask共用同一InternalHandler
    synchronized (AsyncTask.class) {
        // 初始化InternalHandler
        if (sHandler == null) {
            // 向InternalHandler传递主线程的Looper
            sHandler = new InternalHandler(Looper.getMainLooper());
        }
        return sHandler;
    }
}

设置 sDefaultExecutor

// 设置默认Executor
public static void setDefaultExecutor(Executor exec) {
    sDefaultExecutor = exec;
}

四、SerialExecutor

在方法 execute(final Runnable r) 中把新任务r包装到Runnable,达到完成执行任务r后调度下一个任务的目的。

private static class SerialExecutor implements Executor {
    // 存放Runnable的任务队列,ArrayDeque本身非线程安全
    final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();

    // 正在执行的Runnable
    Runnable mActive;

    // 进程内只有一个SerialExecutor实例
    // 方法使用synchronized修饰,保证mTasks操作线程安全,用于放新任务到队列
    public synchronized void execute(final Runnable r) {
        mTasks.offer(new Runnable() {
            public void run() {
                try {
                    r.run();
                } finally {
                    // 此任务运行完成后触发下一个任务执行
                    scheduleNext();
                }
            }
        });
        
        // 线程池首次执行时mActive为空,在此开始调度第一个任务
        if (mActive == null) {
            scheduleNext();
        }
    }

    // SerialExecutor进程内只有一个实例,方法使用synchronized修饰,保证mTasks操作线程安全
    protected synchronized void scheduleNext() {
        // 从任务队列获取下一任务
        if ((mActive = mTasks.poll()) != null) {
            // 向并行执行线程池添加任务
            THREAD_POOL_EXECUTOR.execute(mActive);
        }
    }
}

虽然任务在 THREAD_POOL_EXECUTOR 执行,但是都由 SERIAL_EXECUTOR 调度。从上一个完成任务调用 scheduleNext() 唤醒下一个任务。除非主动把任务添加到并行线程池,否则每次只有一个任务在并行执行线程池内执行。

五、状态枚举

任务状态枚举,表示任务当前运行时状态

public enum Status {
    PENDING, // 任务尚未执行,正在排队等待

    RUNNING, // 任务正在执行(运算)标志

    FINISHED, // 任务执行完毕:先调用onPostExecute(),再把状态置为此值
}

所有任务按照此生命周期单向前进,每个状态只允许设置一次。

AsyncTask_Status

六、构造方法

构建新异步任务,所有构造方法必须在主线程调用

public AsyncTask() {
    this((Looper) null);
}

通过指定Handler构建实例

public AsyncTask(@Nullable Handler handler) {
    this(handler != null ? handler.getLooper() : null);
}

若没有传入其他Looper,构造方法会主动获取主线程Looper并创建Handler

public AsyncTask(@Nullable Looper callbackLooper) {
    mHandler = callbackLooper == null || callbackLooper == Looper.getMainLooper()
        ? getMainHandler()
        : new Handler(callbackLooper);

    // 构建WorkerRunnable
    mWorker = new WorkerRunnable<Params, Result>() {
        public Result call() throws Exception {
            // 设置任务已被触发
            mTaskInvoked.set(true);
            // 任务执行结果
            Result result = null;
            try {
                // 设置线程优先级
                Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
                // 把任务所需参数传入到后台线程执行并获取结果
                result = doInBackground(mParams);
                Binder.flushPendingCommands();
            } catch (Throwable tr) {
                // 任务执行出现异常,则取消任务
                mCancelled.set(true);
                throw tr;
            } finally {
                // 任务执行完成把结果传递给postResult
                postResult(result);
            }
            // 返回结果
            return result;
        }
    };

    // 把WorkerRunnable封装到FutureTask
    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);
            }
        }
    };
}

七、成员方法

如果任务没有被调用过,通过此方法返回结果

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

首先,把结果封装到 AsyncTaskResult 中,结果类型为 Result,然后放到 Message.obj 中发送到目标 Handler

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

获取构造方法传入的自定义 Handler 或主线程 LooperHandler,详情见小节六、构造方法

private Handler getHandler() {
    return mHandler;
}

获取当前任务的执行状态

public final Status getStatus() {
    return mStatus;
}

重写方法实现后台线程的计算逻辑。任务调用者提供参数 paramsexecute()execute() 传递给本方法。在方法内可调用 publishProgress() 向主线程发布实时更新值

@WorkerThread
protected abstract Result doInBackground(Params... params);

doInBackground() 调用前,先在主线程执行此方法

@MainThread
protected void onPreExecute() {
}

doInBackground() 完成后在主线程调用此方法,resultdoInBackground() 返回的结果。如果任务被取消,此方法不会触发

@SuppressWarnings({"UnusedDeclaration"})
@MainThread
protected void onPostExecute(Result result) {
}

publishProgress() 运行后切换到主线程调用此方法,values 表示任务处理的进度

@SuppressWarnings({"UnusedDeclaration"})
@MainThread
protected void onProgressUpdate(Progress... values) {
}

cancel(boolean) 被调用,且 doInBackground(Object[]) 结束后在主线程调用此方法。

方法默认简单回调 onCancelled() 并忽略结果。如果要在子类重写其他实现,不要在重写方法内调用 super.onCancelled(result)。注意,result 可为 null

@SuppressWarnings({"UnusedParameters"})
@MainThread
protected void onCancelled(Result result) {
    onCancelled();
}    

应用最好能重写本方法,以便在任务被取消后做出反应。此方法由 onCancelled(Object) 的默认实现调用。 cancel(boolean) 被调用且 doInBackground(Object[]) 已结束后,方法在主线程上调用。

@MainThread
protected void onCancelled() {
}

若任务在正常完成前被取消,此方法返回true。

public final boolean isCancelled() {
    return mCancelled.get();
}

尝试取消任务,如果任务已执行完毕、已经取消、由于其他原因不能取消的,则取消失败。任务取消成功,且在 cancel() 调用时尚未开始,任务不会执行。

如果任务已经开始,参数 mayInterruptIfRunning 决定任务执行线程是否该被中断。调用此方法,且 doInBackground(Object[]) 结束后,会在主线程调用 onCancelled(Object)。调用此方法能保证 onPostExecute(Object) 不会执行。

此方法调用后,需从 doInBackground(Object[]) 周期性检查由 isCancelled() 返回的值并尽快结束任务。参数 mayInterruptIfRunningtrue,执行任务线程需被中断,否则等任务执行直至完成。

public final boolean cancel(boolean mayInterruptIfRunning) {
    mCancelled.set(true);
    return mFuture.cancel(mayInterruptIfRunning);
}

等待计算完毕后获取执行结果:

  • 任务被取消后调用此方法,抛出CancellationException;

  • 当前线程在等待结果过程被中断,抛出InterruptedException;

public final Result get() throws InterruptedException, ExecutionException {
    return mFuture.get();
}

等待计算完毕后获取执行结果,设置timeout作为等待超时时间:

  • 任务被取消后调用此方法,抛出CancellationException;

  • 任务执行过程中出现异常,抛出ExecutionException;

  • 当前线程等待结果过程中被中断,抛出InterruptedException;

  • 等待结果超时,抛出TimeoutException;

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

方法用 THREAD_POOL_EXECUTOR 实现多任务并行处理,也可以用自定义Executor实现定制。并行执行不能保证任务运行顺序的先后,如果多个任务需有序执行,请使用顺序任务。

@MainThread
public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
        Params... params) {
    // 若任务默认状态不是PENDING状态,直接抛出异常
    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;
}

通过默认Executor执行单个Runnable

@MainThread
public static void execute(Runnable runnable) {
    sDefaultExecutor.execute(runnable);
}

doInBackground() 在后台线程运行过程可(多次)调用此方法。每次调用方法都会在子线程向主线程发布更新进度的 Message,并在主线程触发 onProgressUpdate()。如果任务被取消,onProgressUpdate 不会调用。

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

八、InternalHandler

构造AsyncTask实例时默认主线程Looper

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

    // 消息通过handleMessage在主线程执行分发运行逻辑
    @SuppressWarnings({"unchecked", "RawUseOfParameterizedType"})
    @Override
    public void handleMessage(Message msg) {
        // 从Message中获取异步任务执行结果
        AsyncTaskResult<?> result = (AsyncTaskResult<?>) msg.obj;
        // 按照消息不同类型执行逻辑
        switch (msg.what) {
            case MESSAGE_POST_RESULT:
                // 唯一结果传入finish(),内部再调用onPostExecute()
                result.mTask.finish(result.mData[0]);
                break;
            case MESSAGE_POST_PROGRESS:
                // 通知主线程更新进度
                result.mTask.onProgressUpdate(result.mData);
                break;
        }
    }
}

九、WorkerRunnable

WorkerRunnable实现Callable接口。相比Runnable接口,Callable会在完成后返回结果,子类需实现call()抽象方法

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

十、AsyncTaskResult

异步任务执行后结果

@SuppressWarnings({"RawUseOfParameterizedType"})
private static class AsyncTaskResult<Data> {
    final AsyncTask mTask;
    final Data[] mData;

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