Netty异步读写操作的架构与观察者模式的应用
huaye00 2019-04-22
1.Netty中的Future
Netty中的Future继承了java.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操作的结果或者状态的信息。
2.
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的升级版。它是可写的，可以设置ChannelFuture的状态，例如成功或是失败等。
/**
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);
}
}
}
