Netty(十五)——Future、Promise之略

我会带着你远行 2022-04-05 09:43 139阅读 0赞

在总结Netty的过程中，我也在边总结多线程，因为Netty当中好多用到了多线程，比如Netty中的ChannelFuture、ChannelPromise的其实就是来源于上篇博客中的 [Java高并发（七）——Future模式][Java_Future] ，将线程之间的交互通过异步通知的形式进行实现，提高效率，提升性能。当然在总结多线程中，其实有些东西想说的更明白，又会涉及到JVM的的东西，数据结构的知识，在接下来，我也会陆续将一下JVM的内存模型，GC，类文件结构、JVM加载机制、执行机制等进行总结，还有数据结构的慢慢开始。知识都是想通的，在构建一块一块知识网的同时，不同的知识网也就会慢慢有连接连接起来，最终形成一个大的知识网，然后再提炼，在抽象，在总结，再升华，能搞达到抓住哪些牵一发动全身核心链条，那么我想这个知识网会越来越松弛有度的。

好，废话不说，今天简单总结一下Netty的Future和Promise相关知识，也是系列Netty核心类的最后一篇了。前边我们对ByteBuf、Channel、Unsafe、ChannelPipeline、ChannelHandler、EventLoop等核心类做了功能和源码的相关总结。

一，ChannelFuture功能简介：

ChannelFuture继承自JDK的Future，在原来的功能扩展了很多，主要都是异步I/O操作相关的，使Netty实现任何I/O调用都会立即返回，不进行同步等待。看下ChannelFuture的几个方法吧。添加Listener后，IO操作完成后线程回调GenericFutureListener的operationComplete方法，把ChannelFuture当做入参。

public interface ChannelFuture extends Future<Void> {
    
        /**
         * Returns a channel where the I/O operation associated with this
         * future takes place.
         */
        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();
    
        /**
         * Returns {@code true} if this {@link ChannelFuture} is a void future and so not allow to call any of the
         * following methods:
         * <ul>
         *     <li>{@link #addListener(GenericFutureListener)}</li>
         *     <li>{@link #addListeners(GenericFutureListener[])}</li>
         *     <li>{@link #await()}</li>
         *     <li>{@link #await(long, TimeUnit)} ()}</li>
         *     <li>{@link #await(long)} ()}</li>
         *     <li>{@link #awaitUninterruptibly()}</li>
         *     <li>{@link #sync()}</li>
         *     <li>{@link #syncUninterruptibly()}</li>
         * </ul>
         */
        boolean isVoid();
    }
    
    /**
     * Listens to the result of a {@link Future}.  The result of the asynchronous operation is notified once this listener
     * is added by calling {@link Future#addListener(GenericFutureListener)}.
     */
    public interface GenericFutureListener<F extends Future<?>> extends EventListener {
    
        /**
         * Invoked when the operation associated with the {@link Future} has been completed.
         *
         * @param future  the source {@link Future} which called this callback
         */
        void operationComplete(F future) throws Exception;
    }

AbstractFuture实现Future接口，它不允许I/O操作被取消，看下获取结果的实现（其它子类（很多）的具体方法实现不再一一看了）：

/**
     * Abstract {@link Future} implementation which does not allow for cancellation.
     *
     * @param <V>
     */
    public abstract class AbstractFuture<V> implements Future<V> {
    
        @Override
        public V get() throws InterruptedException, ExecutionException {
            await();
    
            Throwable cause = cause();
            if (cause == null) {
                return getNow();
            }
            throw new ExecutionException(cause);
        }
    
        @Override
        public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            if (await(timeout, unit)) {
                Throwable cause = cause();
                if (cause == null) {
                    return getNow();
                }
                throw new ExecutionException(cause);
            }
            throw new TimeoutException();
        }
    }

二，Promise功能简介：

Promise是可写的Future，Netty通过Promise对Future进行了扩展，用于设置I/O操作的结果，例如在写、读操作时新疆一个Promise然后根基操作结果进行添加操作结果。看一下新增的一些写的操作：

/**
     * Special {@link Future} which is writable.
     */
    public interface Promise<V> extends Future<V> {
    
        /**
         * Marks this future as a success and notifies all
         * listeners.
         *
         * If it is success or failed already it will throw an {@link IllegalStateException}.
         */
        Promise<V> setSuccess(V result);
    
        /**
         * Marks this future as a success and notifies all
         * listeners.
         *
         * @return {@code true} if and only if successfully marked this future as
         *         a success. Otherwise {@code false} because this future is
         *         already marked as either a success or a failure.
         */
        boolean trySuccess(V result);
    
        /**
         * Marks this future as a failure and notifies all
         * listeners.
         *
         * If it is success or failed already it will throw an {@link IllegalStateException}.
         */
        Promise<V> setFailure(Throwable cause);
    
        /**
         * Marks this future as a failure and notifies all
         * listeners.
         *
         * @return {@code true} if and only if successfully marked this future as
         *         a failure. Otherwise {@code false} because this future is
         *         already marked as either a success or a failure.
         */
        boolean tryFailure(Throwable cause);
    
        /**
         * Make this future impossible to cancel.
         *
         * @return {@code true} if and only if successfully marked this future as uncancellable or it is already done
         *         without being cancelled.  {@code false} if this future has been cancelled already.
         */
        boolean setUncancellable();
    
    
        …………………………
    }

这里简单看一下其实现类DefaultPromise的setSuccess方法吧，其它的自己可以看下

// 一，setSuccess方法
        @Override
        public Promise<V> setSuccess(V result) {
            if (setSuccess0(result)) {
                notifyListeners();
                return this;
            }
            throw new IllegalStateException("complete already: " + this);
        }
    
        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;
        }
    
    
        private void notifyListeners() {
            // This method doesn't need synchronization because:
            // 1) This method is always called after synchronized (this) block.
            //    Hence any listener list modification happens-before this method.
            // 2) This method is called only when 'done' is true.  Once 'done'
            //    becomes true, the listener list is never modified - see add/removeListener()
    
            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;
                    }
                });
            }
        }

好，这篇就简单总结了一下Netty中IO操作的异步调用通知的实现ChannelFuture和ChannelPromise，其实明白了JDK中的Future，这边只是在IO操作上的运用，相对来说还是比较容易理解的。异步IO操作结果回调通知，提供性能。多看多反复多理解……

[Java_Future]: https://blog.csdn.net/liujiahan629629/article/details/84928857