学习目标

• 为什么了解Netty异步监听？
• Netty如何实现异步监听的？

Future简介

我们知道Netty的I/O操作都是异步的，例如bind，connect，write等操作，会返回一个Future接口。Netty源码中大量使用了异步回调处理模式。在做Netty应用开发时，我们也会用到，所以，了解Netty的异步监听，无论是做Netty应用开发还是阅读源码都是十分重要的。

Future接口剖析

image

咱们通过上图的Echo Server的bind(...)方法来分析，返回一个ChannelFuture实例，点ChannelFuture接口，来看下类层次结构图：

future-2.png

可知：ChannelFuture继承了Netty的Future，Netty的Future继承了JDK的Future。

JDK的Future，我想大家应该非常熟悉了，用的最多的就是在线程池ThreadPoolExecutor时，通过submit方法提交任务返回一个Future实例，通过它来查询任务的执行状态和执行结果，最用常用的方法isDone()和get()。

Netty的Future，在继承JDK的Future基础上，扩展了自己方法，我们来看下：

public interface Future<V> extends java.util.concurrent.Future<V> {
    // 任务执行成功，返回true
    boolean isSuccess();
    // 任务被取消，返回true
    boolean isCancellable();
    // 支付执行失败，返回异常
    Throwable cause();
    // 添加Listener，异步非阻塞处理回调结果
    Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    // 移除Listener
    Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    // 同步阻塞等待任务结束；执行失败话，会将“异常信息”重新抛出来
    Future<V> sync() throws InterruptedException;
    Future<V> syncUninterruptibly();
    // 同步阻塞等待任务结束，和sync方法一样，只不过不会抛出异常信息
    Future<V> await() throws InterruptedException;
    Future<V> awaitUninterruptibly();
    boolean await(long timeout, TimeUnit unit) throws InterruptedException;
    boolean await(long timeoutMillis) throws InterruptedException;
    boolean awaitUninterruptibly(long timeout, TimeUnit unit);
    boolean awaitUninterruptibly(long timeoutMillis);
    // 非阻塞，获取执行结果
    V getNow();
    // 取消任务
    @Override
    boolean cancel(boolean mayInterruptIfRunning);
}

我们知道JDK的Future的任务结果获取需要主动查询，而Netty的Future通过添加监听器Listener，可以做到异步非阻塞处理任务结果，可以称为被动回调。

同步阻塞有两种方式：sync()和await()，区别：sync()方法在任务失败后，会把异常信息抛出；await()方法对异常信息不做任何处理，当我们不关心异常信息时可以用await()。通过阅读源码可知sync()方法里面其实调的就是await()方法。推荐使用：sync()方法。

// DefaultPromise 类

 @Override
public Promise<V> sync() throws InterruptedException {
    await();
    rethrowIfFailed();
    return this;
}

我们可以看到Future接口没有和IO操作关联在一起，接下来让我们来看看Future的子接口ChannelFuture，它和IO操作中的channel通道关联在一起了，用于异步处理channel事件，这个接口用的最多。

public interface ChannelFuture extends Future<Void> {
    // 获取channel通道
    Channel channel();
    @Override
    ChannelFuture addListener(GenericFutureListener<? extends Future<? super Void>> listener);
    @Override
    ChannelFuture addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
    @Override
    ChannelFuture removeListener(GenericFutureListener<? extends Future<? super Void>> listener);
    @Override
    ChannelFuture removeListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
    @Override
    ChannelFuture sync() throws InterruptedException;
    @Override
    ChannelFuture syncUninterruptibly();
    @Override
    ChannelFuture await() throws InterruptedException;
    @Override
    ChannelFuture awaitUninterruptibly();
    // 标记Futrue是否为Void，如果ChannelFuture是一个void的Future，不允许调// 用addListener()，await()，sync()相关方法
    boolean isVoid();
}

ChannelFuture接口相比父类Future接口，就增加了channel()和isVoid()两个方法，其他方法都是覆盖父类接口，没有别扩展。

到这了，顺便了解下ChannelFuture的状态转换：

* <pre>
 *                                      +---------------------------+
 *                                      | Completed successfully    |
 *                                      +---------------------------+
 *                                 +---->      isDone() = true      |
 * +--------------------------+    |    |   isSuccess() = true      |
 * |        Uncompleted       |    |    +===========================+
 * +--------------------------+    |    | Completed with failure    |
 * |      isDone() = false    |    |    +---------------------------+
 * |   isSuccess() = false    |----+---->      isDone() = true      |
 * | isCancelled() = false    |    |    |       cause() = non-null  |
 * |       cause() = null     |    |    +===========================+
 * +--------------------------+    |    | Completed by cancellation |
 *                                 |    +---------------------------+
 *                                 +---->      isDone() = true      |
 *                                      | isCancelled() = true      |
 *                                      +---------------------------+
 * </pre>

ChannelFuture就两种状态Uncompleted(未完成)和Completed(完成)，Completed包括三种，执行成功，执行失败和任务取消。注意：执行失败和任务取消都属于Completed。

让我们在来看下Future的另一个子接口Promise，它是个可写的Future，到底是写什么东西呢？

public interface Promise<V> extends Future<V> {
    // 执行成功，设置返回值，并通知所有listener,如果已经设置，则会抛出异常
    Promise<V> setSuccess(V result);
    // 设置返回值，如果已经设置，返回false
    boolean trySuccess(V result);
    // 执行失败，设置异常，并通知所有listener
    Promise<V> setFailure(Throwable cause);
    boolean tryFailure(Throwable cause);
    // 标记Future不可取消
    boolean setUncancellable();

    @Override
    Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    @Override
    Promise<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    @Override
    Promise<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    @Override
    Promise<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    @Override
    Promise<V> await() throws InterruptedException;
    @Override
    Promise<V> awaitUninterruptibly();
    @Override
    Promise<V> sync() throws InterruptedException;
    @Override
    Promise<V> syncUninterruptibly();
}

• Future接口只提供了获取返回值的get()方法，不可设置返回值。
• Promise接口在Future基础上，还提供了设置返回值和异常信息，并立即通知listeners。而且，一旦 setSuccess(...) 或 setFailure(...) 后，那些 await() 或 sync() 的线程就会从等待中返回。

接下来，让我们来看看ChannelFuture的可写的子接口ChannelPromise：

public interface ChannelPromise extends ChannelFuture, Promise<Void> {
    // 覆盖ChannelFuture接口
    @Override
    Channel channel();
    // 覆盖Promise接口
    @Override
    ChannelPromise setSuccess(Void result);
    ChannelPromise setSuccess();
    boolean trySuccess();
    @Override
    ChannelPromise setFailure(Throwable cause);

    @Override
    ChannelPromise addListener(GenericFutureListener<? extends Future<? super Void>> listener);
    @Override
    ChannelPromise addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
    @Override
    ChannelPromise removeListener(GenericFutureListener<? extends Future<? super Void>> listener);
    @Override
    ChannelPromise removeListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
    @Override
    ChannelPromise sync() throws InterruptedException;
    @Override
    ChannelPromise syncUninterruptibly();
    @Override
    ChannelPromise await() throws InterruptedException;
    @Override
    ChannelPromise awaitUninterruptibly();
    ChannelPromise unvoid();
}

ChannelPromise接口只是综合了ChannelFuture和Promise接口，没有新增功能。

截至到目前，异步监听相关的接口已经介绍完了，让我们通过一张图来概括下：

image

Promise的实现类

看看DefaultPromise的主要属性：

// setSuccess设置result为null时，设置成SUCCESS
private static final Object SUCCESS = new Object();
// 不可取消值
private static final Object UNCANCELLABLE = new Object();
// 取消异常信息持有者
private static final CauseHolder CANCELLATION_CAUSE_HOLDER = new CauseHolder(
        StacklessCancellationException.newInstance(DefaultPromise.class, "cancel(...)"));
//执行结果，使用volatile修饰，保证可见性
private volatile Object result;
// 当Promise执行完成时需要通知Listener，此时就使用这个executor
private final EventExecutor executor;
// 要通知的listener，因为它可能是不同的类型，所以定义为Object类型，使用时再判断类型
private Object listeners;
// 要使用wait()/notifyAll()机制，这个变量记录了waiter的数量
private short waiters;
// 是否正在通知listener
private boolean notifyingListeners;

从属性中可以看出， DefaultPromise 通过 Object 的 wait/notify 机制实现线程间的同步，通过 volatile 属性保证线程间的可见性。

通过setSuccess(...)方法，来了解下 执行成功后设置返回值并通知Listener的过程：

@Override
public Promise<V> setSuccess(V result) {
    // 第一次成功设置返回值后，返回Promise对象
    if (setSuccess0(result)) {
        return this;
    }
    // 否则抛出异常
    throw new IllegalStateException("complete already: " + this);
}
private boolean setSuccess0(V result) {
    // result为null，设置为SUCCESS
    return setValue0(result == null ? SUCCESS : result);
}
private boolean setValue0(Object objResult) {
    // CAS设置result值，只有当result为null或者UNCANCELLABLE，才可以执行成功
    if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
        RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
        // 唤醒等待的waiter，同时判断是否存在listener
        if (checkNotifyWaiters()) {
            // 通知所有的listener
            notifyListeners();
        }
        return true;
    }
    return false;
}
private synchronized boolean checkNotifyWaiters() {
    if (waiters > 0) {
        // 通知所有waiters
        notifyAll();
    }
    // 判断是否添加了监听者listeners
    return listeners != null;
}
private void notifyListeners() {
    EventExecutor executor = executor();
    // 是否是同一个线程
    if (executor.inEventLoop()) {
        final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
        final int stackDepth = threadLocals.futureListenerStackDepth();
        if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
            threadLocals.setFutureListenerStackDepth(stackDepth + 1);
            try {
                notifyListenersNow();
            } finally {
                threadLocals.setFutureListenerStackDepth(stackDepth);
            }
            return;
        }
    }
    // 用自己的executor执行
    safeExecute(executor, new Runnable() {
        @Override
        public void run() {
            notifyListenersNow();
        }
    });
}

DefaultPromise的方法实现逻辑挺简单，不再全部讲解了，只需知道通过 Object 的 wait/notify 机制实现线程间的同步和观察者设计模式进行通知就可以了。

DefaultPromise还有个子类DefaultChannelPromise，这个类在Netty中用的最多的，其内部逻辑调用的都是父类DefaultPromise的方法。DefaultChannelPromise类层次结构图如下：

image

应用实战

public class ChannelFuture_01 {

    public static void main(String[] args) throws InterruptedException {

        EventExecutor executor = new SelfEventExecutor();
        final Promise<String> promise = new DefaultPromise<String>(executor);


        promise.addListener(new GenericFutureListener<Future<? super String>>() {
            @Override
            public void operationComplete(Future<? super String> future) throws Exception {
                System.out.println("执行完成结果：" + future.get());
            }
        });


        final Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000);
                    promise.setSuccess("hello world!");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        });
        thread.start();

        promise.sync();
        System.out.println("==================");
    }

    static class SelfEventExecutor extends SingleThreadEventExecutor {


        public SelfEventExecutor() {
            this(null);
        }

        public SelfEventExecutor(EventExecutorGroup parent) {
            this(parent, new DefaultThreadFactory(SelfEventExecutor.class));
        }

        public SelfEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory) {
            super(parent, threadFactory, true);
        }

        @Override
        protected void run() {
            Runnable task = takeTask();
            if (task != null) {
                task.run();
            }
        }
    }
}

Future总结

Netty的Future继承JDK的Future，通过 Object 的 wait/notify机制，实现了线程间的同步；使用观察者设计模式，实现了异步非阻塞回调处理。