1、Future与ChannelFuture

• Future未来(期望)，Netty事件都是异步
• Future的get方法是阻塞，Netty对JDK的Future进行异步的封装，如：判断状态isSuccess方法（成功或失败状态）改良点
• ChannelFuture是Future的特化，维护了一个Channel对象
• ChannelPromise(复合管道)是ChannelFuture的子类
• V get() 读取数据

ChannelPromise复合管道

public interface Future<V> extends java.util.concurrent.Future<V> {
    
    boolean isSuccess();  //当且仅当i/o操作成功完成时，返回true
    boolean isCancellable(); //当且仅当可以通过cancel方法取消时，返回true
    Throwable cause();//如果i/o操作失败，返回其失败原因。如果成功完成或者还未完成，返回null
    //将指定的监听器添加到此Future，future完成时，会通知此监听器，如果添加监听器时future已经完成，则立即通知此监听器
    Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    //移除监听器，如果future已经完成，则不会触发
    Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    //等待此future done
   Future<V> sync() throws InterruptedException;
    //等待，不可中断
    Future<V> syncUninterruptibly();
    //等待Future完成
  Future<V> await() throws InterruptedException;
    //等待Future完成，不可中断
  Future<V> awaitUninterruptibly();
  //立即返回，如果此时future未完成，返回null
  V getNow();

    @Override
    boolean cancel(boolean mayInterruptIfRunning);
}

2、DefaultProgressivePromise

• DefaultProgressivePromise类的awaitUninterruptibly、setSuccess、setFailure方法

   @Override//等待唤醒呼叫
    public ProgressivePromise<V> awaitUninterruptibly() {
        super.awaitUninterruptibly();
        return this;
    }

 @Override//设置监听成功状态
    public ProgressivePromise<V> setSuccess(V result) {
        super.setSuccess(result);
        return this;
    }
@Override @Override//设置监听失败状态
    public ProgressivePromise<V> setFailure(Throwable cause) {
        super.setFailure(cause);
        return this;
    }

3、DefaultPromise

• DefaultPromise的awaitUninterruptibly方法

 @Override
    public Promise<V> awaitUninterruptibly() {
        if (isDone()) {
            return this;
        }
        //检测死锁问题
        checkDeadLock();
        boolean interrupted = false;
        synchronized (this) {
            while (!isDone()) {
                incWaiters();//调用等待,自身++
                try {
                    wait();//等待
                } catch (InterruptedException e) {
                    // 打断等待
                    interrupted = true;
                } finally {
                    //减轻,自身 --
                    decWaiters();
                }
            }
        }
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
        return this;
    }

• DefaultPromise的setSuccess方法

 public Promise<V> setSuccess(V result) {
        if (setSuccess0(result)) {
            //成功后通知监听器
            notifyListeners();
            return this;
        }
        throw new IllegalStateException("complete already: " + this);
    }

• DefaultPromise的setFailure方法

   public Promise<V> setFailure(Throwable cause) {
        if (setFailure0(cause)) {
            notifyListeners();//通知监听器
            return this;
        }
        throw new IllegalStateException("complete already: " + this, cause);
    }

• DefaultPromise的notifyListeners方法,其实用到了观察者模式

GOF设计模式中有一种叫做观察者模式（Observer），属于行为型模式。又叫发布-订阅(Publish/Subscribe)模式、模型-视图 (Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。观察者模式定义了一种一对多的依赖关系，让多个观察者对象同时监听某一个主题对象。这个 主题对象在状态上发生变化时，会通知所有观察者对象，

private void notifyListeners() {
       //事件执行器
        EventExecutor executor = executor();
      //判断是否在事件循环里面
        if (executor.inEventLoop()) {
            final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();//从InternalThreadLocalMap 线程的值
        //异步监听堆栈的值
            final int stackDepth = threadLocals.futureListenerStackDepth();
            if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
                threadLocals.setFutureListenerStackDepth(stackDepth + 1);
                try {
                    //立即监听通知
                    notifyListenersNow();
                } finally {
                    threadLocals.setFutureListenerStackDepth(stackDepth);
                }
                return;
            }
        }
        //安全执行监听通知
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                notifyListenersNow();
            }
        });
    }

• DefaultPromise的notifyListeners0方法

  private void notifyListeners0(DefaultFutureListeners listeners) {
        GenericFutureListener<?>[] a = listeners.listeners();
        int size = listeners.size();
        for (int i = 0; i < size; i ++) {
            notifyListener0(this, a[i]);
        }
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private static void notifyListener0(Future future, GenericFutureListener l) {
        try {
            l.operationComplete(future);
        } catch (Throwable t) {
            logger.warn("An exception was thrown by " + l.getClass().getName() + ".operationComplete()", t);
        }
    }

3、JDK的Future

• JDK所提供的Future，只能通过手工方式检查执行结果，而这个操作是阻塞的，Netty对ChannelFuture进行的增强，通过ChannelFutureListerner以回调的方式获取执行结果，去除了手工检查阻塞的操作。
• 值得注意的是ChannelFutureListerner的operatonComplete方法是由I/O线程执行，因此要注意的是 不要在这里执行耗时操作，否则需要通过另外的线程或线程池进行执行

4、JDK的Void

• jdk的Void类实例化对象的关键字的处理

public interface ChannelFuture extends Future<Void> 

5、sync 方法

• 在server的绑定端口的sync是同步处理

  
  sync

6、回调函数

• 回调函数当方法、函数执行完成之后进行回调获取

7、Future关系图

• Future传统处理

  
  Future传统处理

• Netty的Future处理
  Netty的Future对JDK的Future进行改良，加了Listeners监听器进行处理

//添加监听器
Future<V> addListener(GenericFutureListener<? extends Future<? super V>> var1);
//添加监听器
Future<V> addListeners(GenericFutureListener... var1);
//移除监听器
Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> var1);
//移除监听器
Future<V> removeListeners(GenericFutureListener... var1);

Netty的Future处理