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

接口与类结构体系

-- [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 单独的日志名称,所以可以单独配置。

原文地址:https://www.cnblogs.com/simoncook/p/10924578.html