写在前记：
总结：相对于AsyncTask 比较常用的地方是用于IO密集型的，cpu占用率不会很高的地方
AsyncTask是什么
- android sdk 封装的一个异步任务方案
- 模板类
- 提供了异步线程与UI线程的通信方式
实现方式
AsyncTask.java
public abstract class AyncTask<Params, Progress, Result>{
// 泛型的参数说明
// 1. Param 对应excute 中传递参数的类型
// 2. Progress: 异步执行过程中返回进度值， 例如下载进度值的类型
// 3. Result: 异步任务执行完成后的参数类型，doBackInBackground(param ...) 参数类型
}
先回顾一下并发的几个接口
Callable， Runnable, Future, FutureTask
Callable && Runnable 两者之间的区别是
- 在执行任务后是否有返回值
- callable的返回值类型通过泛型的方式约定
public interface Runnable{
void run();
}
// 与runnable的区别在于，有返回值
public interface Callable<V>{
V call() throw Exception;
}
public interface ExecutorService{
<T> Future<T> submit(Callable<T> task)
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable tasl);
}
// future作用，可以监控目标线程调用call，当调用get()的时候，当前线程开始阻塞，知道
public interface Future<V>{
bool cancel(bool mayInterruptIfRunning);
bool isCanceled();
V get();
V get(long timeout, TimeUnit unit);
}
public interface RunnableFuture<V> extends Runnable, Future<V>{
void run();
}
// RunnableFuture的实现类
public class FutureTask<V> impletements RunnableFuture<V>{
// 内部维护了一个状态机
// 一共7中，只会出现4中状态变化
//情况1 NEW->COMPLETING->NORMAL
//情况2 NEW->COMPLETING->EXCEPTIONAL
//情况3 NEW->CANCELLED
//情况4 NEW->INTERRUPTING->INTERRUPTED
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
// 看get方法是如何实现阻塞的
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
// 根据state返回结果或者抛出异常
return report(s);
}
// 核心方法
// 该方法实现了自旋
// 具体实现：
1）若支持中断，判断当前线程是否中断
1.1）中断，退出自旋，从WaitNode等待队列中移除当前节点
1.2）继续下一步
2) 判断当前状态是否完成，如完成[中断，或者正常完成或者取消]直接直接返回状态，并且将当前的thread设置为null，否则进入下一步
3）如果当前状态是正在完成中，暂时让出cpu时间片，将线程从运行态转到就绪态，否则进入下一步
4) 构造一个waitNode， 插入到队列中
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
//退出
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING)
// 让出时间片
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
// 将任务插入到队列中
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
// locksupport park 阻塞对象, 等待LockSupport.park的调用返回，可以看到这个方法在finishCompletion中调用
LockSupport.park(this);
}
}
// 该方法在前面2-6状态切换中都会被调用，正常的逻辑是在run方法中set()方法调用
private void finishCompletion() {
for (WaitNode q; (q = waiters) != null;) {
if (U.compareAndSwapObject(this, WAITERS, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
// 核心的地方，这个地方会唤醒阻塞的对象
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
// 完成
done();
callable = null; // to reduce footprint
}
public void run() {
if (state != NEW ||
!U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 异常状态
setException(ex);
}
if (ran)
// 正常状态
set(result);
}
}
}
}
以上是在执行get方法被阻塞的原理
接下来看executor中的submit(Runnable)/submit(Callable) 如何返回 RunnableFuture<T>
class abstract class AbstractExecutorService implements ExecutorService{
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
}
// FutureTask 中是如何将runable 转换成callable,通过适配器模式实现
public class Executors{
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
}
static class RunnableAdapter<T> implements Callcable<T>{
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
基于上述引入一个非常重要的锁 LockSupport
LockSupport 的使用说明
- LockSupport.unpark()
- LockSupport.park()
LockSupport 的unpark和park无先后顺序， 对于unpark在线，则park不会阻塞，会直接返回
接下来分析一下Executors.java中的几类线程池
- 区别： 核心线程数 & 最大线程数 & 容器的大小
先给结论：1.当前存活线程数小于核心线程数，则新建线程执行任务，2.当前核心线程都在忙，任务会放进队列中，
- 如果队列溢出，则新建线程，如果新建线程数大于最大线程数，则reject
上代码：
class ThreadPoolExecutor{
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 是一个原子操作的interger变量，低29位存储线程数量，高三位存储线程池的状态
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
}
未完成待续...
网友评论