java.util.concurrent解析——FutureTask源码解析

本文涉及的产品
全局流量管理 GTM,标准版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
云解析 DNS,旗舰版 1个月
简介: 和Java异步打交道就不能回避掉`Runnable`,`Callable`,`Future`,`FutureTask`等类,首先来介绍下这几个类的区别

1. Runnable、Callable、Future、FutureTask的区别与联系

和Java异步打交道就不能回避掉Runnable,Callable,Future,FutureTask等类,首先来介绍下这几个类的区别。

1.1 Runnable

Runnable接口是我们最熟悉的,它只有一个run函数。然后使用某个线程去执行该runnable即可实现多线程,Thread类在调用start()函数后就是执行的是Runnable的run()函数。Runnable最大的缺点在于run函数没有返回值。

1.2 Callable

Callable接口和Runnable接口类似,它有一个call函数。使用某个线程执行Callable接口实质就是执行其call函数。call方法和run方法最大的区别就是call方法有返回值:

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;  
} 

1.3 Future

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果、设置结果操作。get方法会阻塞,直到任务返回结果(Future简介)。

1.4 FutureTask

Future只是一个接口,在实际使用过程中,诸如ThreadPoolExecutor返回的都是一个FutureTask实例。

public class FutureTask<V> implements RunnableFuture<V>  

public interface RunnableFuture<V> extends Runnable, Future<V> {  
    /** 
     * Sets this Future to the result of its computation 
     * unless it has been cancelled. 
     */  
    void run();  
}  

可以看到,FutureTask是一个RunnableFuture,而RunnableFuture实现了Runnbale又实现了Futrue这两个接口。

2 FutureTask的构造过程

事实上,通过ExecutorService接口的相关submit方法,实际上都是提交的Callable或者Runnable,包装成一个FutureTask对象。

public abstract class AbstractExecutorService implements ExecutorService {
  ...
  //将Runable包装成FutureTask之后,再调用execute方法
  public Future<?> submit(Runnable task) {
      if (task == null) throw new NullPointerException();
      RunnableFuture<Void> ftask = newTaskFor(task, null);
      execute(ftask);
      return ftask;
  }
  
  //调用newTaskFor方法,利用Callable构造一个FutureTask对象
  protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
      return new FutureTask<T>(callable);
  }
}

可以看到AbstractExecutorService的submit方法调用后返回的就是一个FutureTask对象,接下来看下FutureTask的构造方法:

//接受Callable对象作为参数
public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;    
}
//接受Runnable对象作为参数
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);//将Runnable转为Callable对象
    this.state = NEW;     
}

//callable方法,将Runnable转为一个Callable对象,包装设计模式
public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}
 
//RunnableAdapter是Executors的一个内部类,实现了Callable接口
static final class RunnableAdapter<T> implements Callable<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;
    }
}

可以看到,构造FutureTask时,无论传入的是Runnable还是Callable,最终都实现了Callable接口。

3 FutureTask主要成员

接下来看下FutureTask类的主要成员变量:

public class FutureTask<V> implements RunnableFuture<V> {
     /*
     * FutureTask中定义了一个state变量,用于记录任务执行的相关状态 ,状态的变化过程如下
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
    private volatile int state;
    //主流程状态
    private static final int NEW = 0; //当FutureTask实例刚刚创建到callbale的call方法执行完成前,处于此状态
    private static final int COMPLETING  = 1; //callable的call方法执行完成或出现异常时,首先进行此状态
    private static final int NORMAL    = 2;//callable的call方法正常结束时,进入此状态,将outcom设置为正常结果
    private static final int EXCEPTIONAL = 3;//callable的call方法异常结束时,进入此状态,将outcome设置为抛出的异常
    //取消任务执行时可能处于的状态
    private static final int CANCELLED= 4;// FutureTask任务尚未执行,即还在任务队列的时候,调用了cancel方法,进入此状态
    private static final int INTERRUPTING = 5;// FutureTask的run方法已经在执行,收到中断信号,进入此状态
    private static final int INTERRUPTED  = 6;// 任务成功中断后,进入此状态
    
    private Callable<V> callable;//需要执行的任务,提示:如果提交的是Runnable对象,会先转换为Callable对象,这是构造方法参数
    private Object outcome; //任务运行的结果
    private volatile Thread runner;//执行此任务的线程
  
    //等待该FutureTask的线程链表,对于同一个FutureTask,如果多个线程调用了get方法,对应的线程都会加入到waiters链表中,同时当FutureTask执行完成后,也会告知所有waiters中的线程
    private volatile WaitNode waiters;
    ......
}

FutureTask的成员变量并不复杂,主要记录以下几部分信息:

  • 状态
  • 任务(callable)
  • 结果(outcome)
  • 等待线程(waiters)

4 FutureTask的执行过程

4.1 run

接下来开始看一个FutureTask的执行过程,FutureTask执行任务的方法当然还是run方法:

public void run() {
    //保证callable任务只被运行一次
    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 {
                //执行任务
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
        runner = null;
        int s = state;
        //判断该任务是否正在响应中断,如果中断没有完成,则等待中断操作完成
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}
  • 如果状态不为new或者运行线程runner失败,说明当前任务已经被其他线程启动或者已经被执行过,直接返回false
  • 调用call方法执行核心任务逻辑。如果调用成功则执行set(result)方法,将state状态设置成NORMAL。如果调用失败抛出异常则执行setException(ex)方法,将state状态设置成EXCEPTIONAL,唤醒所有在get()方法上等待的线程
  • 如果当前状态为INTERRUPTING(步骤2已CAS失败),则一直调用Thread.yield()直至状态不为INTERRUPTING

4.2 set、setException方法

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}

两个方法的逻辑基本一致,先通过CAS操作将状态从NEW置为COMPLETING,然后再将最终状态分别置为NORMAL或者EXCEPTIONAL,最后再调用finishCompletion方法。

4.3 finishCompletion

private void finishCompletion() {
    for (WaitNode q; (q = waiters) != null;) {
        //通过CAS把栈顶的元素置为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
}

finishCompletion的逻辑也比较简单:

  • 遍历waiters链表,取出每一个节点:每个节点都代表一个正在等待该FutureTask结果(即调用过get方法)的线程
  • 通过 LockSupport.unpark(t)唤醒每一个节点,通知每个线程,该任务执行完成

4.4 get

在finishCompletion方法中,FutureTask会通知waiters链表中的每一个等待线程,那么这些线程是怎么被加入到waiters链表中的呢?上文已经讲过,当在一个线程中调用了get方法,该线程就会被加入到waiters链表中。所以接下来看下get方法:

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

get方法很简答,主要就是调用awaitDone方法:

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        //如果该线程执行interrupt()方法,则从队列中移除该节点,并抛出异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        //如果state状态大于COMPLETING 则说明任务执行完成,或取消
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        //如果state=COMPLETING,则使用yield,因为此状态的时间特别短,通过yield比挂起响应更快。
        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);
        //如果需要阻塞指定时间,则使用LockSupport.parkNanos阻塞指定时间
        //如果到指定时间还没执行完,则从队列中移除该节点,并返回当前状态
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            //阻塞当前线程
            else
                LockSupport.park(this);
        }
}

整个方法的大致逻辑主要分为以下几步:

  • 如果当前状态值大于COMPLETING,说明已经执行完成或者取消,直接返回
  • 如果state=COMPLETING,则使用yield,因为此状态的时间特别短,通过yield比挂起响应更快
  • 如果当前线程是首次进入循环,为当前线程创建wait节点加入到waiters链表中
  • 根据是否定时将当前线程挂起(LockSupport.parkNanos LockSupport.park)来阻塞当前线程,直到超时或者线程被finishCompletion方法唤醒
  • 当线程挂起超时或者被唤醒后,重新循环执行上述逻辑

get方法是FutureTask中的关键方法,了解了get方法逻辑也就了解为什么当调用get方法时线程会被阻塞直到任务运行完成。

4.5 cancel

cancel方法用于结束当前任务:

public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}
  • 根据mayInterruptIfRunning是否为true,CAS设置状态为INTERRUPTING或CANCELLED,设置成功,继续第二步,否则返回false
  • 如果mayInterruptIfRunning为true,调用runner.interupt(),设置状态为INTERRUPTED
  • 唤醒所有在get()方法等待的线程
相关文章
|
4天前
|
监控 Java 应用服务中间件
高级java面试---spring.factories文件的解析源码API机制
【11月更文挑战第20天】Spring Boot是一个用于快速构建基于Spring框架的应用程序的开源框架。它通过自动配置、起步依赖和内嵌服务器等特性,极大地简化了Spring应用的开发和部署过程。本文将深入探讨Spring Boot的背景历史、业务场景、功能点以及底层原理,并通过Java代码手写模拟Spring Boot的启动过程,特别是spring.factories文件的解析源码API机制。
16 2
|
8天前
|
Java
轻松上手Java字节码编辑:IDEA插件VisualClassBytes全方位解析
本插件VisualClassBytes可修改class字节码,包括class信息、字段信息、内部类,常量池和方法等。
54 6
|
5天前
|
存储 算法 Java
Java Set深度解析:为何它能成为“无重复”的代名词?
Java的集合框架中,Set接口以其“无重复”特性著称。本文解析了Set的实现原理,包括HashSet和TreeSet的不同数据结构和算法,以及如何通过示例代码实现最佳实践。选择合适的Set实现类和正确实现自定义对象的hashCode()和equals()方法是关键。
18 4
|
4天前
|
存储 安全 Linux
Golang的GMP调度模型与源码解析
【11月更文挑战第11天】GMP 调度模型是 Go 语言运行时系统的核心部分,用于高效管理和调度大量协程(goroutine)。它通过少量的操作系统线程(M)和逻辑处理器(P)来调度大量的轻量级协程(G),从而实现高性能的并发处理。GMP 模型通过本地队列和全局队列来减少锁竞争,提高调度效率。在 Go 源码中,`runtime.h` 文件定义了关键数据结构,`schedule()` 和 `findrunnable()` 函数实现了核心调度逻辑。通过深入研究 GMP 模型,可以更好地理解 Go 语言的并发机制。
|
8天前
|
人工智能 监控 数据可视化
Java智慧工地信息管理平台源码 智慧工地信息化解决方案SaaS源码 支持二次开发
智慧工地系统是依托物联网、互联网、AI、可视化建立的大数据管理平台,是一种全新的管理模式,能够实现劳务管理、安全施工、绿色施工的智能化和互联网化。围绕施工现场管理的人、机、料、法、环五大维度,以及施工过程管理的进度、质量、安全三大体系为基础应用,实现全面高效的工程管理需求,满足工地多角色、多视角的有效监管,实现工程建设管理的降本增效,为监管平台提供数据支撑。
25 3
|
8天前
|
Java 编译器 数据库连接
Java中的异常处理机制深度解析####
本文深入探讨了Java编程语言中异常处理机制的核心原理、类型及其最佳实践,旨在帮助开发者更好地理解和应用这一关键特性。通过实例分析,揭示了try-catch-finally结构的重要性,以及如何利用自定义异常提升代码的健壮性和可读性。文章还讨论了异常处理在大型项目中的最佳实践,为提高软件质量提供指导。 ####
|
11天前
|
存储 Java 开发者
Java中的集合框架深入解析
【10月更文挑战第32天】本文旨在为读者揭开Java集合框架的神秘面纱,通过深入浅出的方式介绍其内部结构与运作机制。我们将从集合框架的设计哲学出发,探讨其如何影响我们的编程实践,并配以代码示例,展示如何在真实场景中应用这些知识。无论你是Java新手还是资深开发者,这篇文章都将为你提供新的视角和实用技巧。
12 0
|
9天前
|
安全 Java 测试技术
Java并行流陷阱:为什么指定线程池可能是个坏主意
本文探讨了Java并行流的使用陷阱,尤其是指定线程池的问题。文章分析了并行流的设计思想,指出了指定线程池的弊端,并提供了使用CompletableFuture等替代方案。同时,介绍了Parallel Collector库在处理阻塞任务时的优势和特点。
|
5天前
|
安全 Java 开发者
深入解读JAVA多线程:wait()、notify()、notifyAll()的奥秘
在Java多线程编程中,`wait()`、`notify()`和`notifyAll()`方法是实现线程间通信和同步的关键机制。这些方法定义在`java.lang.Object`类中,每个Java对象都可以作为线程间通信的媒介。本文将详细解析这三个方法的使用方法和最佳实践,帮助开发者更高效地进行多线程编程。 示例代码展示了如何在同步方法中使用这些方法,确保线程安全和高效的通信。
25 9

推荐镜像

更多