Future 和 Promise 源码分析（二十七）

清疚 2022-10-29 14:20 167阅读 0赞

今天我们分析Future 和 Promise 源码

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

**1、JDK**  **中的**  **Future**

Future  最早来源于  JDK  的  java.util.concurrent.Future ，它用于代表异步操作的结果。相关API 如图  19-1  所示。 可以通过  get  方法获取操作结果，如果操作尚未完成，则会同步阻塞当前调用的线程 ; 如果不允许阻塞太长时间或者无限期阻塞，可以通过带超时时间的 get  方法获取结果 ; 如果到达超时时间操作仍然没有完成，则抛出 TimeoutException 。 通过 isDone() 方法可以判断当前的异步操作是否完成，如果完成，无论成功与否，都返回 true ，否则返回  false 。 通过 cancel  可以尝试取消异步操作，它的结果是未知的，如果操作已经完成，或者发

生其他未知的原因拒绝取消，取消操作将会失败。

![20210215160832136.png][]

**2、ChannelFuture**

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

***类层次关系***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

从类层次上来看， Netty  中在  JDK  并发  Future  的基础之上，根据自己的需要， extends 出了一个新的 Future  接口，然后以新  Future  接口为基础，再  extends  出  ChannelFuture  接口。

***3、ChannelFuture***  ***辨析***

ChannelFuture  有两种状态 : uncompleted  和  completed 。当开始一个  I/O  操作时，一个新的 ChannelFuture  被创建，此时它处于  uncompleted  状态——非失败、非成功、非取消，

因为  I/O  操作此时还没有完成。一旦  IO  操作完成 ,ChannelFuture  将会被设置成  completed, 它 的结果有如下三种可能。

操作成功 ;

操作失败 ;

操作被取消。

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

ChannelFuture  提供了一系列新的  API ，用于获取操作结果、添加事件监听器、取消  IO 操作、同步等待等。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

ChannelFuture 可以同时增加一个或者多个 GenericFutureListener，也可以用 remove 方法删除 GenericFutureListener。 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;
    }

当 IO 操作完成之后，IO 线程会回调 GenericFutureListener 的 operationComplete 方法，并把 ChannelFuture 对象当作方法的入参。如果用户需要做上下文相关的操作，需要将上下 文信息保存到对应的 ChannelFuture 中。建议通过添加监听器的方式获取 I/O 操作结果，或者进行后续的相关操作。推荐通过 GenericFutureListener 代替ChannelFuture的get等方法的原因是:当我们进行异步IО操作时，完成的时间是无法预测的，如果不设置超时时间，它会导致调用线程长时间被阻塞，甚至挂死。而设置超时时间，时间又无法精确预测。利用异步通知机制回调 GenericFutureListener 是最佳的解决方案，它的性能最优。

需要注意的是: 不要在  ChannelHandler  中调用  ChannelFuture  的  await() 方法，这会导致死锁。原因是发起 I/O  操作之后，由  I/O  线程负责异步通知发起  I О操作的用户线程，如果

I/ О线程和用户线程是同一个线程，就会导致  I О线程等待自己通知操作完成，这就导致了死锁，这跟经典的两个线程互等待死锁不同，属于自己把自己挂死。

***4、主要***  ***Future***  ***实现类源码解析***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

AbstractFuture

代码很简单，实现了  JDK  里  Future  接口的两个  get  方法。首先，调用 await() 方法进行无限期阻塞，当  I/O  操作完成后会被  notify() 。程序继续向下执行，检查 I/O  操作是否发生了异常，如果没有异常，则通过  getNow() 方法获取结果并返回。否则，将异常堆栈进行包装，抛出 ExecutionException 。支持超时很简单，调用 await(long timeout,TimeUnit unit) 方法即可。如果超时，则抛出TimeoutException。如果没有超时，则依次判断是否发生了  IO  异常等情况，操作与无参数的get 方法相同。 其他 ChannelFuture  的实现子类，由于功能比较简单，不再赘述。

**5、Promise**

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

![20210215161040916.png][]

方法图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

6、 ***主要*** ***Promise*** ***实现类源码解析***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

DefaultPromise 的 *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) {
        return setValue0(result == null ? SUCCESS : result);
    }

继续：

private boolean setValue0(Object objResult) {
        if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
            RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
            checkNotifyWaiters();
            return true;
        }
        return false;
    }

可以看见，Netty 实际是利用了原子类 AtomicReferenceFieldUpdater 来保证更新结果的线程安全。更新成功后，首先调用了 Object 的 notifyAll 方法唤醒等待的线程，然后在 notifyListeners 方法中一路调用，直到 notifyListener0 方法，执行监听器 GenericFutureListener 的operationComplete 方法。

*await*  *方法*

实现非常简单，调用了  Object  的  wait  方法。

*sync()*

调用了  await  方法，除此之外，多了个抛出异常的操作。

DefaultChannelPromise  和  CloseFuture  代码比较简单，请自行查阅

Future 和 Promise 源码分析完成，下篇分析Channel 接口实现类系列源码  ，敬请期待！

[20210215160832136.png]: /images/20221024/6ac646bbaf954b18a527ca01afaba224.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/ccac0ac0fc734f7bbc14ffa6bfb6daba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/4dfb58f35788415ca4202226ea645953.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/7369fbc7dfb3442bbc46a49a32845186.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/162c7976e8094a3d82f924bb25114e70.png
[20210215161040916.png]: /images/20221024/f076990588d44ef38b682a4962f87569.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/3177582fa381499d8d5d882226b52a9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/9e95d4ec748d42fab97147bc5e3e471b.png