netty源码之Future和Promise

小咪咪 2022-12-29 01:47 114阅读 0赞

# netty源码之Future和Promise #

从名字可以看出，Future用于获取异步操作的结果，而Promise则比较抽象，无法直接猜测出其功能，下面我们一一来了解。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

## JDK中的Future ##

Future来源于JDK的`java.util.concurrent.Future`，它用于代表异步操作的结果，相关API如下所示。

// java.util.concurrent.Future
    public interface Future<V> { 
    
        // 通过cancel可以尝试取消异步操作，它的结果是未知的，
        // 如果操作已经完成，或者发生其他未知的原因拒绝取消，取消操作将会失败。
        boolean cancel(boolean mayInterruptIfRunning);
    
        // 返回是否被取消
        boolean isCancelled();
    
        // 操作完成，操作被取消，操作发生异常，都会返回true
        boolean isDone();
    
    	// 获取操作结果，如果操作尚未完成，则会同步阻塞当前调用的线程;
        V get() throws InterruptedException, ExecutionException;
    
    	// 在get()的基础上带上超时时间，如果到达超时时间操作仍然没有完成，则抛出 TimeoutException。
        V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

## Netty中的Future ##

Netty对JDK中的Future主要做了以下扩展：

*  区分操作完成的状态，是执行成功，还是被取消了，还是发生了异常。
 *  使用观察者模式增加了操作事件的监听机制。

// io.netty.util.concurrent.Future
    public interface Future<V> extends java.util.concurrent.Future<V> { 
    
        boolean isSuccess();
    
        boolean isCancellable();
    
        Throwable cause();
    
        Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    
        Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
    
        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();
    
        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();
    
        boolean cancel(boolean mayInterruptIfRunning);
    }

## ChannelFuture ##

由于Netty的Future都是与异步I/O操作相关的，因此，命名为ChannelFuture，代表它与Channel操作相关。

在Netty中，所有的I/O操作都是异步的，这意味着任何I/O调用都会立即返回，而不是像传统BIO那样同步等待操作完成。异步操作会带来一个问题：调用者如何获取异步操作的结果？ChannelFuture就是为了解决这个问题而专门设计的。

// io.netty.channel.ChannelFuture
    public interface ChannelFuture extends Future<Void> { 
    
        Channel channel();
    
        boolean isVoid();
    }

ChannelFuture有两种状态: uncompleted和completed。当开始一个I/O操作时，一个新的ChannelFuture被创建，此时它处于uncompleted状态（非失败、非成功、非取消），因为I/O操作此时还没有完成。一旦IO操作完成，ChannelFuture将会被设置成completed，它的结果有如下三种可能：

*  操作成功;
 *  操作失败;
 *  操作被取消。

ChannelFuture的状态迁移图如图所示：

需要注意的是：不要在ChannelHandler中调用ChannelFuture的await()方法，这会导致死锁。原因是发起I/O操作之后，由I/O线程负责异步通知发起IО操作的用户线程，如果I/О线程和用户线程是同一个线程，就会导致IО线程等待自己通知操作完成，这就导致了死锁，这跟经典的两个线程互等待死锁不同，属于自己把自己挂死。

## AbstractFuture ##

// io.netty.util.concurrent.AbstractFuture
    public abstract class AbstractFuture<V> implements Future<V> { 
    
        @Override
        public V get() throws InterruptedException, ExecutionException { 
            await(); // 调用JDK的await()方法进行无限期阻塞，当I/O操作完成后会被notify()。
    
            Throwable cause = cause();
            if (cause == null) { 
                return getNow(); // 没有异常，则通过getNow()方法获取结果并返回。
            }
            // 有异常则抛出
            if (cause instanceof CancellationException) { 
                throw (CancellationException) cause;
            }
            throw new ExecutionException(cause);
        }
    
        @Override
        public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 
            if (await(timeout, unit)) { 
            	// 后面的代码其实与上面的get()一致
                Throwable cause = cause();
                if (cause == null) { 
                    return getNow();
                }
                if (cause instanceof CancellationException) { 
                    throw (CancellationException) cause;
                }
                throw new ExecutionException(cause);
            }
            throw new TimeoutException();
        }
    }

## Promise ##

Promise是可写的Future，Future自身并没有写操作相关的接口，Netty通过Promise对Future进行扩展，用于设置IO操作的结果。

// io.netty.util.concurrent.Promise
    public interface Promise<V> extends Future<V> { 
    
        Promise<V> setSuccess(V result);
    
        boolean trySuccess(V result);
    
        Promise<V> setFailure(Throwable cause);
    
        boolean tryFailure(Throwable cause);
    
        boolean setUncancellable();
        ...
        ...
    }

## DefaultPromise ##

Netty 实际是利用了原子类AtomicReferenceFieldUpdater来保证更新结果的线程安全。

更新成功后，首先调用了Object的notifyAll方法唤醒等待的线程，然后在notifyListeners方法中一路调用，直到notifyListener0方法，执行监听器GenericFutureListener的operationComplete方法。

public Promise<V> setSuccess(V result) { 
        if (setSuccess0(result)) { 
            return this;
        }
        throw new IllegalStateException("complete already: " + this);
    }
    
    private boolean setSuccess0(V result) { 
        return setValue0(result == null ? SUCCESS : result);
    }
    
    private boolean setFailure0(Throwable cause) { 
        return setValue0(new CauseHolder(checkNotNull(cause, "cause")));
    }
    
    private boolean setValue0(Object objResult) { 
        if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
            RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) { 
            if (checkNotifyWaiters()) { 
                notifyListeners();
            }
            return true;
        }
        return false;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/e63436b4e8984e4eb341f11da393f9e2.png