Android源码之AsyncTask

作者: WangGavin | 来源:发表于2019-06-21 23:10 被阅读8次

参考

本文源码版本：Pie 9.0.0_r3
在线源码地址:AsyncTask.java

1 AsyncTask简单用法

// 三个泛型参数分别代表传入的参数类型，任务执行过程需要更新的数据类型，任务执行结束返回的结果类型，如果无类型则可以用Void类
        AsyncTask<Integer,Integer,Void> asyncTask = new AsyncTask<Integer, Integer, Void>() {
            /**
             * 得到结果，在主线程执行
             * @param aVoid
             */
            @Override
            protected void onPostExecute(Void aVoid) {
                Log.d(TAG, "onPostExecute: >>>");
                super.onPostExecute(aVoid);
            }

            /**
             * 任务内容，在工作线程执行
             * @param integers
             * @return
             */
            @Override
            protected Void doInBackground(Integer... integers) {
                Log.d(TAG, "doInBackground: >>>params: "+Arrays.toString(integers));
                return null;
            }

            /**
             * 任务执行前，在主线程执行
             */
            @Override
            protected void onPreExecute() {
                Log.d(TAG, "onPreExecute: >>");
                super.onPreExecute();
            }

            /**
             * 任务已取消（带结果），在主线程执行
             * @param aVoid
             */
            @Override
            protected void onCancelled(Void aVoid) {
                super.onCancelled(aVoid);
                Log.d(TAG, "onCancelled（有参）: >>>");
            }

            /**
             * 任务已取消，在主线程执行
             */
            @Override
            protected void onCancelled() {
                super.onCancelled();
                Log.d(TAG, "onCancelled: >>>");
            }

            /**
             * 指定过程更新的数据，在主线程执行
             * @param values
             */
            @Override
            protected void onProgressUpdate(Integer... values) {
                super.onProgressUpdate(values);
                Log.d(TAG, "onProgressUpdate: "+ Arrays.toString(values));
            }
        };

        asyncTask.execute(2,3,1);

执行结果：

06-20 18:48:44.440 2761-2761/me.newtrekwang.customwidget D/TaskLibActivity: onPreExecute: >>
06-20 18:48:44.441 2761-2808/me.newtrekwang.customwidget D/TaskLibActivity: doInBackground: >>>params: [2, 3, 1]
06-20 18:48:44.454 2761-2761/me.newtrekwang.customwidget D/TaskLibActivity: onPostExecute: >>>

2 源码解读

2.1 线程池大小相关

    /**
     * cpu核心数
     */
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    /**
     * 核心线程：至少两个线程，最多4个线程。 
     */
    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;

怎么分配线程池合理？

一般来说，设N为CPU核数。

如果是CPU密集型应用，则线程池大小设置为N+1.
如果是IO密集型应用，则线程池大小设置为N*2+1

IO密集型

I/O bound 指的是系统的CPU效能相对硬盘/内存的效能要好很多，此时，系统运作，大部分的状况是 CPU 在等 I/O (硬盘/内存) 的读/写，此时 CPU Loading 不高。

CPU密集型

CPU bound 指的是系统的硬盘/内存效能相对 CPU 的效能要好很多，此时，系统运作，大部分的状况是 CPU Loading 100%，CPU 要读/写 I/O (硬盘/内存)，I/O在很短的时间就可以完成，而 CPU 还有许多运算要处理，CPU Loading 很高。

Android 应用的话应该是属于IO密集型应用，所以数量一般设置为 2N+1，AsyncTask里执行任务的线程池也是这样设置的。

2.2 两个线程池

在AsyncTask中有两个线程池，一个线程池（SerialExecutor）用于处理任务列表，一个线程池(ThreadPoolExecutor)用于执行任务。

构建用于执行任务的ThreadPoolExecutor

/**
     * 处理任务的线程池
     */
    public static final Executor THREAD_POOL_EXECUTOR;
    /**
     * 任务队列
     */
    private static final BlockingQueue<Runnable> sPoolWorkQueue = new LinkedBlockingDeque<>(128);
    /**
     * 线程工厂
     */
    private static final ThreadFactory sThreadFactory = new ThreadFactory() {
        /**
         * 线程安全的计数器
         */
        private final AtomicInteger mCount = new AtomicInteger(1);
        
        @Override
        public Thread newThread(@NonNull Runnable runnable) {
            return new Thread(runnable,"AsyncTask #"+mCount.getAndIncrement());
        }
    };
    
    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;
    }

构建用于任务排队的SerialExecutor


    /**
     * 串行任务执行器
     */
    private static final Executor SERIAL_EXECUTOR = new SerialExecutor();
    /**
     * 默认任务执行器
     */
    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
    /**
     * @className WangAsyncTask
     * @createDate 2019/6/20 17:52
     * @author newtrekWang
     * @email 408030208@qq.com
     * @desc 串行任务执行器类
     *
     */
    private static class SerialExecutor implements Executor{
        final ArrayDeque<Runnable> mTasks = new ArrayDeque<>();
        Runnable mActive;
        @Override
        public synchronized void execute(@NonNull final Runnable runnable) {
            // 入队一个runnable,对原始的runnable加了点修饰
            mTasks.offer(new Runnable() {
                @Override
                public void run() {
                    try {
                        runnable.run();
                    } finally {
                        // 任务执行后，要检查是否有下一个runnable需要执行
                        scheduleNext();
                    }
                }
            });
            // 第一次的时候需要触发下才能执行
            if (mActive == null){
                scheduleNext();
            }
        }

        /**
         * 检查是否有下一个runnable需要执行，如果有，则交给另一个线程池执行
         */
        protected synchronized void scheduleNext(){
            if ((mActive = mTasks.poll()) != null){
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }
    }

为什么用两个线程池？

因为AsyncTask是设计为串行执行任务的，所以另外最好需要一个线程池负责任务的排队。

2.3 任务执行流程

2.3.1 构造AsyncTask

    public AsyncTask(@Nullable Looper callbackLooper) {
        mHandler = callbackLooper == null || callbackLooper == Looper.getMainLooper()
            ? getMainHandler()
            : new Handler(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);
                    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);
                }
            }
        };
    }

AsynckTask有关键的三个成员：mHandler,mWorker,mFuture

mHandler

用于线程间通信，将工作线程的消息传递到主线程并做出处理，也就是实现AsyncTask过程数据回调，结果回调在主线程的关键。

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

然后再看InternalHandler类:

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

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

    private static class AsyncTaskResult<Data> {
        final AsyncTask mTask;
        final Data[] mData;

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

AsyncTaskResult里包含的AsynTask对象的引用。

Handler可以从收到的msg里得到统一的AsynctTaskResult数据对象，然后根据消息类型，进行处理。

比如MESSAGE_POST_RESULT是工作线程传的已完成任务标志，然后此时AsyncTask的结束回调方法应该被调用，通过result.mTask.finish(result.mData[0])既可以实现结束回调，使AsyncTask使用者可以从回调种拿到结果。

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

另外MESSAGE_POST_PROGRESS就是过程中工作线程传的更新消息。也一样通过result.mTask.onProgressUpdate(result.mData)实现在Handler所在线程更新过程数据。

mWorker

mWorker就是一个Callable，即有返回结果的Runable,而它的实现类WorkerRunnable还带上了任务参数Params。

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

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

通过上面代码可以知道，用户实现的doInBackground(mParams)方法就是在这里调用的，相当于是把用户的业务代码包装成一个Callable，然后再包装成FutureTask，接着将FutureTask提交给线程池，这样来实现在工作线程执行用户定义的代码块。

然后具体分析call()方法：首先mTaskInvoked是一个AtomicBoolean对象，它能保证线程安全地更新boolean值，mTaskInvoked.set(true)表示该AsyncTask对象已被调用。然后就是设置当前线程的优先级Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND)。然后就是调用用户实现的doInBackground(mParams)方法。然后Binder.flushPendingCommands();是一个native方法，应该是重新调整线程优先级的。然后是捕获异常，最后是通过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;
    }

// 提交更新数据到Handler所在线程
    protected final void publishProgress(Progress... values) {
        if (!isCancelled()) {
            getHandler().obtainMessage(MESSAGE_POST_PROGRESS,
                    new AsyncTaskResult<Progress>(this, values)).sendToTarget();
        }
    }

mFuture

如果单单是提交callable到线程池执行有callable就完事了,但是还需要支持任务的取消操作，那么这个功能就需要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);
                }
            }
        };

mFuture就是mWorker的一个包装类，它也具体实现了Future的一些任务操作接口，比如取消任务，mFuture重写了done()方法，应该是考虑到AsyncTask的get()和cancel()方法内部出异常时对结果的处理。

至于FutureTask是怎样实现取消和get()阻塞得到结果的，我会在另一篇文字做介绍。

2.3.2 execute():执行AsyncTask

execute(Params... params)是AsyncTask为使用者提供的API，它的具体实现如下：

关键代码：

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

        onPreExecute();

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

        return this;
    }

可以看出executor方法实际及时调用executeOnExecutor方法，如果用AsyncTask默认的串行执行任务的线程池就用executor()，默认线程池就是前面提到的sDefaultExecutor，如果是自定义线程池，就用executeOnExecutor()。

然后看executeOnExecutor方法:首先会判断status，如果是RUNNING和FINISHED，会抛异常，也验证了一个AsyncTask只能被执行一次。

然后更新状态为RUNNING,调用onPreExecute(),因为execute（）是在主线程调用的，此时onPreExecute()就是在主线程执行的。然后将用户传的mParams注入mWorker，最后再将future提交到线程池执行。

通过这些代码分析可以得出：一个任务对应一个AsyncTask,一个AsyncTask对应一个Worker,一个Worker对应一个FutureTask,然后多个AsyncTask共用两个默认的线程池和InternalHandler。

大致类关系图

AsyncTask

结语

分析到这里，AsyncTask其实也并不复杂，它始终还是用的线程池+Handler机制来设计的，只要理解了其中的设计步骤，我们自己也可以定义一个AsyncTask。

平时在业务开发中根本就接触不到并发编程知识，只知道使用别人的框架，这对于个人技术提高并没有什么进步。

在AsyncTask中我就了解到了：

线程池的线程数该如何设置？怎样定义线程池？
怎样使用线程池？
Handler机制
FutureTask，RunnableFuture，Callable，Runnable，Future之间的关系
怎样实现任务的取消？

感觉这些知识点都是通用的，理解了这些问题后再去看其它框架源码或设计一个框架，自己心里也有一个方案。

Android源码之AsyncTask

参考

1 AsyncTask简单用法

2 源码解读

2.1 线程池大小相关

怎么分配线程池合理？

IO密集型

CPU密集型

2.2 两个线程池

构建用于执行任务的ThreadPoolExecutor

构建用于任务排队的SerialExecutor

为什么用两个线程池？

2.3 任务执行流程

2.3.1 构造AsyncTask

mHandler

mWorker

mFuture

2.3.2 execute():执行AsyncTask

大致类关系图

结语

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android开发

Android开发

Android开发经验谈

Android技术知识