在Android中，异步执行是很重要的一块内容，诸如网络请求，大图片的加载，等待等耗时操作都要在后台线程执行，而这些操作又要通过UI线程来调用，这样我们不得不需要通过异步执行来对这些耗时的操作进行处理。安卓与异步执行有关的类包括了Handler和AsyncTask，Handler主要负责各线程间传递消息（大多用于其它线程与UI线程进行信息交互），而AsyncTask则主要负责异步操作的处理，当我们需要进行一些耗时的操作，耗时操作完成后更新主线程，或者在操作过程中对主线程的UI进行更新时，就需要考虑应用AsyncTask了。在AsyncTask中，doInBackGround()方法运行在后台线程中，负责异步任务的处理，onProgressUpdate()方法则在doInBackGround()执行过程中调用publishProgress()时回调，以随时更新UI。onPostExecute()方法则是当异步执行完毕后对UI的更新操作，onProgressUpdate()和onPostExecute()方法都运行在UI线程中。这些功能是如何实现的？AsyncTask用到了哪些线程池技术？现从源码的角度来简单分析一下AsyncTask的原理实现。
在看AsyncTask之前，首先我们列举下AsyncTask用到的java提供的线程相关类和接口，这对于我们理解AsyncTask十分有帮助：
首先是Future接口。Future多用于耗时线程的计算，主线程可以在完成自己的任务后，再去查询该Future是否执行完毕并获取结果。Future接口的定义如下：
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
Future接口的主要方法：
cancel()：取消一个任务，不过当一个任务已经完成时，取消就会失败，这个方法有一个参数mayInterruptIfRunning，如果传入的是true，那么即使你的任务已经开始，那么也会试图中断任务所在线程。
isDone()：如果一个任务执行完毕，那么这里返回true。
get()：get()是一个同步方法，只到任务成功执行完毕才返回，如果任务取消或者中断，都会抛出异常，有参数的get方法是限定了等待时间，如果在指定时间内没有返回，那么就不再等待。
FutureTask类: 首先，FutureTask继承了RunnableFuture接口，RunnableFuture接口又继承了Runnable、和Future接口，所以它可以当做一个Runnable直接调用，也可以在像Future一样在线程执行过程中和完毕后做一些操作（isDone done,cancel）。FutureTask的构造函数既可以包装一个Runnable也可包装一个Callable。他的内部定义了一个很重要的回调函数protected void done()，当任务结束时，done()方法会被触发。因此，只需重载该函数，即可在线程刚结束时做一些事情。（AsyncTask利用该接口调用onPostExecute()方法）
public class FutureTask<V> implements RunnableFuture<V> {
/**
* Protected method invoked when this task transitions to state
* <tt>isDone</tt> (whether normally or via cancellation). The
* default implementation does nothing. Subclasses may override
* this method to invoke completion callbacks or perform
* bookkeeping. Note that you can query status inside the
* implementation of this method to determine whether this task
* has been cancelled.
*/
protected void done() { }
通过方法说明可以看到，FutureTask的done()方法是任务状态变为isdone（无论是手动取消还是任务完成）时调用，有兴趣的朋友可以看FutureTask的源码，看看它是怎么调用的。
Executor接口：提供了一个execute方法，该方法负责将任务保存起来，可能在未来某一个时刻执行。至于具体是在新的线程中执行还是在线程池中执行由实现该接口的类决定的。
public interface Executor {
/**
* Executes the given command at some time in the future. The command
* may execute in a new thread, in a pooled thread, or in the calling
* thread, at the discretion of the <tt>Executor</tt> implementation.
*
* @param command the runnable task
* @throws RejectedExecutionException if this task cannot be
* accepted for execution.
* @throws NullPointerException if command is null
*/
void execute(Runnable command);
}
ExecutorService 接口：继承自Executor，线程池的继承的主要接口，接口定义如下：
public interface ExecutorService extends Executor {
/**
* Initiates an orderly shutdown in which previously submitted
* tasks are executed, but no new tasks will be accepted.
* Invocation has no additional effect if already shut down.
*
* <p>This method does not wait for previously submitted tasks to
* complete execution. Use {@link #awaitTermination awaitTermination}
* to do that.
*/
void shutdown();
/**
* Attempts to stop all actively executing tasks, halts the
* processing of waiting tasks, and returns a list of the tasks
* that were awaiting execution.
*
* <p>This method does not wait for actively executing tasks to
* terminate. Use {@link #awaitTermination awaitTermination} to
* do that.
*
* <p>There are no guarantees beyond best-effort attempts to stop
* processing actively executing tasks. For example, typical
* implementations will cancel via {@link Thread#interrupt}, so any
* task that fails to respond to interrupts may never terminate.
*
* @return list of tasks that never commenced execution
*/
List<Runnable> shutdownNow();
/**
* Returns {@code true} if this executor has been shut down.
*
* @return {@code true} if this executor has been shut down
*/
boolean isShutdown();
/**
* Returns {@code true} if all tasks have completed following shut down.
* Note that {@code isTerminated} is never {@code true} unless
* either {@code shutdown} or {@code shutdownNow} was called first.
*
* @return {@code true} if all tasks have completed following shut down
*/
boolean isTerminated();
/**
* Blocks until all tasks have completed execution after a shutdown
* request, or the timeout occurs, or the current thread is
* interrupted, whichever happens first.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return {@code true} if this executor terminated and
* {@code false} if the timeout elapsed before termination
* @throws InterruptedException if interrupted while waiting
*/
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
/**
* Submits a value-returning task for execution and returns a
* Future representing the pending results of the task. The
* Future's {@code get} method will return the task's result upon
* successful completion.
*
* <p>
* If you would like to immediately block waiting
* for a task, you can use constructions of the form
* {@code result = exec.submit(aCallable).get();}
*
* <p>Note: The {@link Executors} class includes a set of methods
* that can convert some other common closure-like objects,
* for example, {@link java.security.PrivilegedAction} to
* {@link Callable} form so they can be submitted.
*
* @param task the task to submit
* @return a Future representing pending completion of the task
* @throws RejectedExecutionException if the task cannot be
* scheduled for execution
* @throws NullPointerException if the task is null
*/
<T> Future<T> submit(Callable<T> task);
/**
* Submits a Runnable task for execution and returns a Future
* representing that task. The Future's {@code get} method will
* return the given result upon successful completion.
*
* @param task the task to submit
* @param result the result to return
* @return a Future representing pending completion of the task
* @throws RejectedExecutionException if the task cannot be
* scheduled for execution
* @throws NullPointerException if the task is null
*/
<T> Future<T> submit(Runnable task, T result);
/**
* Submits a Runnable task for execution and returns a Future
* representing that task. The Future's {@code get} method will
* return {@code null} upon <em>successful</em> completion.
*
* @param task the task to submit
* @return a Future representing pending completion of the task
* @throws RejectedExecutionException if the task cannot be
* scheduled for execution
* @throws NullPointerException if the task is null
*/
Future<?> submit(Runnable task);
...
}
ExecutorService的主要方法：
shutDown()：关闭线程池，顺序地执行关闭操作。
shutDownNow()：关闭所有已执行的任务，并返回已挂起的线程队列；
submit():执行一个callback或runnable，返回值Furture为这个线程的执行状态。
重点来了：ThreadPoolExecuter 线程池类
ThreadPoolExecuter实现了ExecutorService，对线程池进行了进一步封装，它的构造方法如下：
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory and rejected execution handler.
* It may be more convenient to use one of the {@link Executors} factory
* methods instead of this general purpose constructor.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
ThreadPoolExecutor类为jdk1.5添加的线程池类，功能十分强大，网上有很多的文章对它进行了详细讲解，大家可以参考某某博客，这里我只是对它进行一些简单介绍。首先来看下ThreadPoolExecutor构造函数的参数说明（英文好的朋友可以无视我的翻译，直接看上面的文档说明）：
corePoolSize：池中所保存的线程数，即使是空闲线程也会线程池也会创建它。
maximumPoolSize：池中允许的最大线程数。
keepAliveTime：当线程数大于corePoolSize时，多余的空闲线程等待新任务的最长时间。
unit：keepAliveTime参数的时间单位。
workQueue：执行前用于保持任务的队列，可定义成无界队列和有界队列。
这里我们要重点理解ThreadPoolExecutor中corePoolSize，maximumPoolSize，workQueue这三个参数的意义。
当我们向线程池中添加一个任务时
1、如果线程数<corePoolSize，那么创建一个线程，不管此时是否有线程空闲；
2、如果线程数=corePoolSize，如果有空闲线程，那么空闲线程处理加入的任务，如果没有空闲线程，那么加入workQueue中。只到workQueue线程已经满了，才会创建新的线程来处理新加入的任务，如果此时创建的线程数超过了maximumPoolSize，那么就会抛RejectedExecutionException异常。
3、如果线程数>corePoolSize时，那么说明workQueue已经满了。
通过以上的描述，说明如果workQueue是一个无界队列，那么maximumPoolSize就没有意义了。
上面我们介绍了与线程有关的类和接口，真正的重点来了！现在我们来看来AsyncTask的具体实现：
首先我们来看下AsyncTask的定义，它包含了三个泛型参数：Params、Progress、Result。
public abstract class AsyncTask<Params, Progress, Result> {
}
在讲这三个泛型参数之前，我们来看下AsyncTask几个核心方法：
protected abstract Result doInBackground(Params... params);
protected void onPostExecute(Result result) {}
protected void onPreExecute() {}
protected void onProgressUpdate(Progress... values) {}
public final AsyncTask<Params, Progress, Result> execute(Params... params);
可以看到，doInBackground(Params... params)方法是抽象的，它是AsyncTask的核心方法，指定了后台线程的所执行的任务，它的参数类型为Params，返回类型为Result，这些我们都可以自己定义。onPreExecute()和onPostExecute(Result result)方法则很容易理解，通过重写它们我们可以在后台线程执行前后做一些事情。注意，onPostExecute()方法有一个Result参数，这个参数和doInBackground()返回的结果相同。onProgressUpdate (Progress... values)方法则为进度更新的回调，Progress泛型代表执行过程中所传递的参数类型，通过在doInBackGround方法里调用publishProgress (Progress... value)方法，onProgressUpdate就会被回调。而execute()方法则是AsyncTask的执行方法，调用它AsyncTask就会执行。
结合上面的分析，大家可能会猜出来AsyncTask是如何实现的了，它应该会定义一个ThreadPoolExecutor线程池，这个doInBackground()方法应该会放入一个FutureTask中，当调用AsyncTask.execute()时传入Params参数并放入该线程池里执行，返回值Result通过执行的FutureTask返回，在执行之前调用onPreExecute，之后调用PostExecute，doInBackground执行过程中调用publishProgress时可能是通过发送一个Handler来调用onProgressUpdate？哈哈，没错，这就是AsynckTask的执行过程，让我们来具体看一下AsyncTask源码是如何实现的：
首先我们看下AsyncTask内部线程池THREAD_POOL_EXECUTOR的定义：
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;
private static final int KEEP_ALIVE = 1;
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
= new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE,
TimeUnit.SECONDS, sPoolWorkQueue, sThreadFactory);
结合上面介绍的线程池相关知识，可得出如下几点结论：
1、AsyncTask中的线程池THREAD_POOL_EXECUTOR保存的线程数量是CPU核数+1个，也就是说同时它可以执行CPU核数+1个工作线程，大于CPU_COUNT+1的线程如果空闲1s就会被干掉。
AsyncTask中维护着一个长度为128的线程池，，还有一个缓冲队列，当线程池中已有128个线程，缓冲队列已满时，如果
此时向线程提交任务，将会抛出RejectedExecutionException。
解决：由一个控制线程来处理AsyncTask的调用判断线程池是否满了，如果满了则线程睡眠否则请求AsyncTask继续处理。
2、最大的线程数为CPU_COUNT*2 +1 个，LinkedBlockingQueue的容量是128，按照前面的分析，AsyncTask最多可以添加CPU_COUNT*2+129个任务，再添加任务时，程序就会抛出RejectedExecutionException异常。
3、线程池是static的，也就是说在一个进程中，所有的AsyncTask共用一个线程池。
事实上，AsyncTask内部对THREAD_POOL_EXECUTOR的进行了进一步的封装：SerialExecutor。SerialExecutor实现了Executor接口，内部封装了一个线程队列，实现了按顺序向THREAD_POOL_EXECUTOR execute的功能。SerialExecutor在AsyncTask类中被实例成static final的（整个类只有一份静态化实例）。即AsyncTask的所有任务都会通过该类，向THREAD_POOL_EXECUTOR顺序输送消息并执行。
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);
}
}
}
接下来看下AsyncTask的构造方法：
public AsyncTask() {
mWorker = new WorkerRunnable<Params, Result>() {
public Result call() throws Exception {
mTaskInvoked.set(true);
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
//noinspection unchecked
return postResult(doInBackground(mParams));
}
};
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 occured while executing doInBackground()",
e.getCause());
} catch (CancellationException e) {
postResultIfNotInvoked(null);
}
}
};
}
private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}
这里定义了一个WorkerRunnable类继承了Callable，要说明的是，Callable与Runnable功能相同，也是执行线程的载体，只不过带有自己的返回值罢了。可以看到在AsyncTask的构造函数中，定义了一个WorkerRunnable类型的mWorker，并在call方法里调用了doInBackground方法，并将mWorker包装在mFuture里，而mFuture是一个futureTask，之后会放入线程池中执行。因为doInBackgound是放在Callable中，所以可以得出结论，doInBackgound方法是在后台线程中执行。
大家都知道我们想让AsyncTask执行任务，除了新建一个AsyncTask实例外，还必须调用它的execute()方法。execute()方法是怎么实现的？来接着看源码：
public final AsyncTask<Params, Progress, Result> execute(Params... params) {
return executeOnExecutor(sDefaultExecutor, params);
}
execute(Params... params)方法接着调用了executeOnExecutor()方法：
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;
}
executeOnExecutor方法有两个参数，一个是线程池exec，一个是Params参数，在这里，exec就是我们之前看到的默认的线程池。可见，在mWorker添加了参数之后，mFuture就会加入线程池中并执行。执行之前会调用onPreExecute()方法，可见，onPreExecute()是在主线程中执行的。当mFuture加入线程池后，它封装的call方法即会被适时调用，即doInBackground方法会被适时执行。
我们再来看下这个mFuture的在构造函数中定义，进而分析onPostExecute(Result result)和onProgressUpdate(Progress progress)方法是怎么执行的：
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 occured while executing doInBackground()",
e.getCause());
} catch (CancellationException e) {
postResultIfNotInvoked(null);
}
}
};
通过上面的分析我们知道，当mFuture执行时，如果任务状态变为isDone()（执行完毕或中途cancel），会执行done方法，在这里done()方法调用了postResultIfNotInvoked()方法，postResultIfNotInvoked方法又会调用postResult()方法发送一个handler消息。来看源码：
private void postResultIfNotInvoked(Result result) {
final boolean wasTaskInvoked = mTaskInvoked.get();
if (!wasTaskInvoked) {
postResult(result);
}
}
private Result postResult(Result result) {
@SuppressWarnings("unchecked")
Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
new AsyncTaskResult<Result>(this, result));
message.sendToTarget();
return result;
}
原来如此，当任务结束后，mFuture会调用一个handler来发送消息。这个handler为AsyncTask内部定义的InternalHander，来看下它的源码：
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;
}
}
}
看到这个handler，是不是都明白了？原来当任务结束时，mFuture会发送给InternalHandler一个message，消息为MESSAGE_POST_RESULT，这在InternalHandler里被解析为finish，finish方法会调用onPostExecute(Result result)：
private void finish(Result result) {
if (isCancelled()) {
onCancelled(result);
} else {
onPostExecute(result);
}
mStatus = Status.FINISHED;
}
除此之外，我们也会看到，InternalHandler还会解析MESSAGE_POST_RESULT消息，这个消息在publishProgress(Progress... values)方法里发送：
protected final void publishProgress(Progress... values) {
if (!isCancelled()) {
getHandler().obtainMessage(MESSAGE_POST_PROGRESS,
new AsyncTaskResult<Progress>(this, values)).sendToTarget();
}
}
而 publishProgress在哪里调用呢？当然是我们自己在执行doInBackGround()方法里调用啦。另外，这个InternalHandler首先会调用主线程的Looper，即它操作都是在UI线程中完成，所以，onProgressUpdate()和onPostExecute()方法都是在UI线程中执行。
总结：
AsyncTask主要负责android中异步操作的处理，它同时可以执行CPU_COUNT+1个工作线程，最大容量为CPU_COUNT*2+129，再添加任务时，程序就会抛出异常。我们可以通过一个控制线程来处理AsyncTask的调用判断线程池是否满了，如果满了则线程睡眠否则请求AsyncTask继续处理。AsyncTask内部的线程池为static的，这意味着在一个安卓进程中，所有的线程共用这一个线程池。异步操作由doInBackground(Param... params)方法承担，doInBackground(Param... params)方法运行在非UI线程，可以处理一些耗时的操作。doInBackground(Param... params)执行前后会调用onPreExecute()、onPostExecute (Result result)方法，在doInBackground(Param... params)执行过程中，可以通过publishProgress(Progress... progress)方法，它会回调onProgress (Progress... progress)方法处理线程执行过程中的一些操作。除了doInBackground()，其它方法都是在UI线程中执行。我们在重写AsyncTask的时候，首先要定义它的doInBackground、onPostExecute等方法，然后对其初始化，再执行它的execute即可。
public class MyAsyncTask extends AsyncTask<String, Integer, String> {
@Override
protected String doInBackground(String... params) {
Log.v("lzq", "doInBackground");
for (int i = 0; i < 5; i++)
publishProgress(i);
return "doInBackground Result";
}
@Override
protected void onProgressUpdate(Integer... values) {
Log.v("lzq", "onProgressUpdate " + values[0]);
super.onProgressUpdate(values);
}
@Override
protected void onPostExecute(String s) {
Log.v("lzq", "onPostExecute " + s);
}
}
MyAsyncTask task = new MyAsyncTask();
task.execute();
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
