专栏首页BAT的乌托邦【小家java】一个例子让就能你彻底理解Java的Future模式,Future类的设计思想

【小家java】一个例子让就能你彻底理解Java的Future模式,Future类的设计思想


每篇一句

不是靠泪水博得同情,而是靠汗水赢得掌声

Future模式的设计思想,将在不久的将来大行其道。特别Reactive编程和Spring5的推出,此思想将越来越流行。而Java的Futrue模式属于代码层面的实现案例(也可以说是语法层面。Linux的epoll函数算是操作系统底层的实现)

Futrue模式简介

Future模式有点类似于网上购物,在你购买商品,订单生效之后,你可以去做自己的事情,等待商家通过快递给你送货上门。Future模式就是,当某一程序提交请求,期望得到一个答复。但是可能服务器程序对这个请求的处理比较慢,因此不可能马上收到答复。但是,在传统的单线程环境下,调用函数是同步的,它必须等到服务程序返回结果,才能继续进行其他处理。而Future模式下,调用方法是异步的,原本等待返回的时间段,在主调函数中,则可以处理其他的任务。传统的串行程序调用如下图所示:

Future模式的处理流程:

实现Future模式的客户端在拿到这个返回结果后,并不急于对它进行处理,而是去调用其它的业务逻辑,使call()方法有充分的时间去处理完成,这也是Future模式的精髓所在。在处理完其他业务逻辑后,最后再使用处理比较费时的Future数据。这个在处理过程中,就不存在无谓的等待,充分利用了时间,从而提升了系统的响应和性能。

JDK内置实现介绍

在JDK的内置并发包中,就已经内置了一种Future的实现,提供了更加丰富的线程控制,其基本用意和核心理念与上面实现代码一致。

在JDK中的Future模式中,最重要的是FutureTask类它实现了Runnable接口,可以作为单独的线程运行。在其run()方法中,通过Sync内部类,调用Callable接口,并维护Callable接口的返回对象。当使用FutureTask.get()时,将返回Callable接口的返回对象。FutureTask还可以对任务本身进行其他控制操作。

FutureTask该类在JDK5、6版本的实现和JDK8中的实现有较大的差异.JDK8后退该类的性能进行了较大的提升,后面会结合源码级别进行讲解。

例子剖析:Futrue模式

下面这个例子总体来说是个非常通俗易懂的例子,并且讲解得也比较详细,愿读者能够读懂,有不懂得可以随时留了言哦

先上一个场景:假如你突然想做饭,但是没有厨具,也没有食材。网上购买厨具比较方便,食材去超市买更放心。

实现分析:在快递员送厨具的期间,我们肯定不会闲着,可以去超市买食材。所以,在主线程里面另起一个子线程去网购厨具。

但是,子线程执行的结果是要返回厨具的,而run方法是没有返回值的。所以,这才是难点,需要好好考虑一下。

  • 代码模拟:
package test;

public class CommonCook {

    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        // 第一步 网购厨具
        OnlineShopping thread = new OnlineShopping();
        thread.start();

		//通过join线程来阻断主线程,以先保证厨具送到才继续往下走
        thread.join();  


        // 第二步 去超市购买食材
        Thread.sleep(2000);  // 模拟购买食材时间
        Shicai shicai = new Shicai();
        System.out.println("第二步:食材到位");


        // 第三步 用厨具烹饪食材
        System.out.println("第三步:开始展现厨艺");
        cook(thread.chuju, shicai);
        
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    // 网购厨具线程的线程类 
    static class OnlineShopping extends Thread {
        
        private Chuju chuju;

        @Override
        public void run() {
            System.out.println("第一步:下单");
            System.out.println("第一步:等待送货");
            try {
                Thread.sleep(5000);  // 模拟快递小哥送货时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("第一步:快递送到");
            chuju = new Chuju(); //拿到厨具了
        }
        
    }

    //  用厨具烹饪食材的线程  做饭肯定得厨具、食材都有了才能开始  所以传进去
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 厨具类
    static class Chuju {}
    
    // 食材类
    static class Shicai {}
}

运行结果:

第一步:下单
第一步:等待送货
第一步:快递送到
第二步:食材到位
第三步:开始展现厨艺
总共用时7013ms

从总用时我们可以看到,多线程已经失去了意义。在厨具送到期间,我们不能干任何事。对应代码,就是调用join方法阻塞主线程。

其实这里面虽然用到了Thread,但其实并没有起到多线程的作用。因为很显然,一个join,使得一切都串行化了。那有人会问,不阻塞行不行呢?答案显然是不行的,因为还没有厨具的情况下,你不能做饭。

从代码来看的话,run方法不执行完,属性chuju就没有被赋值,还是null。换句话说,没有厨具,怎么做饭。 Java现在的多线程机制,核心方法run是没有返回值的;如果要保存run方法里面的计算结果,必须等待run方法计算完,无论计算过程多么耗时。 面对这种尴尬的处境,程序员就会想:在子线程run方法计算的期间,能不能在主线程里面继续异步执行???

Where there is a will,there is a way!!!

这就是我们今天的主菜:这种想法的核心就是Future模式,下面先应用一下Java自己实现的Future模式。

模拟代码2:用Future模式改进

package test;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureCook {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();
        
        // 第一步 网购厨具  采用Callable后面配合FutrueTask使用
        Callable<Chuju> onlineShopping = new Callable<Chuju>() {

            @Override
            public Chuju call() throws Exception {
                System.out.println("第一步:下单");
                System.out.println("第一步:等待送货");
                Thread.sleep(5000);  // 模拟送货时间
                System.out.println("第一步:快递送到");
                return new Chuju();
            }
            
        };
        FutureTask<Chuju> task = new FutureTask<Chuju>(onlineShopping);
        new Thread(task).start(); //这样启动,完全无阻塞,并行处理


        // 第二步 去超市购买食材
        Thread.sleep(2000);  // 模拟购买食材时间
        Shicai shicai = new Shicai();
        System.out.println("第二步:食材到位");


        // 第三步 用厨具烹饪食材
        if (!task.isDone()) {  // 联系快递员,询问是否到货
            System.out.println("第三步:厨具还没到,心情好就等着(心情不好就调用cancel方法取消订单)");
            //如果需要取消 这里调用task的cancel方法即可
        }
        Chuju chuju = task.get();
        System.out.println("第三步:厨具到位,开始展现厨艺");

		//完事具备  可以开始做饭了
        cook(chuju, shicai);
        
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    //  用厨具烹饪食材
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 厨具类
    static class Chuju {}
    
    // 食材类
    static class Shicai {}

}

从上面的注释可以明显的看得出来,是有2个子线程在并行处理的。所以看输出结果:

第一步:下单
第一步:等待送货
第二步:食材到位
第三步:厨具还没到,心情好就等着(心情不好就调用cancel方法取消订单)
第一步:快递送到
第三步:厨具到位,开始展现厨艺
总共用时5005ms

可以看见,在快递员送厨具的期间,我们没有闲着,可以去买食材;而且我们知道厨具到没到,甚至可以在厨具没到的时候,取消订单不要了。

源码分析:Futrue模式

Callable接口可以看作是Runnable接口的补充,call方法带有返回值,并且可以抛出异常。

1)把耗时的网购厨具逻辑,封装到了一个Callable的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;
}

2)把Callable实例当作参数,生成一个FutureTask的对象,然后把这个对象当作一个Runnable,作为参数另起线程。所以继续看看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();
}

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

这个继承体系中的核心接口是Future。Future的核心思想是:一个方法f,计算过程可能非常耗时,等待f返回,显然不明智。可以在调用f的时候,立马返回一个Future,可以通过Future这个数据结构去控制方法f的计算过程。 这里的控制包括:

  • get方法:获取计算结果(如果还没计算完,也是必须等待的,阻塞)
  • cancel方法:还没计算完,可以取消计算过程
  • isDone方法:判断是否计算完
  • isCancelled方法:判断计算是否被取消
  • 这些接口的设计很完美,FutureTask的实现注定不会简单,后面再说。

3)在第三步里面,调用了isDone方法查看状态,然后直接调用task.get方法获取厨具,不过这时还没送到,所以还是会等待3秒。对比第一段代码的执行结果,这里我们节省了2秒。这是因为在快递员送货期间,我们去超市购买食材,这两件事在同一时间段内异步执行。

通过以上3步,我们就完成了对Java原生Future模式最基本的应用。下面具体分析下FutureTask的实现,先看JDK8的,再比较一下JDK6的实现。

既然FutureTask也是一个Runnable,那就看看它的run方法:

public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable; // 这里的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); // 保存call方法抛出的异常
                }
                if (ran)
                    set(result); // 保存call方法的执行结果
            }
        } 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);
        }
    }

先看try语句块里面的逻辑,发现run方法的主要逻辑就是运行Callable的call方法,然后将保存结果或者异常(用的一个属性result)。这里比较难想到的是,将call方法抛出的异常也保存起来了。

这里表示状态的属性state是个什么鬼:

* Possible state transitions:
     * 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;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED  = 6;

把FutureTask看作一个Future,那么它的作用就是控制Callable的call方法的执行过程,在执行的过程中自然会有状态的转换:

1)一个FutureTask新建出来,state就是NEW状态;COMPETING和INTERRUPTING用的进行时,表示瞬时状态,存在时间极短(为什么要设立这种状态???不解);NORMAL代表顺利完成;EXCEPTIONAL代表执行过程出现异常;CANCELED代表执行过程被取消;INTERRUPTED被中断 2)执行过程顺利完成:NEW -> COMPLETING -> NORMAL 3)执行过程出现异常:NEW -> COMPLETING -> EXCEPTIONAL 4)执行过程被取消:NEW -> CANCELLED 5)执行过程中,线程中断:NEW -> INTERRUPTING -> INTERRUPTED 代码中状态判断、CAS操作等细节,请读者自己阅读。

再看看get方法的实现:

 public V get() throws InterruptedException, ExecutionException {
        int s = state;
        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;
                }
                LockSupport.parkNanos(this, nanos);
            }
            else
                LockSupport.park(this);
        }
    }

get方法的逻辑很简单,如果call方法的执行过程已完成,就把结果给出去;如果未完成,就将当前线程挂起等待。awaitDone方法里面死循环的逻辑,推演几遍就能弄懂;它里面挂起线程的主要创新是定义了WaitNode类,来将多个等待线程组织成队列,这是与JDK6的实现最大的不同。

挂起的线程何时被唤醒:

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
    }

以上就是JDK8的大体实现逻辑,像cancel、set等方法,也请读者自己阅读。

再来看看JDK6的实现。

JDK6的FutureTask的基本操作都是通过自己的内部类Sync来实现的,而Sync继承自AbstractQueuedSynchronizer这个出镜率极高的并发工具类

/** State value representing that task is running */
        private static final int RUNNING   = 1;
        /** State value representing that task ran */
        private static final int RAN       = 2;
        /** State value representing that task was cancelled */
        private static final int CANCELLED = 4;

        /** The underlying callable */
        private final Callable<V> callable;
        /** The result to return from get() */
        private V result;
        /** The exception to throw from get() */
        private Throwable exception;

里面的状态只有基本的几个,而且计算结果和异常是分开保存的。

V innerGet() throws InterruptedException, ExecutionException {
            acquireSharedInterruptibly(0);
            if (getState() == CANCELLED)
                throw new CancellationException();
            if (exception != null)
                throw new ExecutionException(exception);
            return result;
        }

这个get方法里面处理等待线程队列的方式是调用了acquireSharedInterruptibly方法,其中的等待线程队列、线程挂起和唤醒等逻辑,这里不再赘述。

最后

Futrue模式的变成思想非常的好,并且随着分布式、微服务、云计算等领域的兴起,该思想会越来越流行的,所以希望读者能够在日常开发中使用起来,提高程序的运行效率,从而提高并发。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【小家Java】一次Java线程池误用(newFixedThreadPool)引发的线上血案和总结

    自从最近的某年某月某天起,线上服务开始变得不那么稳定(软病)。在高峰期,时常有几台机器的内存持续飙升,并且无法回收,导致服务不可用。

    BAT的乌托邦
  • 【小家java】聊聊Java中的Runtime类

    Runtime类封装了运行时的环境。每个 Java 应用程序都有一个 Runtime 类实例,使应用程序能够与其运行的环境相连接。

    BAT的乌托邦
  • 【小家java】Java里的进程、线程、协程 、Thread、守护线程、join线程的总结

    说到线程,很多人最直观的感受就是多线程。本章不讨论高并发、多线程之类的。返璞归真,咱们来讨论讨论线程这个东西到底是什么东西,着眼于线程本身,我们怎么玩? 为了...

    BAT的乌托邦
  • ThreadLocal VS FastThreadlocal

    ThreadLocal:线程局部,但更多听到是线程局部变量,所谓局部即:单个线程内变量可共享,在并发线程中见到他的概率更多,但解决的并非线程安全问题。

    疯狂的KK
  • VulnHub通关日记-DC_5-Walkthrough

    DC-5 is another purposely built vulnerable lab with the intent of gaining experi...

    Gcow安全团队
  • 线程池的创建和使用

    这里需要注意,线程的数量是固定的,但是队列大小是无界的(Integer.MAX_VALUE足够大,大到可以任务无界。)

    IT云清
  • 一步步学习MQX实时操作系统(2)

    上次简单介绍了MQX的一些背景和知识,今天我们来学习以IAR为开发环境,以kv4x系列控制器为平台,学习分析例程,搭建SDK+MQX开发环境。 准备工...

    用户1605515
  • .NET框架设计(常被忽视的C#设计技巧)

    阅读目录: 1.开篇介绍 2.尽量使用Lambda匿名函数调用代替反射调用(走进声明式设计) 3.被忽视的特性(Attribute)设计方式 4.扩展方法让你的...

    王清培
  • Spring Boot集成JPA的Column注解命名字段无效的问题

    偶然发现,Spring Boot集成jpa编写实体类的时候,默认使用的命名策略是下划线分隔的字段命名。

    pollyduan
  • Ajax(20171102)

    天天_哥

扫码关注云+社区

领取腾讯云代金券

玩转腾讯云 有奖征文活动