Netty异步读写操作的架构与观察者模式的应用

1.Netty中的Future

Netty中的Future继承了ava.util.concurrent.Future,java的Future主要是运行或者取消任务，而Netty中增加了更多的功能。

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.ChannelFuture

1.ChannelFuture是异步channel i/o操作的结果。它继承了Netty中的Future.Netty中的所有I/O操作都是异步的，这意味着任何I/O操作都将立即返回，并且不保证在调用结束时所有的I/O操作已经完成，相反，将会返回一个ChannelFuture实例，该实例提供了I/O操作的结果或者状态的信息。

3.ChannelFuture提供了各种方法来检查I/O操作是否已经完成，等待完成，并返回结果。还可以添加ChannelFutureListener，以便在完成时得到通知。
4.addListener(GenericFutureListener)方法是非阻塞的。它只是将指定的ChannelFutureListener添加到ChannelFuture,并且I/O线程会在与I/O操作关联的Future完成时（isDone）通知监听器。ChannelFutureListener产生最佳性能和资源利用率，因为它根本不会阻塞。
5.相比之下，await() 是一个阻塞操作。一旦被调用，调用线程会一直阻塞直到操作完成。使用await()实现顺序逻辑很容易，但是调用线程会在I/O操作完成前不必要的阻塞。并且线程间通知的成本相对较高。还有可能发生死锁。因为I/O操作可能永远不会完成。

/*
BAD!!!
*/
public void channelRead(ChannelHandlerContext ctx,Object msg){
    ChannelFuture future = ctx.channel().close();
    //不要在ChannelHandler中使用await()
    future.awaitUninterrupted();
    
}

/**
GOOD!
**/
public void channelRead(ChannelHandlerContext ctx,Object msg){
    ChannelFuture future = ctx.channel().close();
    future.addListener(new ChannelFutureListener(){
        public void operationComplete(Channel Future){
        
        }
    });
}

6.不要在I/O线程中使用await(),否则将会抛出BlockingOperationException 防止死锁。
7.不要混淆 I/O的timeout与await()的timeout。使用await(long)方法设置的超时值与I/O超时无关。如果I/O操作超时，则Future将被标记为“已经完成但是失败”。

/**
*BAD!
***/
Bootstrap b = new Bootstrap();
ChannelFuture f = b.connect(...);
f.awaitUninterrputibly(10,TimeUnit.SECONDS);
if(f.isCancelled()){
    //取消链接尝试
}else if(!f.isSuccess()){
    //可能会抛出空指针异常，因为future可能未完成
    f.cause.printStackTrace();
}else{
    //连接成功
}

/**
    GOOD
*/
Bootstrap b = new Bootstrap();
b.option(ChannelOption.CONNECT_TIMEOUT_MILLS,10000);
ChannelFuture f = b.connect(...);
f.awaitUninterrputibly();

assert f.isDone();

if(f.isCancelled()){
    //取消连接尝试
}else if(!f.isSuccess()){
    f.cause.printStackTrace();
}else{
    //连接成功
}

3.ChannelFutureListener

1.ChannelFutureListener继承了GenericFutureListener<ChannelFuture>接口。

public interface GenericFutureListener<F extends Future<?>> extends EventListener {

  /*******
  *当与Future关联的操作完成时，调用此方法
  ********/
    void operationComplete(F future) throws Exception;
}

2.它用于监听ChannelFuture的结果，通过调用ChannelFuture.addListener(GenericFutureListener)添加监听器，将通知异步ChannelI/O操作的结果。

4.ChannelPromise
1.继承了ChannelFuture接口，是ChannelFuture的升级版。它是可写的，可以设置hannelFuture的状态，例如成功或是失败等。

/**
 DefaultPromise.awaitUninterruptibly()
*/
  public Promise<V> awaitUninterruptibly() {
        //如果已经完成，则立即返回
        if (isDone()) {
            return this;
        }
        //检查死锁
        checkDeadLock();
    
        boolean interrupted = false;
        //使用同步，使修改waiter的线程只有一个
        synchronized (this) {
        //如果未完成一直循环检测
            while (!isDone()) {
                //++waiters
                incWaiters();
                try {
                
                    wait();
                } catch (InterruptedException e) {
                    // Interrupted while waiting.
                    interrupted = true;
                } finally {
                //--waiters
                    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);
    }

private void notifyListeners() {
        EventExecutor executor = executor();
        //如果当前执行的线程是包含在此事件循环组中的那个线程
        if (executor.inEventLoop()) {
            final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
            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();
            }
        });
    }

 private void notifyListeners0(DefaultFutureListeners listeners) {
        GenericFutureListener<?>[] a = listeners.listeners();
        int size = listeners.size();
        for (int i = 0; i < size; i ++) {
        //通知数组中对应的监听器
            notifyListener0(this, a[i]);
        }
    }

 private static void notifyListener0(Future future, GenericFutureListener l) {
        try {
        //调用operationComplete方法
            l.operationComplete(future);
        } catch (Throwable t) {
            if (logger.isWarnEnabled()) {
                logger.warn("An exception was thrown by " + l.getClass().getName() + ".operationComplete()", t);
            }
        }
    }