[netty4][netty-common]Future与Promise分析

Posted 2021-12-06 simoncook

接口与类结构体系

-- [I]java.util.concurrent.Future<V>  
---- [I]io.netty.util.concurrent.Future<V>  
------ [AC]AbstractFuture, [I]ChannelFuture, [I]Promise 

-- [AC]AbstractFuture, [I]Promise         -- [I]ChannelFuture, [I]Promise      
---- DefaultPromise                       ---- [I]ChannelPromise

-- [I]ChannelPromise, [I]FlushCheckpoint, [C]DefaultPromise  
---- [C]DefaultChannelPromise

JDK的Future提供特性

1. 是否完成
   
2. 是否取消
   
3. 结果获取
4. 取消执行

netty的Future增加的特性

1. 是否成功(完成后的结果，完成不一定成功)
2. 是否能被取消
3. 如果失败时的异常获取
4. 支持监听器，监听操作完成的回调
5. sync 阻塞等待直至完成 // 跟get有什么区别？A: 只阻塞，不取结果，在一些实现逻辑中而有是否死锁等检查。
6. await 阻塞等待直至完成 // 跟get有什么区别？跟sync有什么区别？A: 在一些实现逻辑中多了在调完await之后再调用rethrowIfFailed(字面意思)。
7. getNow，非阻塞获取结果(可以理解成JDK Future的是否完成和结果获取的组合)
8. 重新覆写cancel行为定义

Promise增加的特性

整体定位是一个支持可写的Future，可以理解成：可以通过API设置结果的成功还是失败。对应netty的Future的特性1。

1. 设置是否成功的结果，并触发监听器operationComplete事件。操作失败抛异常
2. 尝试设置是否成功的结果， 并触发监听器operationComplete事件。操作失返回false
3. 设置是否失败的结果，并触发监听器operationComplete事件。操作失败抛异常
4. 尝试设置是否失败的结果， 并触发监听器operationComplete事件。操作失返回false
5. 设置是否可以取消
6. 覆写返回Future的方法签名为返回Promise

注意： 是否可以取消，是否成功等，在DefaultPromise实现中，是用一个result字段来实现的。并且用AtomicReferenceFieldUpdater结合volatile来完成在并发情况下字段的正确设置值。

ChannelPromise增加的特性

1. 覆写返回ChannelFuture的方法签名为返回ChannelPromise
2. unvoid行为(如果是void的则返回新的ChannelPromise，否则返回当前实例)

AbstractFuture完成的逻辑

完成get的实现逻辑，或者说定义的行为顺序，包含超时的get与一直等的get

               +-----------+
               |   await   |
               +-----+-----+
                     |
                     |
            +--------v-------+
            |  get cause     |
       +----+ cause == null  +---+
       |    +----------------+   |
       |                         |
       |                         |
+------v------+           +------v------+
| throw exp   |           |  getNow     |
+-------------+           +-------------+

DefaultPromise完成的逻辑

实现是线程安全的

1. 实现监听器添加、删除与触发逻辑。引入EventExecutor实例，一对一。 用于触发监听器时使用。触发监听器逻辑有栈深度保护策略。
2. 通过volatile Object result字段完成是否成功，是否取消的状态描述。
   
3. 实现设置(含尝试)成功，设置失败(含尝试)，设置不可取消的逻辑
4. 实现是否成功，是否取消的判断逻辑
5. 异常的获取，结果的获取
6. await逻辑的实现。依靠Object的wait方法实现。同时用short waiters来描述wait的次数。
7. 获取执行结果。执行结果也是volatile Object result字段承载。
8. 取消逻辑实现。设置result字段为CANCELLATION_CAUSE_HOLDER。notifyAll wait。notify所有监听器。
9. 是否取消的判断逻辑实现。比对result字段值，比对cause字段的异常类型
10. 是否完成的判断逻辑实现。与是否取消逻辑类似。
11. sync逻辑实现。在调完await之后再调用rethrowIfFailed(字面意思)。
12. 死锁检测逻辑实现。如果executor在eventLoop则死锁(executor.inEventLoop)。死锁扔异常BlockingOperationException

await逻辑

                  +----------------+
       +----Y-----+     isDone     |
  +----v---+      +----------------+
  | return |              N
  +------+-+              |
         ^        +-------v--------+
         +----Y---+  interrupted   |
                  +----------------+
                          N
                          |
+----------+      +-------v--------+
| throw exp<--Y---+  checkDeadLock |
+----------+      +----------------+
                          N
                          +---------synchronized----------+
                          |                               |
                          |                               |
                  +-------v--------+                      |
       while-loop-+   !isDone      +-----------+          |
          |       +----------------+   +-------v------+   |
          |                            |  incWaiters  |   |
          |                            +-------+------+   |
          |       +----------------+           |          |
          |       |     wait       <-----------+          |
          |       +-------+--------+                      |
          |               |            +--------------+   |
          |               +------------> decWaiters   |   |
          |                            +--------------+   |
          +-------------------------------------------+   |
                                                          |
                          +-------------------------------+

如果有其他人做了notify 但是此时任务还没有done，那么则会继续wait，因为这是一个while loop！

触发监听器逻辑有栈深度保护策略
前提是在同一个线程上，如果不是同一个线程就没有意义了。所以要判断executor.inEventLoop()。
在UnpaddedInternalThreadLocalMap中有个字段int futureListenerStackDepth字段来维护FutureListener的栈深度。
在同一个线程上，做notifyListener0之前会将futureListenerStackDepth加1，做完之后恢复原值。
这样如果在同一个线程递归调用到notifyListener0即notifyListener则会触发触发监听器逻辑有栈深度保护策略。
栈深度保护阈值默认是8，可以通过io.netty.defaultPromise.maxListenerStackDepth系统参数配置。

关于Promise的监听器
监听器是用listeners字段，这个字段是Object类型，竟然没有给一个明确的类型。在逻辑实现中有DefaultFutureListeners、GenericProgressiveFutureListener与GenericFutureListener等。
里面包了一个GenericFutureListener数组，达成一个复合的(列表型的)Listener。
GenericProgressiveFutureListener在netty自身里面没有用到具体实现。

安全执行任务的包装

   private static void safeExecute(EventExecutor executor, Runnable task) {
        try {
            executor.execute(task);
        } catch (Throwable t) {
            rejectedExecutionLogger.error("Failed to submit a listener notification task. Event loop shut down?", t);
        }
    }

注意： rejectedExecutionLogger 单独的日志名称，所以可以单独配置。