Java并发编程笔记之FutureTask源码分析

时间:2022-12-05 20:26:25

FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask,直接调用其run方法或者放入线程池执行,之后可以在外部通过FutureTask的get方法异步获取执行结果,因此,FutureTask非常适合用于耗时的计算,主线程可以在完成自己的任务后,再去获取结果。另外,FutureTask还可以确保即使调用了多次run方法,它都只会执行一次Runnable或者Callable任务,或者通过cancel取消FutureTask的执行等。

类图结构如下所示:

Java并发编程笔记之FutureTask源码分析

线程池使用 FutureTask 时候需要注意的一点事,FutureTask 使用不当可能会造成调用线程一直阻塞,如何避免?

线程池使用 FutureTask 的时候如果拒绝策略设置为了 DiscardPolicyDiscardOldestPolicy并且在被拒绝的任务的 Future 对象上调用无参 get 方法那么调用线程会一直被阻塞。

下面先通过一个简单的例子来复现问题,代码如下:

public class FutureTest {

    //(1)线程池单个线程,线程池队列元素个数为1
private final static ThreadPoolExecutor executorService = new ThreadPoolExecutor(, , 1L, TimeUnit.MINUTES,
new ArrayBlockingQueue<Runnable>(),new ThreadPoolExecutor.DiscardPolicy()); public static void main(String[] args) throws Exception { //(2)添加任务one
Future futureOne = executorService.submit(new Runnable() {
@Override
public void run() { System.out.println("start runable one");
try {
Thread.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}); //(3)添加任务two
Future futureTwo = executorService.submit(new Runnable() {
@Override
public void run() {
System.out.println("start runable two");
}
}); //(4)添加任务three
Future futureThree=null;
try {
futureThree = executorService.submit(new Runnable() {
@Override
public void run() {
System.out.println("start runable three");
}
});
} catch (Exception e) {
System.out.println(e.getLocalizedMessage());
} System.out.println("task one " + futureOne.get());//(5)等待任务one执行完毕
System.out.println("task two " + futureTwo.get());//(6)等待任务two执行完毕
System.out.println("task three " + (futureThree==null?null:futureThree.get()));// (7)等待任务three执行完毕 executorService.shutdown();//(8)关闭线程池,阻塞直到所有任务执行完毕
}

运行结果如下:

Java并发编程笔记之FutureTask源码分析

代码 (1) 创建了一个单线程并且队列元素个数为 1 的线程池,并且拒绝策略设置为了DiscardPolicy

代码(2)向线程池提交了一个任务 one,那么这个任务会使用唯一的一个线程进行执行,任务在打印 start runable one后会阻塞该线程 5s.

代码(3)向线程池提交了一个任务 two,这时候会把任务 two 放入到阻塞队列

代码(4)向线程池提交任务 three,由于队列已经满了则会触发拒绝策略丢弃任务 three, 从执行结果看在任务 one 阻塞的 5s 内,主线程执行到了代码 (5) 等待任务 one 执行完毕,当任务 one 执行完毕后代码(5)返回,主线程打印出 task one null。任务 one 执行完成后线程池的唯一线程会去队列里面取出任务 two 并执行所以输出 start runable two 然后代码(6)会返回,这时候主线程输出 task two null,然后执行代码(7)等待任务 three 执行完毕,从执行结果看代码(7)会一直阻塞不会返回,至此问题产生,如果把拒绝策略修改为 DiscardOldestPolicy 也会存在有一个任务的 get 方法一直阻塞只是现在是任务 two 被阻塞。但是如果拒绝策略设置为默认的 AbortPolicy 则会正常返回,并且会输出如下结果:

Java并发编程笔记之FutureTask源码分析

要分析这个问题需要看下线程池的 submit 方法里面做了什么,submit 方法源码如下:

public Future<?> submit(Runnable task) {
...
//(1)装饰Runnable为Future对象
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
//(6)返回future对象
return ftask;
} protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
public void execute(Runnable command) {
...
//(2) 如果线程个数消息核心线程数则新增处理线程处理
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//(3)如果当前线程个数已经达到核心线程数则任务放入队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == )
addWorker(null, false);
}
//(4)尝试新增处理线程进行处理
else if (!addWorker(command, false))
reject(command);//(5)新增失败则调用拒绝策略
}

代码(1)装饰 Runnable 为 FutureTask 对象,然后调用线程池的 execute 方法。

代码 (2) 如果线程个数消息核心线程数则新增处理线程处理

代码(3)如果当前线程个数已经达到核心线程数则任务放入队列

代码(4)尝试新增处理线程进行处理,失败则进行代码(5),否者直接使用新线程处理

代码(5)执行具体拒绝策略,从这里也可以看出拒绝策略执行是使用的业务线程。

所以要分析上面例子中问题所在只需要看步骤(5)对被拒绝任务的影响,这里先看下拒绝策略 DiscardPolicy 的源码,如下:

public static class DiscardPolicy implements RejectedExecutionHandler {
  public DiscardPolicy() { }
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}

可知拒绝策略 rejectedExecution 方法里面什么都没做,所以代码(4)调用 submit 后会返回一个 future 对象,这里有必要在重新说 future 是有状态的,FutureTask 内部有一个state用来展示任务的状态,并且是volatile修饰的,future 的状态枚举值如下:

/** Possible state transitions:
* NEW -> COMPLETING -> NORMAL 正常的状态转移
* NEW -> COMPLETING -> EXCEPTIONAL 异常
* NEW -> CANCELLED 取消
* NEW -> INTERRUPTING -> INTERRUPTED 中断
*/ private volatile int state;
private static final int NEW = ;
private static final int COMPLETING = ;
private static final int NORMAL = ;
private static final int EXCEPTIONAL = ;
private static final int CANCELLED = ;
private static final int INTERRUPTING = ;
private static final int INTERRUPTED = ;

在代码(1)的时候使用 newTaskFor 方法转换 Runnable 任务为 FutureTask,而 FutureTask 的构造函数里面设置的状态就是 New。FutureTask的构造函数源码如下:

public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}

把FutureTask提交到线程池或者线程执行start时候会调用run方法,源码如下:

public void run() {

    //如果当前不是new状态,或者当前cas设置当前线程失败则返回,只有一个线程可以成功。
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
//当前状态为new 则调用任务的call方法执行任务
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);完成NEW -> COMPLETING -> EXCEPTIONAL 状态转移
} //执行任务成功则保存结果更新状态,unpark所有等待线程。
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);
}
} protected void set(V v) {
//状态从new->COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
//状态从COMPLETING-》NORMAL
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//unpark所有等待线程。
finishCompletion();
}
}

所以使用 DiscardPolicy 策略提交任务后返回了一个状态值为NEW的future对象。那么我们下面就要看下当future的无参get()方法的时候,future变为什么状态才会返回,这时候就要看一下FutureTask的get方法的源码,源码如下:

  public V get() throws InterruptedException, ExecutionException {
int s = state;
//当状态值<=COMPLETING时候需要等待,否者调用report返回
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
} 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;
}
//否者设置park超时时间
LockSupport.parkNanos(this, nanos);
}
else
//直接挂起当前线程
LockSupport.park(this);
}
} private V report(int s) throws ExecutionException {
Object x = outcome;
//状态值为NORMAL正常返回
if (s == NORMAL)
return (V)x;
//状态值大于等于CANCELLED则抛异常
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}

也就是说当 future 的状态 > COMPLETING 时候调用 get 方法才会返回,而明显 DiscardPolicy 策略在拒绝元素的时候并没有设置该 future 的状态,后面也没有其他机会可以设置该 future 的状态,所以 future 的状态一直是 NEW,所以一直不会返回,同理 DiscardOldestPolicy 策略也是这样的问题,最老的任务被淘汰时候没有设置被淘汰任务对于 future 的状态。、

在submit任务后还可以调用futuretask的cancel来取消任务:

public boolean cancel(boolean mayInterruptIfRunning) {
//只有任务是new的才能取消
if (state != NEW)
return false;
//运行时允许中断
if (mayInterruptIfRunning) {
//完成new->INTERRUPTING
if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
return false;
Thread t = runner;
if (t != null)
t.interrupt();
//完成INTERRUPTING->INTERRUPTED
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
}
//不允许中断则直接new->CANCELLED
else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
return false;
finishCompletion();
return true;
}

那么默认的 AbortPolicy 策略为啥没问题呢?

也就是说当 future 的状态 > COMPLETING 时候调用 get 方法才会返回,而明显 DiscardPolicy 策略在拒绝元素的时候并没有设置该 future 的状态,后面也没有其他机会可以设置该 future 的状态,所以 future 的状态一直是 NEW,所以一直不会返回,同理 DiscardOldestPolicy 策略也是这样的问题,最老的任务被淘汰时候没有设置被淘汰任务对于 future 的状态。

所以当使用 Future 的时候,尽量使用带超时时间的 get 方法,这样即使使用了 DiscardPolicy 拒绝策略也不至于一直等待,等待超时时间到了会自动返回的,如果非要使用不带参数的 get 方法则可以重写 DiscardPolicy 的拒绝策略在执行策略时候设置该 Future 的状态大于 COMPLETING 即可,但是查看 FutureTask 提供的方法发现只有 cancel 方法是 public 的并且可以设置 FutureTask 的状态大于 COMPLETING,重写拒绝策略具体代码可以如下:

/**
* Created by cong on 2018/7/13.
*/
public class MyRejectedExecutionHandler implements RejectedExecutionHandler {
public void rejectedExecution(Runnable runnable, ThreadPoolExecutor threadPoolExecutor) {
if (!threadPoolExecutor.isShutdown()) {
if(null != runnable && runnable instanceof FutureTask){
((FutureTask) runnable).cancel(true);
}
}
}
}

使用这个策略时候由于从 report 方法知道在 cancel 的任务上调用 get() 方法会抛出异常所以代码(7)需要使用 try-catch 捕获异常代码(7)修改为如下:

package com.hjc;

import java.util.concurrent.*;

/**
* Created by cong on 2018/7/13.
*/
public class FutureTest { //(1)线程池单个线程,线程池队列元素个数为1
private final static ThreadPoolExecutor executorService = new ThreadPoolExecutor(, , 1L, TimeUnit.MINUTES,
new ArrayBlockingQueue<Runnable>(), new MyRejectedExecutionHandler()); public static void main(String[] args) throws Exception { //(2)添加任务one
Future futureOne = executorService.submit(new Runnable() { public void run() { System.out.println("start runable one");
try {
Thread.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}); //(3)添加任务two
Future futureTwo = executorService.submit(new Runnable() { public void run() {
System.out.println("start runable two");
}
}); //(4)添加任务three
Future futureThree = null;
try {
futureThree = executorService.submit(new Runnable() { public void run() {
System.out.println("start runable three");
}
});
} catch (Exception e) {
System.out.println(e.getLocalizedMessage());
} System.out.println("task one " + futureOne.get());//(5)等待任务one执行完毕
System.out.println("task two " + futureTwo.get());//(6)等待任务two执行完毕
try{
System.out.println("task three " + (futureThree==null?null:futureThree.get()));// (7)等待任务three
}catch(Exception e){
System.out.println(e.getLocalizedMessage());
} executorService.shutdown();//(8)关闭线程池,阻塞直到所有任务执行完毕
}
}

运行结果如下:

Java并发编程笔记之FutureTask源码分析

当然这相比正常情况下多了一个异常捕获,其实最好的情况是重写拒绝策略时候设置 FutureTask 的状态为 NORMAL,但是这需要重写 FutureTask 方法了,因为 FutureTask 并没有提供接口进行设置。

Java并发编程笔记之FutureTask源码分析的更多相关文章

  1. Java并发编程笔记之CopyOnWriteArrayList源码分析

    并发包中并发List只有CopyOnWriteArrayList这一个,CopyOnWriteArrayList是一个线程安全的ArrayList,对其进行修改操作和元素迭代操作都是在底层创建一个拷贝 ...

  2. Java并发编程笔记之ThreadLocalRandom源码分析

    JDK 并发包中 ThreadLocalRandom 类原理剖析,经常使用的随机数生成器 Random 类的原理是什么?及其局限性是什么?ThreadLocalRandom 是如何利用 ThreadL ...

  3. Java并发编程笔记之ThreadLocal源码分析

    多线程的线程安全问题是微妙而且出乎意料的,因为在没有进行适当同步的情况下多线程中各个操作的顺序是不可预期的,多线程访问同一个共享变量特别容易出现并发问题,特别是多个线程需要对一个共享变量进行写入时候, ...

  4. Java并发编程笔记之SimpleDateFormat源码分析

    SimpleDateFormat 是 Java 提供的一个格式化和解析日期的工具类,日常开发中应该经常会用到,但是由于它是线程不安全的,多线程公用一个 SimpleDateFormat 实例对日期进行 ...

  5. Java并发编程笔记之Timer源码分析

    timer在JDK里面,是很早的一个API了.具有延时的,并具有周期性的任务,在newScheduledThreadPool出来之前我们一般会用Timer和TimerTask来做,但是Timer存在一 ...

  6. Java并发编程笔记之CyclicBarrier源码分析

    JUC 中 回环屏障 CyclicBarrier 的使用与分析,它也可以实现像 CountDownLatch 一样让一组线程全部到达一个状态后再全部同时执行,但是 CyclicBarrier 可以被复 ...

  7. Java并发编程笔记之PriorityBlockingQueue源码分析

    JDK 中*优先级队列PriorityBlockingQueue 内部使用堆算法保证每次出队都是优先级最高的元素,元素入队时候是如何建堆的,元素出队后如何调整堆的平衡的? PriorityBlock ...

  8. Java并发编程笔记之ArrayBlockingQueue源码分析

    JDK 中基于数组的阻塞队列 ArrayBlockingQueue 原理剖析,ArrayBlockingQueue 内部如何基于一把独占锁以及对应的两个条件变量实现出入队操作的线程安全? 首先我们先大 ...

  9. Java并发编程笔记之ReentrantLock源码分析

    ReentrantLock是可重入的独占锁,同时只能有一个线程可以获取该锁,其他获取该锁的线程会被阻塞后放入该锁的AQS阻塞队列里面. 首先我们先看一下ReentrantLock的类图结构,如下图所示 ...

随机推荐

  1. Bootstrap入门(二十三)JS插件1:模态框

    Bootstrap入门(二十三)JS插件1:模态框 1.静态实例 2.动态实例 3.模态框的尺寸和效果 4.包含表单的模态框 模态框经过了优化,更加灵活,以弹出对话框的形式出现,具有最小和最实用的功能 ...

  2. python 用户登录

    要求: 1.用户在登录之前要判断密码文件是否已经是锁定状态,如果是提示不让登录 2.登录用户密码输入错误3次,就锁定改用户,并更新文件内容 3.登录正确,提示欢迎登录信息 data.txt 文件内容: ...

  3. leetcode之word ladder

    对于之前没有接触过该类型题目的人来说,此题无疑是个难题,本人提交了10次才正确通过,期间遇到了非常多的问题,感觉几乎把OJ的所有错误遍历了一遍,下面详细说说自己做该题的经验. 首先承认,我一开始并没有 ...

  4. javascript、ruby和C性能一瞥&lpar;3&rpar; &colon;上汇编

    在博文(1)和(2)里分别用了4中方式写一个素数筛选的算法,分别是javascript in browser.node.js.ruby和c:最终的结果是c最快,node.js其次,js in b虽然也 ...

  5. Memcached技术

    Memcached技术 介绍: memcached是一种缓存技术, 他可以把你的数据放入内存,从而通过内存访问提速,因为内存最快的, memcached技术的主要目的提速, 在memachec 中维护 ...

  6. sql语句(一)— —判断是否有这条数据的优化

    今天发现一个业务上的存储过程写的不够完善,和老板反应后,老板说你来完善吧,我:苦瓜脸~.说实话,我对SQL语句的熟练程度真的是不提也罢[捂脸],大概的判断流程我知道,但是真的让我自己写,还真得上网查查 ...

  7. 关于kafka重新消费数据问题

    我们在使用consumer消费数据时,有些情况下我们需要对已经消费过的数据进行重新消费,这里介绍kafka中两种重新消费数据的方法. 1. 修改offset 我们在使用consumer消费的时候,每个 ...

  8. Android ThreadPoolExecutor线程池

    引言 Android的线程池概念来自于Java的Executor,真正的线程池实现为ThreadPoolExecutor.在Android中,提供了4类不同的线程池,具体下面会说到.为什么使用线程池呢 ...

  9. 20135337——Linux实践三:ELF文件格式(64位系统,简单分析)

    ELF文件格式简单分析 (具体分析见上一篇ELF文件格式32位系统) ELF-header 第一行: 457f 464c :魔数: 0201 :64位系统,小端法 01 :文件头版本 剩余默认0: 第 ...

  10. linux系统用户和组管理

    用户和组管理 Linux是多用户多任务的网络操作系统,作为网络管理员,掌握用户的组的创建与管理至关重要. 学习要点: 了解用户和组的群的配置文件. 熟悉掌握Linux下用户的创建和维护管理. 熟悉掌握 ...