Netty中的Future / 憋错料

先看下Future的整个继承体系，还有一个ChannelFuture不在里面；

在并发编程中，我们通常会用到一组非阻塞的模型：Promise，Future 和 Callback。其中的 Future 表示一个可能还没有实际完成的异步任务的结果，针对这个结果可以添加 Callback 以便在任务执行成功或失败后做出对应的操作，而 Promise 交由任务执行者，任务执行者通过 Promise 可以标记任务完成或者失败。可以说这一套模型是很多异步非阻塞架构的基础。

这一套经典的模型在 Scala、C# 中得到了原生的支持，但 JDK 中暂时还只有无 Callback 的 Future 出现，当然也并非在 JAVA 界就没有发展了，比如 Guava 就提供了ListenableFuture 接口，而 Netty 4+ 更是提供了完整的 Promise、Future 和 Listener 机制，在 Netty 的官方文档 Using as a generic library 中也介绍了将 Netty 作为一个 lib 包依赖，并且使用 Listenable futures 的示例。在实际的项目使用中，发现 Netty 的 EventLoop 机制不一定适用其他场景，因此想去除对 EventLoop 的依赖，实现一个简化版本。

Netty自己实现了一套并发库，Future是其中的一块，下篇文章讲下他的并发库的线程池实现。Netty的Future的特性

Future<V>的V为异步结果的返回类型
getNow 是无阻塞调用，返回异步执行结果，如果未完成那么返回null
await 是阻塞调用,等到异步执行完成
isSuccess 执行成功是否成功
sync 阻塞调用，等待这个future直到isDone（可能由于正常终止、异常或取消而完成）返回true; 如果该future失败，重新抛出失败的原因。和await区别就是返回结果不同，它返回一个Future对象，通过这个Future知道任务执行结果。
添加GenericFutureListener，执行完成(future可能由于正常终止、异常或取消而完成)后调用该监听器。

如果我实现这个Future怎么实现：1：任务执行器，这样我们可以控制任务的执行了。2：执行结果，用于返回。3：监听器集合。4：执行结果状态

5：触发监听器的函数。那么怎么执行完成后自定触发监听器，应该在Future里面又启动了一个线程去执行这个触发机制。这些都是我的猜想。看下他的子类怎么实现的：

AbstractFuture实现了JDK中的get方法，调用netty中future的方法，一个模板模式出现了。所有的Future都会继承该类

Promise：任务执行者可以标记任务执行成功或失败，添加了setFailure(Throwable)可以知道Promise的子类需要有个成员变量来保存异常，添加了setSuccess(V) 方法，setUncancellable()方法，和tryFailure和trySuccess方法。这样所有的操作都可以返回一个Promise，你自己检测是否执行成功，好处是啥，接口统一吗？

ScheduledFuture：一个定时任务的执行结果

ProgressiveFuture：可以跟踪任务的执行进度。

下面看下几个类的具体实现：

DefaultPromise：成员变量如下：


 private final EventExecutor executor; //任务执行器
 private volatile Object result;  //不仅仅是结果，也有可能是异常

* 一个或多个监听器，可能是GenericFutureListener或者DefaultFutureListeners。如果是NULL有两种可能* 1：没有添加触发器* 2：已经出发了

 private Object listeners;
 private LateListeners lateListeners;
 private short waiters;

看来异常也是结果。

看一个重要方法的实现：

isDone:可能由于正常终止、异常或取消而完成


    private static boolean isDone0(Object result) {
        return result != null && result != UNCANCELLABLE;
    }

isSuccess: 任务执行成功


    public boolean isSuccess() {
        Object result = this.result;
        if (result == null || result == UNCANCELLABLE) {
            return false;
        }
        return !(result instanceof CauseHolder);
    }

getNow:返回执行结果


    public V getNow() {
        Object result = this.result;
        if (result instanceof CauseHolder || result == SUCCESS) {
            return null;
        }
        return (V) result;
    }

sync:同步阻塞方法


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

看来比await多了抛出异常。

await：等待任务执行完成


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

        if (Thread.interrupted()) {
            throw new InterruptedException(toString());
        }

        synchronized (this) {
            while (!isDone()) {
                checkDeadLock();   //判断当前线程是否是执行线程。如果是抛出异常。
                incWaiters();     //添加等待个数
                try {
                    wait();        //释放锁，等待唤醒，阻塞该线程
                } finally {
                    decWaiters();
                }
            }
        }
        return this;
    }

执行器所在的线程不能调用await(),只能是调用者所在的线程才可以，waiters有什么用呢？

cancle方法使用：


    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        Object result = this.result;
        if (isDone0(result) || result == UNCANCELLABLE) {
            return false;
        }

        synchronized (this) {
            // Allow only once.
            result = this.result;
            if (isDone0(result) || result == UNCANCELLABLE) {
                return false;
            }

            this.result = CANCELLATION_CAUSE_HOLDER;
            if (hasWaiters()) {
                notifyAll();
            }
        }

        notifyListeners();
        return true;
    }

当我们修改DefaultPromise的状态时，要触发监听器。

notifyListeners:


    /**
     * 该方法不需要异步，为啥呢
     * 1：这个方法在同步代码块里面调用，因此任何监听器列表的改变都happens-before该方法
     * 2：该方法只有isDone==true的时候调用，一但 isDone==true 那么监听器列表将不会改变
     */
    private void notifyListeners() {

        Object listeners = this.listeners;
        if (listeners == null) {
            return;
        }

        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 {
                    if (listeners instanceof DefaultFutureListeners) {
                        notifyListeners0(this, (DefaultFutureListeners) listeners);
                    } else {
                        final GenericFutureListener<? extends Future<V>> l =
                                (GenericFutureListener<? extends Future<V>>) listeners;
                        notifyListener0(this, l);
                    }
                } finally {
                    this.listeners = null;
                    threadLocals.setFutureListenerStackDepth(stackDepth);
                }
                return;
            }
        }

        if (listeners instanceof DefaultFutureListeners) {
            final DefaultFutureListeners dfl = (DefaultFutureListeners) listeners;
            execute(executor, new Runnable() {
                @Override
                public void run() {
                    notifyListeners0(DefaultPromise.this, dfl);
                    DefaultPromise.this.listeners = null;
                }
            });
        } else {
            final GenericFutureListener<? extends Future<V>> l =
                    (GenericFutureListener<? extends Future<V>>) listeners;
            execute(executor, new Runnable() {
                @Override
                public void run() {
                    notifyListener0(DefaultPromise.this, l);
                    DefaultPromise.this.listeners = null;
                }
            });
        }
    }

任务永远不能在主线程中执行，需要放到执行器所在的线程执行。DefaultFutureListeners和GenericFutureListener，一个是容器，一个是元素

还有几个修改状态的方法：


    @Override
    public boolean setUncancellable() {
        Object result = this.result;
        if (isDone0(result)) {
            return !isCancelled0(result);
        }

        synchronized (this) {
            // Allow only once.
            result = this.result;
            if (isDone0(result)) {
                return !isCancelled0(result);
            }

            this.result = UNCANCELLABLE;
        }
        return true;
    }

    private boolean setFailure0(Throwable cause) {
        if (cause == null) {
            throw new NullPointerException("cause");
        }

        if (isDone()) {
            return false;
        }

        synchronized (this) {
            // Allow only once.
            if (isDone()) {
                return false;
            }

            result = new CauseHolder(cause);
            if (hasWaiters()) {
                notifyAll();
            }
        }
        return true;
    }

    private boolean setSuccess0(V result) {
        if (isDone()) {
            return false;
        }

        synchronized (this) {
            // Allow only once.
            if (isDone()) {
                return false;
            }
            if (result == null) {
                this.result = SUCCESS;
            } else {
                this.result = result;
            }
            if (hasWaiters()) {
                notifyAll();
            }
        }
        return true;
    }

CompleteFuture的几个子类是状态Promise

PromiseTask:该类继承了RunnableFuture接口，该类表示异步操作的结果也可以异步获得，类似JDK中的FutureTask，实例化该对象时候需要传一个Callable的对象，如果没有该对象可以传递一个Runnable和一个Result构造一个Callable对象。


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

        @Override
        public T call() {
            task.run();
            return result;
        }

        @Override
        public String toString() {
            return "Callable(task: " + task + ", result: " + result + ‘)‘;
        }
    }


看下他的run方法。

    @Override
    public void run() {
        try {
            if (setUncancellableInternal()) {
                V result = task.call();
                setSuccessInternal(result);
            }
        } catch (Throwable e) {
            setFailureInternal(e);
        }
    


该类的setFailure,setSuccess等方法都会抛出一异常，而加了internal的方法会成功执行，他们是protected，子类或者同一个package中可以调用。
ScheduledFutureTask:该类是定时任务返回的
ChannelFuture是这个结构中最重要的类，从名称可以知道是通道异步执行的结果：在netty中所有的IO操作都是异步的。这意味这所有的IO调用都会立即返回，且不保证IO操作完成。IO调用会返回一个ChannelFuture的实例，通过该实例可以查看IO操作的结果和状态，ChannelFuture有完成和未完成两种状态，当IO操作开始，就会创建一个ChannelFuture的实例，该实例初始是未完成状态，它不是成功，失败，或者取消，因为IO操作还没有完成，如果IO操作完成了那么将会有成功，失败，和取消状态，
*                                      +---------------------------+*                                      | Completed successfully    |*                                      +---------------------------+*                                 +---->      isDone() = <b>true</b>      |* +--------------------------+    |    |   isSuccess() = <b>true</b>      |* |        Uncompleted       |    |    +===========================+* +--------------------------+    |    | Completed with failure    |* |      isDone() = <b>false</b>    |    |    +---------------------------+* |   isSuccess() = false    |----+---->   isDone() = <b>true</b>         |* | isCancelled() = false    |    |    | cause() = <b>non-null</b>     |* |    cause() = null     |    |    +===========================+* +--------------------------+    |    | Completed by cancellation |*                                 |    +---------------------------+*                                 +---->      isDone() = <b>true</b>      |*                                      | isCancelled() = <b>true</b>      |*                                      +---------------------------+
 该类提供了很多方法用来检查IO操作是否完成，等待完成，和接受IO操作的结果。还可以添加ChannelFutureListener的监听器，这样IO操作完成时就可以得到提醒
 * 强烈建议使用addListener而不是await。
 * addListener是非阻塞的，它简单的添加指定的ChannelFutureListener到ChannelFuture中，
 * IO线程将在当绑定在这个future的IO操作完成时，触发这个触发器，优点是提高效率和资源的利用率
 * await()是一个阻塞方法，一旦调用，调用线程将会阻塞直到IO操作完成。优点是容易实现顺序逻辑

来自为知笔记(Wiz)

时间： 2024-10-05 08:08:59

Netty中的Future

Netty中的Future的相关文章

Netty中的那些坑

Netty 中 IOException: Connection reset by peer 与 java.nio.channels.ClosedChannelException: null

Netty中NioEventLoopGroup的创建源码分析

netty中使用IdleStateHandler来发起心跳

【转】Netty那点事（二）Netty中的buffer

理解Netty中的零拷贝（Zero-Copy）机制【转】

Netty（五）序列化protobuf在netty中的使用

Netty中execution包功能详解

Netty中LineBasedFrameDecoder解码器使用与分析：解决TCP粘包问题