ps: 以前好多东西都写在一起了，现在拆开
我们使用 Runnable 可以构建一个 Thread 对象
// 普通线程任务
Runnable runnable = new Runnable() {
@Override
public void run() {
Log.d("AA", Thread.currentThread().getName() + "启动,time:" + System.currentTimeMillis());
}
};
但是 Runnable 不能给我们一个返回值，基本上我们使用多线程都需要返回值的，所以就引出了下面这3个类：
- Callable - 构建带返回值的任务
- Future - 和异步线程通信的接口
- FutureTask - Future 的唯一实现类，通过 get() 可以拿到结果，但是阻塞当前线程
public interface Callable<V> {
V call() throws Exception;
}
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;
}
先来跑个小例子，先启动一个线程，睡眠2秒，之后返回结果，我们在 UI 线程里等待结果(此时会阻塞 UI 线程) ，收到结果后再启动一个线程，2个线程都打印下时间
// 带返回结果的线程任务，String 泛型是我们预订的返回数据类型
Callable<String> callable1 = new Callable<String>() {
@Override
public String call() throws Exception {
Log.d("AA", Thread.currentThread().getName() + "启动,time:" + System.currentTimeMillis());
Thread.sleep(2000);
return "AA";
}
};
// 用 Callable 对象构建一个 FutureTask 线程返回结果
FutureTask<String> futureTask = new FutureTask<String>(callable1);
// 创建出2个线程
Thread thread1 = new Thread(futureTask);
Thread thread2 = new Thread(runnable);
// 先启动会返回结果的线程
thread1.start();
try {
// 这里阻塞 UI 线程，等待线程1 的返回数据，再去启动线程2
futureTask.get();
thread2.start();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
结果
ok 看过例子我们就知道怎么回事了，然后我们再继续深入吧，我们重点是看看 FutureTask 给我们提供的方法：
- get() - 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成
- get(Long timeout , TimeUnit unit) - 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的timeout时间，该方法将返回null
- isDone() - 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回tru
- isCanceller() 如果任务完成前被取消，则返回true
- cancel(boolean mayInterruptRunning) - 取消任务，cancel(false) 等待任务执行完毕然后回收资源，cancel(true) 尝试直接终中断任务
FutureT 执行示意图
上面是结合 Thread 的使用场景，但是一般我们真不这么写，我们都是在线程池中开干
Callable<String> callable = new Callable<String>() {
@Override
public String call() throws Exception {
return "AA";
}
};
// new 一个线程池对象
ExecutorService executorService = Executors.newFixedThreadPool(5);
// 把任务添加到线程池中去执行，获取到一个 Future 任务结果操作对象
Future<String> future = executorService.submit(callable);
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
Log.d("AA", Thread.currentThread().getName() + "获取到异步线程结果,time:" + System.currentTimeMillis());
线程池中关于 FutureTask 的使用
FutureTask 有必要说意下，因为 FutureTask 带给我们在线程池外堆线程的控制手段，我觉得搞清楚很有必要
1. 线程池中添加任务的2种方式：
//Runnable
fun main(args: Array<String>) {
val pool= Executors.newFixedThreadPool(3)
pool.execute{
Thread.sleep(2000)
println("无返回值")
pool.shutdown()
}
}
//FutureTask
fun main(args: Array<String>) {
val pool= Executors.newFixedThreadPool(3)
val futureTask=FutureTask<String>(Callable<String>{
Thread.sleep(2000)
return@Callable "有返回值"
})
pool.submit(futureTask)
//block current thread
val result=futureTask.get()
println(result)
pool.shutdown()
}
2. 添加 FutureTask 到线程池中后的执行特征
FutureTask.get() 会阻塞所在线程，那么我们是不是每次调用 get() 都会阻塞所在线程？
我们设计一个 demo 自己跑起来试试：我们往线程池中添加一个 FutureTask 任务，等待 FutureTask 任务执行时间过后再点击按钮获取数据
class Main2Activity : AppCompatActivity() {
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main2)
val pool = Executors.newFixedThreadPool(3)
val task = FutureTask<String>({
Log.d("AA", "FutureTask_111：${Thread.currentThread().name}")
Thread.sleep(1000)
Log.d("AA", "FutureTask_222")
Log.d("AA", "FutureTask_333")
return@FutureTask "AA"
})
pool.submit(task)
btn_right.setOnClickListener {
val data = task.get()
Log.d("AA", "task 执行结果：$data")
}
}
}
-
页面启动后，添加入线程池的任务会直接执行，才不会管你 get 与否，线程池只认 Runnable 接口，由任务就执行
页面启动后打印
-
我们等几秒点击按钮，不管按几次都没有阻塞当前线程，而是直接获取到结果，这说明 FutureTask 会保存任务结算的结果，并且会判断任务进度，任务执行结束之后就不再阻塞当前线程了
多次点击按钮
3. 探寻 FutureTask 的执行过程
基于上面的执行结果，我们心里清楚了 FutureTask 的运行特征，这样在使用时就不会有没有由来的担忧了，怕会不会这样，会不会那样，如果你有这样的疑问，说明这个知识点你没学透，言归正传，这激起了我对 FutureTask 探寻一番的好奇
3.1 先来看 run 方法：
FutureTask 实现了 Runnable 接口，自然也是有 run 方法的
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//调用callback的方法，并获取执行结果
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);//将执行结果设置到实例变量中。
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
- run 方法首先判断了状态，不是 new 直接退出
- 之后调了 callable.call() 获取结果
- 完事用 set(result) 保存结果，这说明 FutureTask 的确会保存结果值
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//保存执行结果
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//唤醒当前线程。
finishCompletion();
}
}
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, 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 V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
//阻塞当前线程
s = awaitDone(false, 0L);
//当线程从阻塞状态被唤醒，将结果返回。
return report(s);
}
//Awaits completion or aborts on interrupt or timeout.
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) // cannot time out yet
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(this);
}
}
- 首先又是判断了任务状态，不是已经完成就阻塞线程，但是任务结速之后直接返回结果，就不再阻塞线程了，这和我们上面的执行特性吻合
- awaitDone 里面就是根据任务的状态各种阻塞线程
简单的看过源码之后，我想大家对于 FutureTask 就能做到比较通透的了解了，至少在使用时不会再束手束脚了，放心大胆的使用，别人问起也知道问题在哪，怎么说，我觉得这就是我们学习源码最大的动力和目的，搞清楚运行机制
FutureTask 也有自己的回调函数
FutureTask 自己的回调函数就是 done 这个方法，我们可以继承 FutureTask 来重写 done 这个函数实现自己的目的
比如下面我简单实现一个：
class MyFutureTask<V>(callable: Callable<V>) : FutureTask<V>(callable) {
override fun done() {
Log.d("AA", "futrueTask 回调执行，所在线程：${Thread.currentThread().name}")
}
}
val pool = Executors.newFixedThreadPool(3)
val task = MyFutureTask<String>(object : Callable<String> {
override fun call(): String {
Log.d("AA", "FutureTask_111：${Thread.currentThread().name}")
Thread.sleep(1000)
Log.d("AA", "FutureTask_222")
Log.d("AA", "FutureTask_333")
return "AA"
}
})
btn_right.setOnClickListener({
pool.submit(task)
val data = task.get()
Log.d("AA", "task 执行结果：$data")
})
Snip20190425_18.png
从结果上看 done 的确是在最后执行的，但是吧不是我们期盼的在线程池调用线程执行，而是在线程池所在线程执行，这样的话，这个 done 的应用范围就很狭窄了，至于 done 是在哪里调的 finishCompletion 最后是也
网友评论