结合线程池细说FutureTask及Future源码

2020-10-10 440

1.FutureTask继承体系结构

2.从简单示例开始分析

多线程Runnable和Callable接口这里就不多说了，Callable有返回值，Runnable无返回值。

public class FutureTaskTest {

    public static void main(String[] args) {
        ExecutorService executor = null;


        try {
            //线程池提交Runnable接口任务
            executor.execute(new MyRunnable());
            //线程池提交Callable接口任务
            executor = Executors.newFixedThreadPool(2);
            Future f = executor.submit(new MyCallLable<Integer>());
            System.out.println(f.get());

            //单线程方式
            FutureTask ft = new FutureTask(new MyCallLable<Integer>());
            Thread t = new Thread(ft);
            t.start();
            System.out.println(ft.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            if (executor != null) {
                executor.shutdown();
            }
        }

    }

    static class MyCallLable<Integer> implements Callable {
        @Override
        public Object call() throws Exception {
            return 1;
        }
    }

    static class MyRunnable implements  Runnable {
        @Override
        public void run() {
            System.out.println(2);
        }
    }

}

3.以线程池提交方式分析FutureTask源码

3.1.Executors.newFixedThreadPool分析

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

该方法创建了一个核心线程和最大线程数一样的线程池，使用LinkedBlockingQueue这种无界队列存储多余的任务，也就是说，如果我们使用这种jdk自带的线程提交任务的时候，由于队列是无界的，当任务达到一定数量会造成内存溢出。这里不再分析ThreadPoolExecutor代码，有兴趣的可以看我的另一篇博文专门分析ThreadPoolExecutor源码的。该方法返回一个ExecutorService。

ThreadPoolExecutor继承体系如下图:

3.2.ExecutorService.submit方法分析

该方法实际调用的是实现类AbstractExecutorService.submit方法

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

这里的newTaskFor方法就会将Callable任务传递到FutureTask类中，并封装到其Callable属性中

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

3.3.FutureTask属性分析

    /* 线程状态可能的转换:
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
     */
//当前任务状态
private volatile int state;
//新创建
private static final int NEW          = 0;
//即将结束，但还没有结束
private static final int COMPLETING   = 1;
//正常结束
private static final int NORMAL       = 2;
//异常状态:Callable接口的Call方法中具体业务逻辑出现异常
private static final int EXCEPTIONAL  = 3;
//任务被取消
private static final int CANCELLED    = 4;
//任务处于中断中
private static final int INTERRUPTING = 5;
//任务被中断
private static final int INTERRUPTED  = 6;

//任务提交传入的Callable，用来调用call方法
private Callable<V> callable;
//Call方法返回值
//1.如果任务正常结束，返回call方法的返回值
//2.如果call方法发生异常，返回具体的异常信息
private Object outcome;
//当前执行的线程
private volatile Thread runner;
//一个栈结构的数据类型，存储被get方法阻塞的线程的引用
private volatile WaitNode waiters;

3.4.FutureTask构造方法分析

public FutureTask(Callable<V> callable) {
    //外部需要传入Callable接口的实现
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    //将线程状态设置为先创建
    this.state = NEW;
}

3.5.FutureTask执行过程分析

3.5.1.使用线程池提交Callable接口情况下调用过程分析

从示例的线程池提交Calllable接口的案例中一步步分析:
1.executor.submit(new MyCallLable<Integer>())方法提交一个Callable实现；
2.第一步实际会调用AbstractExecutorService.submit方法；
3.AbstractExecutorService.submit内部调用newTaskFor方法生成一个FutureTask对象，并将MyCallLable任务封装到其Calllable属性中；
4.AbstractExecutorService.submit方法内部调用ThreadPoolExecutor.execute方法提交FutureTask对象到线程池；
5-6-7-8.实际就是线程池提交一个任务的执行过程，具体源码可以看我的另一篇博客，这里比较复杂，概况的说了下;
9-10.线程池execute实际会执行FutureTask的run方法，在run方法中调用Callable.call,这就是线程池提交Callable执行的流程；

3.5.2.FutureTask.run方法分析

public void run() {
    //条件1:当前任务状态不是新建状态
    //条件2:当前线程不是FutureTask持有的线程
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        //退出执行
        return;
    try {
        //当前FutureTask持有的callable
        Callable<V> c = callable;
        //条件1：当前提交的Callable不能为空
        //条件2：当前线程任务状态为新创建
        if (c != null && state == NEW) {
            //Callable的返回值
            V result;
            //任务是否成功执行
            boolean ran;
            try {
                //调用用户自定义call方法的逻辑
                result = c.call();
                //任务成功执行
                ran = true;
            } catch (Throwable ex) {
                //发生异常
                result = null;
                ran = false;
                setException(ex);
            }
            //任务成功执行设置返回值
            if (ran)
                set(result);
        }
    } finally {
        //run方法结束持有线程设置为空，help gc
        //这里可能正常执行完run方法也可能出现异常退出
        runner = null;
        //当前任务执行状态
        int s = state;
        //如果处于中断的状态，包含中断中和已中断，释放cpu资源
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

3.5.3.FutureTask.set方法分析

该方法设置任务成功执行后的执行结果状态和返回值，将返回值封装到outcome属性中，由于get方法是阻塞的，还需要唤醒阻塞的线程。

protected void set(V v) {
    //将状态从新建设置为结束中
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //返回值赋值
        outcome = v;
        //设置任务状态为正常结束
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); 
        //唤醒被get方法阻塞的线程
        finishCompletion();
    }
}

3.5.6.FutureTask静态内部类WaitNode分析

在分析finishCompletion方法前，先介绍下WaitNode类。为什么会有这个类？我们知道FutureTask.get方法是阻塞的，如果我们在一个线程内多次调用get方法，这个从理论上考虑其实不需要WaitNode的；如果我们又多次创建了线程在其他线程内部调用get方法呢？由于FutureTask.get方法内部会调用LockSupport.park(Thread)或LockSupport.parkNanos阻塞线程，所以就需要唤醒；而LockSupport.unpark(Thread)解除线程阻塞也需要指定线程，所以这里就需要一种数据结构来存储当前线程的引用了。这里就设计了WaitNode这个类，它是一个单链表，而且采用的是头插法，在遍历的时候也是从前往后遍历的，这就是一个典型的栈的结构，先进后出，后进先出。这里为什么又是一个单链表结构呢？这是为了方便在任务结束的时候遍历。

static final class WaitNode {
    //当前线程的引用
    volatile Thread thread;
    //指向下一个节点
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

3.5.7.FutureTask.finishCompletion方法分析

用于唤醒被get方法阻塞的线程

private void finishCompletion() {
    // assert state > COMPLETING;
    //从头开始遍历
    for (WaitNode q; (q = waiters) != null;) {
        //使用cas方式设置当前waiters为空,防止外部线程调用cancel导致finishCompletion该方法被调用
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                //获取当前WaitNode对应的线程
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null; //help gc
                    //唤醒当前节点对应的线程
                    LockSupport.unpark(t);
                }
                //获取当前节点的下一个节点
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null;//help gc
                //将q指向下要给节点
                q = next;
            }
            break;
        }
    }

    done();
    //将callable置为空，help gc
    callable = null; 
}

3.5.8.FutureTask.setException方法分析

该方法将返回值设置为抛出的异常，将任务状态设置为EXCEPTIONAL状态，并调用finishCompletion方法唤醒被get阻塞的线程。

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

3.5.9.FutureTask.handlePossibleCancellationInterrupt方法分析

private void handlePossibleCancellationInterrupt(int s) {
    //如果任务状态处于中断中，释放cpu资源
    if (s == INTERRUPTING)
        while (state == INTERRUPTING)
            Thread.yield(); // wait out pending interrupt
}

3.5.9.FutureTask.get和FutureTask.get(long timeout, TimeUnit unit)方法分析

两个方法区别不大，唯一的区别是阻塞线程的时候使用的LockSupport.parkNanos(this, nanos)和LockSupport.park(this)，当有时间条件的时候LockSupport.parkNanos(this, nanos)会在指定时间内结束后自动唤醒线程。

这里讲讲sleep和LockSupport.parkNanos区别：sleep在指定时间到期后会判断中断状态，根据中断状态来判断是否需要抛出异常，而LockSupport.parkNanos不会根据中断状态做出响应。

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    //unit.toNanos(timeout)将指定时间格式转化为对应的毫微秒
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

3.6.0.FutureTask.awaitDone方法分析

t.interrupted()也是可以唤醒被LockSupport.park()阻塞的线程的

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? S
        ystem.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    //自旋
    for (;;) {
        //条件成立说明当前线程是被其他线程调用t.interrupted()这种中断方式唤醒
        if (Thread.interrupted()) {
            //从队列中移除线程被中断的节点
            removeWaiter(q);
            throw new InterruptedException();
        }
        
        int s = state;
        //(4).s>COMPLETING成立，说明当前任务已经执行完，结果可能有好有坏
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            //返回当前任务状态
            return s;
        }
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        //(1).第一次自旋，q=null，创建当前线程对应的WaitNode对象
        else if (q == null)
            q = new WaitNode();
        //(2).第二次自旋，queued为false，q.next = waiters采用头插法将当前节点入栈
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        //(3).第三次自旋，会走到这里，将线程阻塞，等待后续唤醒后继续自旋调用，也可能因为超时后自动唤醒
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                //从队列中移除get超时的节点
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            LockSupport.park(this);
    }
}

3.6.1.FutureTask.removeWaiter方法分析

每次调用get方法都会将线程封装成WaitNode入栈，当调用get方法的线程由于被中断唤醒或者超时自动唤醒的都需要从队列中移除, 并重新组装栈结构。

一张图概况该方法做的事情:

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                if (q.thread != null)
                    pred = q;
                else if (pred != null) {
                    pred.next = s;
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}

3.6.2.FutureTask.report方法分析

将返回值封装到outcome属性中返回，可能是正常的值也可能是一个异常信息

private V report(int s) throws ExecutionException {
    Object x = outcome;
    if (s == NORMAL)
        return (V)x;
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

3.6.3.FutureTask.cancel方法分析

public boolean cancel(boolean mayInterruptIfRunning) {
    //条件1:说明当前任务处于运行中
    //条件2:任务状态修改
    //条件1和条件2成立则执行下面cancel的核心处理逻辑，否则返回false代表取消失败
    //可能会有多个线程调用cancel方法导致cancel失败的情况
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        //mayInterruptIfRunning是否中断线程
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    //中断线程
                    t.interrupt();
            } finally { // final state
                //设置任务为中断状态
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        //唤醒所有get阻塞的线程
        finishCompletion();
    }
    return true;
}

4.回顾

常见问题，FutureTask的get方法的执行过程:

get方法会将当前线程封装到WaitNode属性中，并采用头插法插入到waiters这个栈结构中。get方法会使用LockSupport阻塞线程，等待任务结束后调用finishCompletion唤醒调用get方法的对应线程。特别的是get使用超时方法获取返回值的时候，在时间结束后也会自动唤醒线程，不过超时的线程对应的WaitNode会调用removeWaiter方法被回收，重新构建waiters这个栈结构。当然，如果调用get方法的线程被外部线程使用中断方法中断的时候，也会调用removeWaiter方法回收当前WaitNode，重新构建waiters这个栈结构。

5.推荐

分享一个朋友的公众号，有很多干货，包含netty，spring，线程，spring cloud等详细讲解，也有详细的学习规划图，面试题整理等，我感觉在讲课这块比我讲的清楚:

微信关注我们

原文链接：https://my.oschina.net/u/3023261/blog/4667360

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

必知必会 | 关于Redis缓存这三大问题，必知必会

HI！我是小小，我们昨天没有见面，是因为昨天小小一天都在火车上呆，准备去广州的考试，所以脱稿一天，不过今天，小小的小blog已经开始恢复更新了。汇报一下进度小小今天坐车路过武汉，武汉武昌的一张路过照片。下面开始今天的正文，看见小小怎么辛苦的份上，滑到底下，给个素质三连？缓存雪崩缓存雪崩是指在某一个时间段内，缓存集中过期失效，如果这个时间段内有大量请求，而查询数据量巨大，所有的请求都会达到存储层，存储层的调用量会暴增，引起数据库压力过大甚至宕机。原因 Redis突然宕机大部分数据失效举个栗子比如我们基本上都经历过购物狂欢节，假设商家举办 23:00-24:00 商品打骨折促销活动。程序小哥哥在设计的时候，在 23:00 把商家打骨折的商品放到缓存中，并通过redis的expire设置了过期时间为1小时。这个时间段许多用户访问这些商品信息、购买等等。但是刚好到了24:00点的时候，恰好还有许多用户在访问这些商品，这时候对这些商品的访问都会落到数据库上，导致数据库要抗住巨大的压力，稍有不慎会导致，数据库直接宕机（over）。当商品没有失效的时候是这样的：当缓存GG（失效...

2020-10-10

571

这篇文章以两个典型的实际案例为基础，聊一聊分布式系统如何实现幂等性。案例一：转账系统在之前的文章，有多次提到转账系统这个案例，由于这个案例太典型了，很多大学教授数据库事务的时候就是用的这个案例。对于一个单体应用版的转账系统，我们可以直接利用数据库的事务来保证整个转账操作的ACID。但是，随着用户量级的增加，单个数据库的瓶颈也随之出现，于是就出现了分库分表的设计，即：一部分用户信息存储在一个数据库，另一部分存储在另一个数据库。基于这样的设计，单个数据库的事务肯定就不可用了，我们需要采用跨数据库的分布式事务，比如基于XA协议的分布式事务，但是这种方式有一些自身的问题，并且有应用场景的局限性。所以，一般来说实际场景都是采用基于BASE的最终一致性解决方案。如下则是一个简单的最终一致性方案设计： Step 1：Application收到用户发出的一个转账请求之后，首先执行转出方的逻辑，如下： begin transaction记账单（包括：转账请求uuid+转账状态in progress）扣钱（转出方余额减少）commit/rollback 这段逻辑包含在一个transaction里...

2020-10-09

581

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。