5. 高并发基础之Future异步回调

随着业务模块系统越来越多,各个系统的业务架构变得越来越错综复杂,特别是这几年微服务架构的兴起,跨设备跨服务的接口调用越来越频繁。打个简单的比方:现在的一个业务流程,可能需要调用N次第三方接口,获取N种上游数据。因此,面临一个大的问题是:如何高效率地异步去调取这些接口,然后同步去处理这些接口的返回结果呢?这里涉及线程的异步回调问题,这也是高并发的一个基础问题。

在Netty源代码中,大量地使用了异步回调技术,并且基于Java的异步回调,设计了自己的一整套异步回调接口和实现。

在本章中,我们从Java Future异步回调技术入手,然后介绍比较常用的第三方异步回调技术——谷歌公司的Guava Future相关技术,最后介绍一下Netty的异步回调技术。

总之,学习高并发编程,掌握异步回调技术是编程人员必须具备的一项基础技术。

5.1 从泡茶的案例说起

在进入异步回调模式的正式解读之前,先来看一个比较好理解的异步生活示例。笔者尼恩就想到自己中学8年级的语文中,有一篇华罗庚的课文——《统筹方法》,其中举了一个合理安排工序、以便提升效率的泡茶案例。下面分别是用阻塞模式和异步回调模式来实现其中的异步泡茶流程。强调一下:这里直接略过顺序执行的冒泡工序,那个效率太低了。

为了异步执行整个泡茶流程,分别设计三条线程:主线程、清洗线程、烧水线程。

  1. 主线程(MainThread)的工作是:启动清洗线程、启动烧水线程,等清洗、烧水的工作完成后,泡茶喝。
  2. 清洗线程(WashThread)的工作是:洗茶壶、洗茶杯。
  3. 烧水线程(HotWarterThread)的工作是:洗好水壶,灌上凉水,放在火上,一直等水烧开。

下面分别使用阻塞模式、异步回调模式来实现泡茶喝的案例。

5.2 join异步阻塞

阻塞模式实现泡茶实例,首先从最为基础的多线程join合并开始。join操作的原理是:阻塞当前的线程,直到准备合并的目标线程的执行完成。

5.2.1 线程的join合并流程

在Java中,线程(Thread)的合并流程是:假设线程A调用了线程B的B.join方法,合并B线程。那么,线程A进入阻塞状态,直到B线程执行完成。

在泡茶喝的例子中,主线程通过分别调用烧水线程和清洗线程的join方法,等待烧水线程和清洗线程执行完成,然后执行自己的泡茶操作。具体如图5-1所示。

图5-1 使用join实现泡茶实例的流程

5.2.2 使用join实现异步泡茶喝的实践案例

使用join实现泡茶喝,这是一个异步阻塞版本,具体的代码实现如下:

//...
public class JoinDemo {
    public static final int SLEEP_GAP = 500;
    public static String getCurThreadName() {
        return Thread.currentThread().getName();
    }
    static class HotWarterThread extends Thread {
        public HotWarterThread() {
            super("** 烧水-Thread");
        }
        public void run() {
            try {
                Logger.info("洗好水壶");
                Logger.info("灌上凉水");
                Logger.info("放在火上");
                //线程睡眠一段时间,代表烧水中
                Thread.sleep(SLEEP_GAP);
                Logger.info("水开了");
            } catch (InterruptedException e) {
                Logger.info(" 发生异常被中断.");
            }
            Logger.info(" 运行结束.");
        }
    }
    static class WashThread extends Thread {
        public WashThread() {
            super("$$ 清洗-Thread");
        }
        public void run() {
            try {
                Logger.info("洗茶壶");
                Logger.info("洗茶杯");
                Logger.info("拿茶叶");
                //线程睡眠一段时间,代表清洗中
                Thread.sleep(SLEEP_GAP);
                Logger.info("洗完了");
            } catch (InterruptedException e) {
                Logger.info(" 发生异常被中断.");
            }
            Logger.info(" 运行结束.");
        }
    }
    public static void main(String args[]) {
        Thread hThread = new HotWarterThread();
        Thread wThread = new WashThread();
        hThread.start();
        wThread.start();
        try {
            // 合并烧水-线程
            hThread.join();
            // 合并清洗-线程
            wThread.join();
            Thread.currentThread().setName("主线程");
            Logger.info("泡茶喝");
        } catch (InterruptedException e) {
            Logger.info(getCurThreadName() + "发生异常被中断.");
        }
        Logger.info(getCurThreadName() + " 运行结束.");
    }
}

程序中有三个线程:(1)主线程main;(2)烧水线程hThread;(3)清洗线程wThread。main主线程调用了hThread.join()实例方法,合并烧水线程,也调用了wThread.join()实例方法,合并清洗线程。

说明一下:hThread、wThread是线程实例,在示例代码中,hThread对应的线程名称为"** 烧水-Thread",wThread对应的线程名称为“$$ 清洗-Thread”。

执行结果如下:

[** 烧水-Thread|JoinDemo$HotWarterThread.run] |>  洗好水壶 
[** 烧水-Thread|JoinDemo$HotWarterThread.run] |>  灌上凉水 
[** 烧水-Thread|JoinDemo$HotWarterThread.run] |>  放在火上 
[$$ 清洗-Thread|JoinDemo$WashThread.run] |>  洗茶壶 
[$$ 清洗-Thread|JoinDemo$WashThread.run] |>  洗茶杯 
[$$ 清洗-Thread|JoinDemo$WashThread.run] |>  拿茶叶 
[** 烧水-Thread|JoinDemo$HotWarterThread.run] |>  水开了 
[** 烧水-Thread|JoinDemo$HotWarterThread.run] |>  运行结束. 
[$$ 清洗-Thread|JoinDemo$WashThread.run] |>  洗完了 
[$$ 清洗-Thread|JoinDemo$WashThread.run] |>  运行结束. 
 [主线程|JoinDemo.main] |>  泡茶喝 
 [主线程|JoinDemo.main] |>  主线程 运行结束. 

5.2.3 详解join合并方法

join方法的应用场景:A线程调用B线程的join方法,等待B线程执行完成;在B线程没有完成前,A线程阻塞。

join方法有三个重载版本:

  1. void join():A线程等待B线程执行结束后,A线程重新恢复执行。
  2. void join(long millis):A线程等待B线程执行一段时间,最长等待时间为millis毫秒。超过millis毫秒后,不论B线程是否结束,A线程重新恢复执行。
  3. void join(long millis,int nanos):等待B线程执行一段时间,最长等待时间为millis毫秒,加nanos纳秒。超过时间后,不论B线程是否结束,A线程重新恢复执行。

强调一下容易混淆的几点:

  1. join是实例方法,不是静态方法,需要使用线程对象去调用,如thread.join()。
  2. join调用时,不是线程所指向的目标线程阻塞,而是当前线程阻塞。
  3. 只有等到当前线程所指向的线程执行完成,或者超时,当前线程才能重新恢复执行。

join有一个问题:被合并的线程没有返回值。例如,在烧水的实例中,如果烧水线程的执行结束,main线程是无法知道结果的。同样,清洗线程的执行结果,main线程也是无法知道的。形象地说,join线程合并就像一个一个闷葫芦。只能发起合并线程,不能取到执行结果。

如果需要获得异步线程的执行结果,怎么办呢?可以使用Java的FutureTask系列类。

5.3 FutureTask异步回调之重武器

为了获取异步线程的返回结果,Java在1.5版本之后提供了一种新的多线程的创建方式——FutureTask方式。FutureTask方式包含了一系列的Java相关的类,在java.util.concurrent包中。其中最为重要的是FutureTask类和Callable接口。

5.3.1 Callable接口

在介绍Callable接口之前,先得重提一下旧的Runnable接口。Runnable接口是在Java多线程中表示线程的业务代码的抽象接口。但是,Runnable有一个重要的问题,它的run方法是没有返回值的。正因为如此,Runnable不能用于需要有返回值的应用场景。

为了解决Runnable接口的问题,Java定义了一个新的和Runnable类似的接口——Callable接口。并且将其中的代表业务处理的方法命名为call,call方法有返回值。

Callable的代码如下:

package java.util.concurrent;
@FunctionalInterface
public interface Callable<V> {
    //call方法有返回值
    V call() throws Exception;
}

Callable接口是一个泛型接口,也声明为了“函数式接口”。其唯一的抽象方法call有返回值,返回值的类型为泛型形参的实际类型。call抽象方法还有一个Exception的异常声明,容许方法内部的异常不经过捕获。

总之,Callable接口可以对应到Runnable接口;Callable接口的call方法可以对应到Runnable接口的run方法。相比较而言,Callable接口的功能更强大一些。

Callable接口与Runnable接口相比,还有一个很大的不同:Callable接口的实例不能作为Thread线程实例的target来使用;而Runnable接口实例可以作为Thread线程实例的target构造参数,开启一个Thread线程。

问题来了,Java中的线程类型,只有一个Thread类,没有其他的类型。如果Callable实例需要异步执行,就要想办法赋值给Thread的target成员,一个Runnable类型的成员。为此,Java提供了在Callable实例和Thread的target成员之间一个搭桥的类——FutureTask类。

5.3.2 初探FutureTask类

顾名思义,FutureTask类代表一个未来执行的任务,表示新线程所执行的操作。FutureTask类也位于java.util.concurrent包中。FutureTask类的构造函数的参数为Callable类型,实际上是对Callable类型的二次封装,可以执行Callable的call方法。FutureTask类间接地继承了Runnable接口,从而可以作为Thread实例的target执行目标。

FutureTask类的构造函数的源代码,如下:

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

FutureTask类就像一座搭在Callable实例与Thread线程实例之间的桥。FutureTask类的内部封装一个Callable实例,然后自身又作为Thread线程的target。

在外部,如何要获取Callable实例的异步执行结果,不是调用其call方法,而是需要通过FutureTask类的相应方法去获取。

总体来说,FutureTask类首先是一个搭桥类的角色,FutureTask类能当作Thread线程去执行目标target,被异步执行;其次,如果要获取异步执行的结果,需要通过FutureTask类的方法去获取,在FutureTask类的内部,会将Callable的call方法的真正结果保存起来,以供外部获取。

在Java语言中,将FutureTask类的一系列操作,抽象出来作为一个重要的接口——Future接口。当然,FutureTask类也实现了此接口。

5.3.3 Future接口

Future接口不复杂,主要是对并发任务的执行及获取其结果的一些操作。主要提供了3大功能:

  1. 判断并发任务是否执行完成。
  2. 获取并发的任务完成后的结果。
  3. 取消并发执行中的任务。

Future接口的源代码如下:

package java.util.concurrent;
public interface Future<V> {
    boolean  cancel(booleanmayInterruptRunning);
    booleanisCancelled();
    booleanisDone();
    V get() throws InterruptedException,ExecutionException;
    V get(long timeout,TimeUnitunit) throws InterruptedException,
                  ExecutionException,TimeoutException;
}

关于Future接口的方法,详细说明如下:

  • V get():获取并发任务执行的结果。注意,这个方法是阻塞性的。如果并发任务没有执行完成,调用此方法的线程会一直阻塞,直到并发任务执行完成。
  • V get(Long timeout,TimeUnit unit):获取并发任务执行的结果。也是阻塞性的,但是会有阻塞的时间限制,如果阻塞时间超过设定的timeout时间,该方法将抛出异常。
  • booleanisDone():获取并发任务的执行状态。如果任务执行结束,则返回true。
  • booleanisCancelled():获取并发任务的取消状态。如果任务完成前被取消,则返回true。
  • boolean cancel(booleanmayInterruptRunning):取消并发任务的执行。

5.3.4 再探FutureTask类

FutureTask类实现了Future接口,提供了外部操作异步任务的能力。现在,回到FutureTask类自身。为了完成异步执行Callable类型的任务、获取任务结果的使命,在FutureTask类的内部,又有哪些成员和方法呢?

首先,FutureTask内部有一个Callable类型的成员,代表异步执行的逻辑:

private Callable<V> callable;

callable实例属性必须要在FutureTask类的实例构造时进行初始化。

其次,FutureTask内部有一个run方法。这个run方法是Runnable接口的抽象方法,在FutureTask类的内部提供了自己的实现。在Thread线程实例执行时,会将这个run方法作为target目标去异步执行。在FutureTask内部的run实现代码中,会执行其callable成员的call方法。执行完成后,将结果保存起来。保存在哪里呢?

再次,FutureTask内部有另一个重要的成员——outcome属性,用于保存结果:

private Object outcome;

outcome属性所保存的结果,可供FutureTask类的结果获取方法(如get)来获取。

至此,这个很重要的FutureTask搭桥类就介绍完了。如果还不是很清楚,也不要紧,相信通过实践一个FutureTask版本的喝茶示例,就明白了。

5.3.5 使用FutureTask类实现异步泡茶喝的实践案例

在前面的join版本喝茶示例中,有一个很大的问题:就是主线程获取不到异步线程的返回值。打个比方,如果烧水线程出了问题,或者清洗线程出了问题,主线程是无办法知道的。哪怕不具备泡茶条件,主线程也只能继续泡茶喝。

使用FutureTask类和Callable接口,进行异步结果的获取,代码如下:

//...
public class JavaFutureDemo {
    public static final int SLEEP_GAP = 500;
    public static String getCurThreadName() {
        return Thread.currentThread().getName();
    } 
    static class HotWarterJob implements Callable<Boolean> //①
    {
        @Override
        public Boolean call() throws Exception //②
        {
            try {
                Logger.info("洗好水壶");
                Logger.info("灌上凉水");
                Logger.info("放在火上");
                //线程睡眠一段时间,代表烧水中
                Thread.sleep(SLEEP_GAP);
                Logger.info("水开了");
            } catch (InterruptedException e) {
                Logger.info(" 发生异常被中断.");
                return false;
            }
            Logger.info(" 运行结束.");
            return true;
        }
    }
    static class WashJob implements Callable<Boolean> {
        @Override
        public Boolean call() throws Exception {
            try {
                Logger.info("洗茶壶");
                Logger.info("洗茶杯");
                Logger.info("拿茶叶");
                //线程睡眠一段时间,代表清洗中
                Thread.sleep(SLEEP_GAP);
                Logger.info("洗完了");
            } catch (InterruptedException e) {
                Logger.info(" 清洗工作发生异常被中断.");
                return false;
            }
            Logger.info(" 清洗工作运行结束.");
            return true;
        }
    }
    public static void drinkTea(booleanwarterOk, booleancupOk) {
        if (warterOk&amp;&amp;cupOk) {
            Logger.info("泡茶喝");
        } else if (!warterOk) {
            Logger.info("烧水失败,没有茶喝了");
        } else if (!cupOk) {
            Logger.info("杯子洗不了,没有茶喝了");
        }
    }
    public static void main(String args[]) {
        Callable<Boolean> hJob = new HotWarterJob();//③
        FutureTask<Boolean> hTask = new FutureTask<>(hJob);//④
        Thread hThread = new Thread(hTask, "** 烧水-Thread");//⑤
        Callable<Boolean> wJob = new WashJob();//③
        FutureTask<Boolean> wTask = new FutureTask<>(wJob);//④
        Thread wThread = new Thread(wTask, "$$ 清洗-Thread");//⑤
        hThread.start();
        wThread.start();
        Thread.currentThread().setName("主线程");
            try {
                booleanwarterOk = hTask.get();
                booleancupOk = wTask.get();
                drinkTea(warterOk, cupOk);
            } catch (InterruptedException e) {
                Logger.info(getCurThreadName() + "发生异常被中断.");
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        Logger.info(getCurThreadName() + " 运行结束.");
    }
}

首先,在上面的喝茶实例代码中使用了Callable接口,替代了Runnable接口,并且在call方法中返回了异步线程的执行结果。

static class WashJob implements Callable<Boolean>
{
        @Override
        public Boolean call() throws Exception
        {
            //..业务代码,并且有清洗的结果
        }
}

其次,从Callable异步逻辑到异步线程,需要创建一个FutureTask类的实例,并通过FutureTask类的实例,创建新的线程:

Callable<Boolean> hJob = new HotWarterJob();//异步逻辑
FutureTask<Boolean> hTask = new FutureTask<Boolean> (hJob);//搭桥实例
Thread hThread = new Thread(hTask, "** 烧水-Thread");//异步线程

FutureTask和Callable都是泛型类,泛型参数表示返回结果的类型。所以,在使用的时候,它们两个实例的泛型参数一定需要保持一致的。

最后,通过FutureTask类的实例,取得异步线程的执行结果。一般来说,通过FutureTask实例的get方法,可以获取线程的执行结果。

总之,FutureTask类的实现比join线程合并操作更加高明,能取得异步线程的结果。但是,也就未必高明到哪里去了。为啥呢?

因为通过FutureTask类的get方法,获取异步结果时,主线程也会被阻塞的。这一点,FutureTask和join也是一样的,它们俩都是异步阻塞模式。

异步阻塞的效率往往是比较低的,被阻塞的主线程不能干任何事情,唯一能干的,就是在傻傻地等待。原生Java API,除了阻塞模式的获取结果外,并没有实现非阻塞的异步结果获取方法。如果需要用到获取异步的结果,则需要引入一些额外的框架,这里首先介绍谷歌公司的Guava框架。

5.4 Guava的异步回调

何为Guava?它是谷歌公司提供的Java扩展包,提供了一种异步回调的解决方案。相关的源代码在com.google.common.util.concurrent包中。包中的很多类,都是对java.util.concurrent能力的扩展和增强。例如,Guava的异步任务接口ListenableFuture,扩展了Java的Future接口,实现了非阻塞获取异步结果的功能。

总体来说,Guava的主要手段是增强而不是另起炉灶。为了实现非阻塞获取异步线程的结果,Guava对Java的异步回调机制,做了以下的增强:

  1. 引入了一个新的接口ListenableFuture,继承了Java的Future接口,使得Java的Future异步任务,在Guava中能被监控和获得非阻塞异步执行的结果。
  2. 引入了一个新的接口FutureCallback,这是一个独立的新接口。该接口的目的,是在异步任务执行完成后,根据异步结果,完成不同的回调处理,并且可以处理异步结果。

5.4.1 详解FutureCallback

FutureCallback是一个新增的接口,用来填写异步任务执行完后的监听逻辑。FutureCallback拥有两个回调方法:

  1. onSuccess方法,在异步任务执行成功后被回调;调用时,异步任务的执行结果,作为onSuccess方法的参数被传入。
  2. onFailure方法,在异步任务执行过程中,抛出异常时被回调;调用时,异步任务所抛出的异常,作为onFailure方法的参数被传入。

FutureCallback的源代码如下:

package com.google.common.util.concurrent;
public interface FutureCallback<V> {
    void onSuccess(@Nullable V var1);
    void onFailure(Throwable var1);
}

注意,Guava的FutureCallback与Java的Callable,名字相近,但实质不同,存在本质的区别:

  1. Java的Callable接口,代表的是异步执行的逻辑。
  2. Guava的FutureCallback接口,代表的是Callable异步逻辑执行完成之后,根据成功或者异常两种情况,所需要执行的善后工作。

Guava是对Java Future异步回调的增强,使用Guava异步回调,也需要用到Java的Callable接口。简单地说,只有在Java的Callable任务执行的结果出来之后,才可能执行Guava中的FutureCallback结果回调。

Guava如何实现异步任务? Callable和FutureCallback结果回调之间的监控关系呢?Guava引入了一个新接口ListenableFuture,它继承了Java的Future接口,增强了监控的能力。

5.4.2 详解ListenableFuture

看ListenableFuture接口的名称,就知道它与Java中Future接口的亲戚关系。没错,Guava的ListenableFuture接口是对Java的Future接口的扩展,可以理解为异步任务的实例。源代码如下:

package com.google.common.util.concurrent;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
public interface ListenableFuture<V> extends Future<V> {
    //此方法由Guava内部调用
    void addListener(Runnable r, Executor e);
}

ListenableFuture仅仅增加了一个方法—addListener方法。它的作用就是将前一小节的FutureCallback善后回调工作,封装成一个内部的Runnable异步回调任务,在Callable异步任务完成后,回调FutureCallback进行善后处理。

注意,这个addListener方法只在Guava内部调用,如果对它感兴趣,可以查看Guava源代码。在实际编程中,我们不会调用addListener。

在实际编程中,如何将FutureCallback回调逻辑绑定到异步的ListenableFuture任务呢?可以使用Guava的Futures工具类,它有一个addCallback静态方法,可以将FutureCallback的回调实例绑定到ListenableFuture异步任务。下面是一个简单的绑定实例:

Futures.addCallback(listenableFuture,  newFutureCallback<Boolean>()
{
    public void onSuccess(Boolean r)
    {
        // listenableFuture内部的Callable 成功时回调此方法
    }
    public void onFailure(Throwable t)
    {
       // listenableFuture内部的Callable异常时回调此方法
    }
});

现在的问题来了,既然Guava的ListenableFuture接口是对Java的Future接口的扩展,都表示异步任务。那么Guava的异步任务实例,从何而来呢?

5.4.3 ListenableFuture异步任务

如果要获取Guava的ListenableFuture异步任务实例,主要是通过向线程池(ThreadPool)提交Callable任务的方式来获取。不过,这里所说的线程池,不是Java的线程池,而是Guava自己定制的Guava线程池。

Guava线程池,是对Java线程池的一种装饰。创建Guava线程池的方法如下:

//java 线程池
ExecutorService jPool=  Executors.newFixedThreadPool(10);
//Guava线程池
ListeningExecutorService gPool=  MoreExecutors.listeningDecorator(jPool);

首先创建Java线程池,然后以它作为Guava线程池的参数,再构造一个Guava线程池。有了Guava的线程池之后,就可以通过submit方法来提交任务了;任务提交之后的返回结果,就是我们所要的ListenableFuture异步任务实例了。

简单地说,获取异步任务实例的方式,是通过向线程池提交Callable业务逻辑来实现。代码如下:

//调用submit方法来提交任务,返回异步任务实例
ListenableFuture<Boolean> hFuture = gPool.submit(hJob);
//绑定回调实例
Futures.addCallback(listenableFuture,  newFutureCallback<Boolean>()
{
     //实现回调方法,有两个
});

获取了ListenableFuture实例之后,通过Futures.addCallback方法,将FutureCallback回调逻辑的实例绑定到ListenableFuture异步任务实例,实现异步执行完成后的回调。

总结一下,Guava异步回调的流程如下:
第1步:实现Java的Callable接口,创建异步执行逻辑。还有一种情况,如果不需要返回值,异步执行逻辑也可以实现Java的Runnable接口。
第2步:创建Guava线程池。
第3步:将第1步创建的Callable/Runnable异步执行逻辑的实例,通过submit提交到Guava线程池,从而获取ListenableFuture异步任务实例。
第4步:创建FutureCallback回调实例,通过Futures.addCallback将回调实例绑定到ListenableFuture异步任务上。

完成以上四步,当Callable/Runnable异步执行逻辑完成后,就会回调异步回调实例FutureCallback的回调方法onSuccess/onFailure。

5.4.4 使用Guava实现泡茶喝的实践案例

前面已经完成了join版本、FutureTask版本的泡茶喝实践案例。大家对此实例的业务功能,应该已经非常熟悉了,这里不再赘述。下面是Guava的异步回调的演进版本,代码如下:

import cc.gongchang.util.Logger;
import com.google.common.util.concurrent.*;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class GuavaFutureDemo {

    public static final int SLEEP_GAP = 500;


    public static String getCurThreadName() {
        return Thread.currentThread().getName();
    }

    static class HotWarterJob implements Callable<Boolean> //①
    {

        public Boolean call() throws Exception //②
        {

            try {
                Logger.info("洗好水壶");
                Logger.info("灌上凉水");
                Logger.info("放在火上");

                //线程睡眠一段时间,代表烧水中
                Thread.sleep(SLEEP_GAP);
                Logger.info("水开了");

            } catch (InterruptedException e) {
                Logger.info(" 发生异常被中断.");
                return false;
            }
            Logger.info(" 烧水工作,运行结束.");

            return true;
        }
    }

    static class WashJob implements Callable<Boolean> {

        public Boolean call() throws Exception {

            try {
                Logger.info("洗茶壶");
                Logger.info("洗茶杯");
                Logger.info("拿茶叶");
                //线程睡眠一段时间,代表清洗中
                Thread.sleep(SLEEP_GAP);
                Logger.info("洗完了");

            } catch (InterruptedException e) {
                Logger.info(" 清洗工作 发生异常被中断.");
                return false;
            }
            Logger.info(" 清洗工作  运行结束.");
            return true;
        }

    }

    //泡茶线程
    static class MainJob implements Runnable {

        boolean warterOk = false;
        boolean cupOk = false;
        int gap = SLEEP_GAP / 10;

        public void run() {
            while (true) {
                try {
                    Thread.sleep(gap);
                    Logger.info("读书中......");
                } catch (InterruptedException e) {
                    Logger.info(getCurThreadName() + "发生异常被中断.");
                }

                if (warterOk && cupOk) {
                    drinkTea(warterOk, cupOk);
                }
            }
        }


        public void drinkTea(Boolean wOk, Boolean cOK) {
            if (wOk && cOK) {
                Logger.info("泡茶喝,茶喝完");
                this.warterOk = false;
                this.gap = SLEEP_GAP * 100;
            } else if (!wOk) {
                Logger.info("烧水失败,没有茶喝了");
            } else if (!cOK) {
                Logger.info("杯子洗不了,没有茶喝了");
            }

        }
    }

    public static void main(String args[]) {

        //新起一个线程,作为泡茶主线程
        final MainJob mainJob = new MainJob();
        Thread mainThread = new Thread(mainJob);
        mainThread.setName("主线程");
        mainThread.start();

        //烧水的业务逻辑
        Callable<Boolean> hotJob = new HotWarterJob();
        //清洗的业务逻辑
        Callable<Boolean> washJob = new WashJob();

        //创建java 线程池
        ExecutorService jPool =
                Executors.newFixedThreadPool(10);

        //包装java线程池,构造guava 线程池
        ListeningExecutorService gPool =
                MoreExecutors.listeningDecorator(jPool);

        //提交烧水的业务逻辑,取到异步任务
        ListenableFuture<Boolean> hotFuture = gPool.submit(hotJob);
        //绑定任务执行完成后的回调,到异步任务
        Futures.addCallback(hotFuture, new FutureCallback<Boolean>() {
            public void onSuccess(Boolean r) {
                if (r) {
                    mainJob.warterOk = true;
                }
            }

            public void onFailure(Throwable t) {
                Logger.info("烧水失败,没有茶喝了");
            }
        },jPool);
        //提交清洗的业务逻辑,取到异步任务

        ListenableFuture<Boolean> washFuture = gPool.submit(washJob);
        //绑定任务执行完成后的回调,到异步任务
        Futures.addCallback(washFuture, new FutureCallback<Boolean>() {
            public void onSuccess(Boolean r) {
                if (r) {
                    mainJob.cupOk = true;
                }
            }

            public void onFailure(Throwable t) {
                Logger.info("杯子洗不了,没有茶喝了");
            }
        },jPool);
    }
}

官方代码有误,修改后得执行结果为:

[pool-1-thread-1|GuavaFutureDemo$HotWarterJob.call] |>  洗好水壶 
[pool-1-thread-1|GuavaFutureDemo$HotWarterJob.call] |>  灌上凉水 
[pool-1-thread-1|GuavaFutureDemo$HotWarterJob.call] |>  放在火上 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[pool-1-thread-2|GuavaFutureDemo$WashJob.call] |>  洗茶壶 
[pool-1-thread-2|GuavaFutureDemo$WashJob.call] |>  洗茶杯 
[pool-1-thread-2|GuavaFutureDemo$WashJob.call] |>  拿茶叶 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[pool-1-thread-2|GuavaFutureDemo$WashJob.call] |>  洗完了 
[pool-1-thread-2|GuavaFutureDemo$WashJob.call] |>  清洗工作  运行结束. 
[pool-1-thread-1|GuavaFutureDemo$HotWarterJob.call] |>  水开了 
[pool-1-thread-1|GuavaFutureDemo$HotWarterJob.call] |>  烧水工作,运行结束. 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 
[主线程|GuavaFutureDemo$MainJob.drinkTea] |>  泡茶喝,茶喝完 
[主线程|GuavaFutureDemo$MainJob.run] |>  读书中...... 

通过代码,我们可以看到在等待泡茶的过程中,我们可以读书。

Guava异步回调和Java的FutureTask异步回调,本质的不同在于:

  • Guava是非阻塞的异步回调,调用线程是不阻塞的,可以继续执行自己的业务逻辑。
  • FutureTask是阻塞的异步回调,调用线程是阻塞的,在获取异步结果的过程中,一直阻塞,等待异步线程返回结果。

5.5 Netty的异步回调模式

Netty官方文档中指出Netty的网络操作都是异步的。在Netty源代码中,大量使用了异步回调处理模式。在Netty的业务开发层面,Netty应用的Handler处理器中的业务处理代码,也都是异步执行的。所以,了解Netty的异步回调,无论是Netty应用级的开发还是源代码级的开发,都是十分重要的。

Netty和Guava一样,实现了自己的异步回调体系:Netty继承和扩展了JDK Future系列异步回调的API,定义了自身的Future系列接口和类,实现了异步任务的监控、异步执行结果的获取。

总体来说,Netty对JavaFuture异步任务的扩展如下:

  1. 继承Java的Future接口,得到了一个新的属于Netty自己的Future异步任务接口;该接口对原有的接口进行了增强,使得Netty异步任务能够以非阻塞的方式处理回调的结果;注意,Netty没有修改Future的名称,只是调整了所在的包名,Netty的Future类的包名和Java的Future接口的包名不同。

  2. 引入了一个新接口——GenericFutureListener,用于表示异步执行完成的监听器。这个接口和Guava的FutureCallbak回调接口不同。Netty使用了监听器的模式,异步任务的执行完成后的回调逻辑抽象成了Listener监听器接口。可以将Netty的GenericFutureListener监听器接口加入Netty异步任务Future中,实现对异步任务执行状态的事件监听。

总体上说,在异步非阻塞回调的设计思路上,Netty和Guava的思路是一致的。对应关系为:

  • Netty的Future接口,可以对应到Guava的ListenableFuture接口。
  • Netty的GenericFutureListener接口,可以对应到Guava的FutureCallback接口。

5.5.1 详解GenericFutureListener接口

前面提到,和Guava的FutureCallback一样,Netty新增了一个接口来封装异步非阻塞回调的逻辑——它就是GenericFutureListener接口。

GenericFutureListener位于io.netty.util.concurrent包中,源代码如下:

package io.netty.util.concurrent;
import java.util.EventListener;
public interface GenericFutureListener<F extends Future<?>> extends EventListener {
   //监听器的回调方法
    void operationComplete(F var1) throws Exception;
}

GenericFutureListener拥有一个回调方法:operationComplete,表示异步任务操作完成。在Future异步任务执行完成后,将回调此方法。在大多数情况下,Netty的异步回调的代码编写在GenericFutureListener接口的实现类中的operationComplete方法中。

说明一下,GenericFutureListener的父接口EventListener是一个空接口,没有任何的抽象方法,是一个仅仅具有标识作用的接口。

5.5.2 详解Netty的Future接口

Netty也对Java的Future接口进行了扩展,并且名称没有变,还是叫作Future接口,代码实现位于io.netty.util.concurrent包中。

和Guava的ListenableFuture一样,Netty的Future接口,扩展了一系列的方法,对执行的过程的进行监控,对异步回调完成事件进行监听(Listen)。Netty的Future接口的源代码如下:

public interface Future<V>  extendsjava.util.concurrent.Future<V> {
        booleanisSuccess(); // 判断异步执行是否成功
        booleanisCancellable(); // 判断异步执行是否取消
        Throwable cause();//获取异步任务异常的原因
    //增加异步任务执行完成与否的监听器Listener
        Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener); 
     //移除异步任务执行完成与否的监听器Listener
     Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);  
        //....
}

Netty的Future接口一般不会直接使用,而是会使用子接口。Netty有一系列的子接口,代表不同类型的异步任务,如ChannelFuture接口。

ChannelFuture子接口表示通道IO操作的异步任务;如果在通道的异步IO操作完成后,需要执行回调操作,就需要使用到ChannelFuture接口。

5.5.3 ChannelFuture的使用

在Netty的网络编程中,网络连接通道的输入和输出处理都是异步进行的,都会返回一个ChannelFuture接口的实例。通过返回的异步任务实例,可以为它增加异步回调的监听器。在异步任务真正完成后,回调才会执行。

Netty的网络连接的异步回调,实例代码如下:

//connect是异步的,仅提交异步任务
ChannelFuture future    = bootstrap.connect(
                new     InetSocketAddress("www.manning.com",80));
//connect的异步任务真正执行完成后,future回调监听器才会执行
future.addListener(new  ChannelFutureListener() {
                @Override
                public void  operationComplete(ChannelFuture    channelFuture)
                        throws  Exception       {
                                if(channelFuture.isSuccess()){
                                        System.out.println("Connection  established");
                                }       else    {
                                        System.err.println("Connection  attempt failed");
                                        channelFuture.cause().printStackTrace();
                                }
                        }
                });

GenericFutureListener接口在Netty中是一个基础类型接口。在网络编程的异步回调中,一般使用Netty中提供的某个子接口,如ChannelFutureListener接口。在上面的代码中,使用的是这个子接口。

5.5.4 Netty的出站和入站异步回调

Netty的出站和入站操作都是异步的。异步回调的方法,和上面Netty建立的异步回调是一样的。

以最为经典的NIO出站操作——write为例,说明一下ChannelFuture的使用。

在调用write操作后,Netty并没有完成对Java NIO底层连接的写入操作,因为是异步执行的。代码如下:

//write输出方法,返回的是一个异步任务
ChannelFuture future = ctx.channel().write(msg);
//为异步任务,加上监听器
future.addListener(
                new ChannelFutureListener()
                {
                                @Override
                                public void operationComplete(ChannelFuture future)
                                {
                                                // write操作完成后的回调代码
                                }
                });

在调用write操作后,是立即返回,返回的是一个ChannelFuture接口的实例。通过这个实例,可以绑定异步回调监听器,这里的异步回调逻辑需要我们编写。

如果大家运行以上的EchoServer实践案例,就会发现一个很大的问题:客户端接收到的回显信息和发送到服务器的信息,不是一对一对应输出的。看到的比较多的情况是:客户端发出很多次信息后,客户端才收到一次服务器的回显。

这是什么原因呢?这就是网络通信中的粘包/半包问题。对于这个问题的解决方案,在后面会做非常详细的解答,这里暂时搁置。粘包/半包问题的出现,说明了一个问题:仅仅基于Java的NIO,开发一套高性能、没有Bug的通信服务器程序,远远没有大家想象的简单,有一系列的“坑”、一大堆的基础问题等着大家解决。

在进行大型的Java通信程序的开发时,尽量基于一些实现了成熟、稳定的基础通信的Java开源中间件(如Netty)。这些中间件已经帮助大家解决了很多的基础问题,如前面出现的粘包/半包问题。

至此,大家已经学习了Java NIO、Reactor反应器模式、Future模式,这些都是学习Netty应用开发的基础。基础知识已经铺垫得差不多了,接下来到了正式进入学习Netty的阶段。

5.6 本章小结

随着高并发系统越来越多,异步回调模式也越来越重要。在Netty源代码中,大量地使用了异步回调技术,所以,在开始介绍Netty之前,开辟整整一章,非常详细地、由浅入深地为大家介绍了异步回调模式。

本章首先为大家介绍了Java的join合并线程时“闷葫芦式”的异步阻塞,然后介绍了Java的FutureTask阻塞式的获取异步任务结果,最后介绍了Guava和Netty的异步回调方式。

Guava和Netty的异步回调是非阻塞的,而Java的join、FutureTask都是阻塞的。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 194,242评论 5 459
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 81,769评论 2 371
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 141,484评论 0 319
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 52,133评论 1 263
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 61,007评论 4 355
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 46,080评论 1 272
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 36,496评论 3 381
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 35,190评论 0 253
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 39,464评论 1 290
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,549评论 2 309
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,330评论 1 326
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 32,205评论 3 312
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 37,567评论 3 298
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 28,889评论 0 17
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,160评论 1 250
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 41,475评论 2 341
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 40,650评论 2 335

推荐阅读更多精彩内容