AsyncTask在Android中是很常用的异步线程，那么AsyncTask和Thread有什么区别呢？这里将从源码角度深入理解AsyncTask的设计和工作原理
这里的AsyncTask基于SDK-25
分析知识准备
首先我们来看一个生产者与消费者模型的例子1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64public class ThreadTest {
//产品
static class ProductObject{
public volatile static String value; //volatile线程操作变量可见
}
//生产者线程
static class Producer extends Thread{
Object lock;
public Producer(Object lock) {
this.lock = lock;
}
public void run() {
while(true){
synchronized (lock) {
if(ProductObject.value != null){
try {
lock.wait(); //产品还没有被消费，等待
} catch (InterruptedException e) {
e.printStackTrace();
}
}
ProductObject.value = "NO:"+System.currentTimeMillis();
System.out.println("生产产品："+ProductObject.value);
lock.notify(); //生产完成，通知消费者消费
}
}
}
}
//消费者线程
static class Consumer extends Thread{
Object lock;
public Consumer(Object lock) {
this.lock = lock;
}
public void run() {
while(true){
synchronized (lock) {
if(ProductObject.value == null){
try {
lock.wait(); //等待，阻塞
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("消费产品："+ProductObject.value);
ProductObject.value = null;
lock.notify(); //消费完成，通知生产者，继续生产
}
}
}
}
public static void main(String[] args) {
Object lock = new Object();
new Producer(lock).start();
new Consumer(lock).start();
}
}
上面的例子关键点在于两个，其一是volatile，使得线程间可见，第二个点在于互斥锁，这样就可以使得有商品的时候就要通知消费者消费，同时wait，那么消费者收到消息开始消费，消费完毕通知生产者继续生产，从而不断生产，这样比轮询方式更加节省资源
在了解完上面的例子以后，我们就可以着手分析AsyncTask的源代码了
首先，我们在AsyncTask首先看其构造方法1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39private final WorkerRunnable<Params, Result> mWorker;
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) {
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);
}
}
};
}
这里首先给WorkerRunnable和Future进行了初始化，那么为何要初始化这两个变量呢？
这里就要说到常用的两个方法了，doInBackground()，这个方法是在子线程里面完成的，另一个方法就是onPostExecute()，而这个方法是存在于主线程的，那么也就是说子线程执行完将执行的结果传递到了主线程中，实现了线程间的通信，那么最关键的问题来了，这个通信是怎么实现的呢？
通常在子线程中执行的任务，是没有返回结果的，例如Runnable的源代码如下，就没有返回结果1
2
3
4
5
6
7
8
9
10
11
12
13
14public interface Runnable {
/**
* When an object implementing interface <code>Runnable</code> is used
* to create a thread, starting the thread causes the object's
* <code>run</code> method to be called in that separately executing
* thread.
* <p>
* The general contract of the method <code>run</code> is that it may
* take any action whatsoever.
*
* @see java.lang.Thread#run()
*/
public abstract void run();
}
那么，要怎么才能得到返回值呢，这里首先想到的就是Callable接口，那么再看看Callable的源代码1
2
3
4
5
6
7
8
9
10
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
可以看到，这是一个泛型方法，是有返回值的，但是其本身确是不能直接执行的，需要借助其他类，接下来再看一看源代码中涉及到的Future接口1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74public interface Future<V> {
/**
* Attempts to cancel execution of this task. This attempt will
* fail if the task has already completed, has already been cancelled,
* or could not be cancelled for some other reason. If successful,
* and this task has not started when {@code cancel} is called,
* this task should never run. If the task has already started,
* then the {@code mayInterruptIfRunning} parameter determines
* whether the thread executing this task should be interrupted in
* an attempt to stop the task.
*
* <p>After this method returns, subsequent calls to {@link #isDone} will
* always return {@code true}. Subsequent calls to {@link #isCancelled}
* will always return {@code true} if this method returned {@code true}.
*
* @param mayInterruptIfRunning {@code true} if the thread executing this
* task should be interrupted; otherwise, in-progress tasks are allowed
* to complete
* @return {@code false} if the task could not be cancelled,
* typically because it has already completed normally;
* {@code true} otherwise
*/
boolean cancel(boolean mayInterruptIfRunning);
/**
* Returns {@code true} if this task was cancelled before it completed
* normally.
*
* @return {@code true} if this task was cancelled before it completed
*/
boolean isCancelled();
/**
* Returns {@code true} if this task completed.
*
* Completion may be due to normal termination, an exception, or
* cancellation -- in all of these cases, this method will return
* {@code true}.
*
* @return {@code true} if this task completed
*/
boolean isDone();
/**
* Waits if necessary for the computation to complete, and then
* retrieves its result.
*
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
*/
V get() throws InterruptedException, ExecutionException;
/**
* Waits if necessary for at most the given time for the computation
* to complete, and then retrieves its result, if available.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
* @throws TimeoutException if the wait timed out
*/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
在Future类中，有好几个方法，而这些方法都是有返回值的，那么Runnable与Future和FutureTask有什么关系呢，产看源码便可得知，FutureTask实际上是实现了RunnableFuture接口1
2
3public class FutureTask<V> implements RunnableFuture<V>{
···
}
而RunnableFuture又继承了Runnable和Future1
2
3public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}
那也就是说，FutureTask既可以在子线程中执行，也可以获得执行结果，下面使用一个例子来说明FutureTask1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39public class FutureTest {
public static void main(String[] args) {
Task work = new Task();
FutureTask<Integer> future = new FutureTask<Integer>(work){
protected void done() { //异步任务执行完成，回调
try {
System.out.println("done:" + get()); //get()获取异步任务的返回值，这是个阻塞方法
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
};
//线程池（使用了预定义的配置）
ExecutorService executor = Executors.newCachedThreadPool();
executor.execute(future);
}
//异步任务
static class Task implements Callable<Integer>{
public Integer call() throws Exception {//返回异步任务的执行结果
int i = 0;
for (; i < 10; i++) {
try {
System.out.println(Thread.currentThread().getName() + "_" + i);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
return i;
}
}
}
上面的例子可以看出，在使用了Callable的时候，需要借助FutureTask来包装，然后使用Executor的execute()方法来执行，那么是怎么得到异步任务的返回值呢，在上面的例子中，我们可以看到，其返回值的获取是通过future.get()得到的，然而这个get()方法确是被阻塞的，只有在异步任务完成的时候才能获取到其结果，那我们怎么才能知道异步任务时候执行完毕呢，这里就可以实现FutureTask的done()方法，当异步任务执行完毕以后会回调这个方法，上述例子其实解释了AsyncTask的实现逻辑，call()方法是在子线程中完成，这也就是doInBackground()的实现，在主线程中获得结果，这是在onPostExecute()使用了get()方法，那也就是说AsyncTask就是通过这一套方法去实现的
从这里我们可以总结出FutureTask为异步任务提供了诸多便利性，包括
- 获取异步任务的返回值
- 监听异步任务的执行情况
- 取消异步任务
那么在AsyncTask中，WorkerRunnable又是啥呢，其实就是一个内部类，对Callable进行了封装1
2
3private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}
源代码分析
有了以上知识储备，我们就可以动手分析AsyncTask源代码了
拿到源代码，不同的人有不同的分析习惯，这里我按照我的习惯对源代码进行一次分析
构造方法分析
首先，因为我们分析源代码是为了更好的去使用，而使用的话，第一个关注的就应该是构造方法，回到之前的的构造方法，这里要开始对构造方法开始入手分析了1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47private final WorkerRunnable<Params, Result> mWorker;
private final FutureTask<Result> mFuture;
private final AtomicBoolean mCancelled = new AtomicBoolean();
private final AtomicBoolean mTaskInvoked = new AtomicBoolean();
···
public AsyncTask() {
mWorker = new WorkerRunnable<Params, Result>() {
public Result call() throws Exception {
//设置线程调用
mTaskInvoked.set(true);
Result result = null;
try {
//设置线程优先级，其给定值为10
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
//调用doInBackground()方法得到返回值
result = doInBackground(mParams);
//将当前线程中的Binder命令发送至kernel
Binder.flushPendingCommands();
} catch (Throwable tr) {
//发生异常则取消线程调用设置
mCancelled.set(true);
throw tr;
} finally {
//执行postResult()方法
postResult(result);
}
return result;
}
};
mFuture = new FutureTask<Result>(mWorker) {
protected void done() {
try {
//执行postResultIfNotInvoked()方法
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);
}
}
};
}
以上就是AsyncTask的构造方法了，在构造方法上有一句说明，这个构造方法必须在UI线程中创建，这一点很好理解，因为其有需要再主线程中执行的地方，后面会说到，那么这个构造方法干了什么事情呢，很简单，这里新建了两个对象，首先是WorkerRunnable，而这个WorkerRunnable则是实现自Callable接口，主要是要使用其call()方法，为了返回参数，并没有什么特别之处，再其内部则实现了call()方法，而其自生是无法执行的，需要找一个包装类，而这个包装类就是FutureTask，通过之前的分析，这里就不再多赘述关于FutureTask的东西了，这里实现了done()方法，也就是线程执行完毕调用的方法，简单点来说就是在call()方法中执行，在done()中获得执行的返回结果，上述涉及到一个内部类和三个自定义的方法，那么接下来我们看一看这个内部类和三个方法都干了啥
构造方法中出现的内部类
这里的WorkerRunnable，正如前面所说，这里除了实现Callable就啥也没干，还是个抽象方法，这里将实现放在了构造方法中1
2
3private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}
构造方法中调用的方法
首先是doInBackground()方法，前面讲到，这个方法在子线程中完成，那么这里的子线程是哪个呢，其实就是WorkerThread，这个方法是一个抽象方法，放在子线程中执行，其具体实现由调用者完成1
2
protected abstract Result doInBackground(Params... params);
再看postResult()方法，这里获取了一个Handler，然后发送了一个消息，这里就是子线程能够通信主线程的地方了1
2
3
4
5
6
7private Result postResult(Result result) {
("unchecked")
Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
new AsyncTaskResult<Result>(this, result));
message.sendToTarget();
return result;
}
那么到这里，我们关注的重点就来了，子线程是怎么告诉主线程的呢，要知道其中的原因，我们就需要去看看代码里面是怎么实现的
我们先看发送了什么消息，也就是AsyncTaskResult里面干了啥，查看代码发现，其就是做了参数传递的任务1
2
3
4
5
6
7
8
9
10({"RawUseOfParameterizedType"})
private static class AsyncTaskResult<Data> {
final AsyncTask mTask;
final Data[] mData;
AsyncTaskResult(AsyncTask task, Data... data) {
mTask = task;
mData = data;
}
}
那么接下来的重点就是getHandler()方法了，这里拿到AsyncTask.class就上了锁了，这也很好理解，不上锁其他线程走到这里会产生安全隐患，然后返回sHandler，那再继续看看InternalHandler又是个什么吧1
2
3
4
5
6
7
8
9
10private static InternalHandler sHandler;
···
private static Handler getHandler() {
synchronized (AsyncTask.class) {
if (sHandler == null) {
sHandler = new InternalHandler();
}
return sHandler;
}
}
这里就是了，在构造方法里面super(Looper.getMainLooper())，也就说明了这个方法是在主线程中执行的，在主线程中对Message进行处理，这里又涉及到两个方法，一个是finish()，还有一个是onProgressUpdate()，那么好吧，再去看看这两个方法在干啥1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23private static final int MESSAGE_POST_RESULT = 0x1;
private static final int MESSAGE_POST_PROGRESS = 0x2;
···
private static class InternalHandler extends Handler {
public InternalHandler() {
super(Looper.getMainLooper());
}
({"unchecked", "RawUseOfParameterizedType"})
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;
}
}
}
首先是又调了isCancelled()，判断是否取消，前面构造函数的时候见过这个，这是在异常发生的时候才设置为true的，那么如果不发生异常，这里应该就是为false的，但在找源代码时发现，另一个方法也对这个参数进行了设置，那就是cancel()，所以在不发生异常和取消的时候应该是为true的，接下来是onCancelled()方法，这里是不做任何操作的，这也是主线程中的方法，还有就是onPostExecute()方法，然后会设置状态，其默认状态是PENDING1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16private volatile Status mStatus = Status.PENDING;
···
public enum Status {
PENDING,
RUNNING,
FINISHED,
}
···
private void finish(Result result) {
if (isCancelled()) {
onCancelled(result);
} else {
onPostExecute(result);
}
mStatus = Status.FINISHED;
}
1 | ({"UnusedDeclaration"}) |
然后是onProgressUpdate()方法，那么这里做了啥呢，嗯~啥也没有，交给调用者在继承时可以使用1
2
3
4({"UnusedDeclaration"})
protected void onProgressUpdate(Progress... values) {
}
来看构造方法中涉及到的最后一个方法postResultIfNotInvoked()，这个方法又干了啥了，首先获得了mTaskInvoked的状态，整个AsyncTask只有构造方法处设置了这个值，然后判断是否执行postResult()方法1
2
3
4
5
6private void postResultIfNotInvoked(Result result) {
final boolean wasTaskInvoked = mTaskInvoked.get();
if (!wasTaskInvoked) {
postResult(result);
}
}
至此构造方法分析完成，可以看到在构造方法中，其主要做的工作最主要的就是搭建好了子线程和主线程沟通的桥梁
执行入口分析
在新建AsyncTask对象以后，要执行的话，需要使用execute()去开始执行
那么我们就从这里入手，看看其具体是怎么工作的，可以看到无论是构造方法还是启动方法，都是需要在主线程中完成的，在execute()中，做了些什么呢，在这之前我们先看看传递的sDefaultExecutor是啥1
2
3
4
public final AsyncTask<Params, Progress, Result> execute(Params... params) {
return executeOnExecutor(sDefaultExecutor, params);
}
这其实是一个线程池，任务调度的线程池，可以看到SerialExecutor实际上是实现了Executor接口，其作用就是将任务添加到双向队列，然后不断地取出执行取出执行，那么THREAD_POOL_EXECUTOR也应该是一个线程池，那这又是啥呢，去看一看这个玩意儿就是到了1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
private static volatile Executor sDefaultExecutor = SERIAL_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()方法
scheduleNext();
}
}
});
if (mActive == null) {
scheduleNext();
}
}
//从队列中取出线程并执行
protected synchronized void scheduleNext() {
if ((mActive = mTasks.poll()) != null) {
THREAD_POOL_EXECUTOR.execute(mActive);
}
}
}
下列代码就是初始化了线程池的参数，指定了线程数量1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32//获得可用CPU数量
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
//设置核心线程池数量其范围[2,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;
//设置线程工厂，通过这个创建线程
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);
//设置线程池
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;
}
所以在执行scheduleNext()的时候，会将THREAD_POOL_EXECUTOR中设置好的线程全部取出来，用来执行后面的任务，其执行的任务就是execute()方法所指定的任务，在executeOnExecutor()方法中，由于前面初始化完成，这里的状态应该是PENDING，之后还设置了mWorker的参数，然后会执行线程池的方法，然后据开始执行任务了，前面没有涉及到的方法还有一个，那我们接下来看看1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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;
}
onPreExecute()，这个方法由调用者在继承时候能够使用1
2
3
protected void onPreExecute() {
}
至此，AsyncTask的执行方法也分析完了，那么我们接下来看看还有什么方法没有涉及到，没有涉及到的方法都是public属性和方法
AsyncTask的公共方法
1 | //这个方法设置为public，那么就意味着我们可以自定义线程池 |
1 | //这个方法意味着我们可以获得其状态，配合枚举值使用 |
1 | //查看是非被取消 |
1 | //取消异步任务 |
1 | //获取返回结果，注意：这个方法是阻塞式的 |
1 | //同上 |
1 | //自定义的线程池可以从这个方法启动 |
1 | //使用默认线程池启动异步任务 |
总结
AsyncTask的实例化过程，其本质上就是实例化了一个FutureTask
其执行过程Executor.execute(mFuture) -> SerialExecutor.mTasks(队列) -> (线程池)THREAD_POOL_EXECUTOR.execute
线程池中的所有线程，为了执行异步任务
如果当前线程池中的数量小于corePoolSize，创建并添加的任务
如果当前线程池中的数量等于corePoolSize，缓冲队列workQueue未满，那么任务被放入缓冲队列、等待任务调度执行
如果当前线程池中的数量大于corePoolSize，缓冲队列workQueue已满，并且线程池中的数量小于maximumPoolSize，新提交任务会创建新线程执行任务
如果当前线程池中的数量大于corePoolSize，缓冲队列workQueue已满，并且线程池中的数量等于maximumPoolSize，新提交任务由Handler处理
当线程池中的线程大于corePoolSize时，多余线程空闲时间超过keepAliveTime时，会关闭这部分线程
线程池在添加时候是串行的，在执行任务的时候是并行的
附录(源代码)
1 | package android.os; |
