javadoc笔记点#
观察者的核心思想就是,在适当的时机回调观察者的指定动作函数
我们知道,在使用netty创建channel时,一般都是把这个channel设置成非阻塞的模式,这意味着什么呢? 意味着所有io操作一经调用,即刻返回
这让netty对io的吞吐量有了飞跃性的提升,但是异步编程相对于传统的串行化的编程模式来说,控制起来可太麻烦了
jdk提供了原生的Futrue接口,意为在未来任务,其实就是把任务封装起来交给新的线程执行,在这个线程执行任务的期间,我们的主线程可以腾出时间去做别的事情
下面的netty给出的实例代码,我们可以看到,任务线程有返回一个Futrue对象,这个对象中封装着任务执行的情况
* * void showSearch(final String target) * * throws InterruptedException { * * Future<String> future * * = executor.submit(new Callable<String>() { * * public String call() { * * return searcher.search(target); * * }}); * * displayOtherThings(); // do other things while searching * * try { * * displayText(future.get()); // use future * * } catch (ExecutionException ex) { cleanup(); return; } * * } *
虽然jdk原生Futrue可以实现异步提交任务,并且返回了任务执行信息的Futrue,但是有一个致命的缺点,从futrue获取任务执行情况方法,是阻塞的,这是不被允许的,因为在netty中,一条channel可能关系着上千的客户端的链接,其中一个客户端的阻塞导致几千的客户端不可用是不被允许的,netty的Future设计成,继承jdk原生的future,而且进行扩展如下
// todo 这个接口继承了 java并发包总的Futrue , 并在其基础上增加了很多方法 // todo Future 表示对未来任务的封装 public interface Future<V> extends java.util.concurrent.Future<V> { // todo 判断IO是否成功返回 boolean isSuccess(); // todo 判断是否是 cancel()方法取消 boolean isCancellable(); // todo 返回IO 操作失败的原因 Throwable cause(); /** * todo 使用了观察者设计模式, 给这个future添加监听器, 一旦Future 完成, listenner 立即被通知 */ Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener); // todo 添加多个listenner Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners); Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener); // todo 移除多个 listenner Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners); // todo sync(同步) 等待着 future 的完成, 并且,一旦future失败了,就会抛出 future 失败的原因 // todo bind()是个异步操作,我们需要同步等待他执行成功 Future<V> sync() throws InterruptedException; // todo 不会被中断的 sync等待 Future<V> syncUninterruptibly(); // todo 等待 Future<V> await() throws InterruptedException; Future<V> awaitUninterruptibly(); // todo 无阻塞的返回Future对象, 如果没有,返回null // todo 有时 future成功执行后返回值为null, 这是null就是成功的标识, 如 Runable就没有返回值, 因此文档建议还要 通过isDone() 判断一下真的完成了吗 V getNow(); @Override boolean cancel(boolean mayInterruptIfRunning); ...
netty的观察者模式#
最常用的关于异步执行的方法writeAndFlush()就是典型的观察者的实现, 在netty中,当一个IO操作刚开始的时候,一个ChannelFutrue对象就会创建出来,此时,这个futrue对象既不是成功的,也不是失败的,更不是被取消的,因为这个IO操作还没有结束
如果我们想在IO操作结束后立刻执行其他的操作时,netty推荐我们使用addListenner()添加监听者的方法而不是使用await()阻塞式等待,使用监听者,我们就不用关系具体什么时候IO操作结束,只需要提供回调方法就可以,当IO操作结束后,方法会自动被回调
在netty中,一个IO操作是状态分为如下几种
* +---------------------------+ * | Completed successfully | * +---------------------------+ * +----> isDone() = true | * +--------- -----------------+ | | isSuccess() = true | * | Uncompleted | | +===========================+ * +--------------------------+ | | Completed with failure | * | isDone() = false | | +---------------------------+ * | isSuccess() = false |----+----> isDone() = true | * | isCancelled() = false | | | cause() = non-null 非空| * | cause() = null | | +===========================+ * +--------------------------+ | | Completed by cancellation | * | +---------------------------+ * +----> isDone() = true | * | isCancelled() = true | * +---------------------------+
源码追踪#
对writeAndFlush的使用
ChannelFuture channelFuture = ctx.writeAndFlush("from client : " + UUID.randomUUID()); channelFuture.addListener(future->{ if(future.isSuccess()){ todo }else{ todo } });
注意点: 我们使用writeAndFlush() 程序立即返回,随后我们使用返回的对象添加监听者,添加回调,这个时writeAndFlush()有可能已经完成了,也有可能没有完成,这是不确定的事
首先我们知道,writeAndFlush()是出站的动作,属于channelOutboundHandler,而且他是从pipeline的尾部开始传播的,源码如下:
@Override public final ChannelFuture writeAndFlush(Object msg) { return tail.writeAndFlush(msg); }
尾节点数据AbstractChannelHandlerContext类, 继续跟进查看源码如下:
@Override public final ChannelFuture writeAndFlush(Object msg) { return tail.writeAndFlush(msg); } @Override public ChannelPromise newPromise() { return new DefaultChannelPromise(channel(), executor()); }
悄无声息的做了一个很重要的事情,创建了Promise,这个DefaultChannelPromise就是被观察者,过一会由它完成方法的回调
继续跟进writeAndFlush() ,源码如下, 我们可以看到promise被返回了, DefaultChannelPromise是ChannelPromise的实现类,而ChannelPromise又继承了ChannelFuture,这也是为什么明明每次使用writeAndFlush()返回的都是ChannelFuture而我们这里却返回了DafaultChannelPromise
// todo 调用本类的 write(msg, true, promise) @Override public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { if (msg == null) { throw new NullPointerException("msg"); } if (isNotValidPromise(promise, true)) { ReferenceCountUtil.release(msg); return promise; } write(msg, true, promise); return promise;
在去目标地之前,先看一下addListenner()干了什么,我们进入到DefaultChannelPromise 源码如下:
@Override public ChannelPromise addListener(GenericFutureListener<? extends Future<? super Void>> listener) { super.addListener(listener); return this; } 随机进入它的父类 DefaultChannelPromise中 @Override public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) { checkNotNull(listener, "listener"); synchronized (this) { addListener0(listener); } if (isDone()) { notifyListeners(); } return this; }
这个函数分两步进行
第一步: 为什么添加监听事件的方法需要同步?#
在这种多线程并发执行的情况下,这个addListener0(listener);任意一个线程都能使用,存在同步添加的情况 这个动作不像将channel和EventLoop做的唯一绑定一样,没有任何必须使用inEventloop()去判断在哪个线程中,直接使用同步
接着进入addListener0(listener)
private void addListener0(GenericFutureListener<? extends Future<? super V>> listener) { if (listeners == null) { listeners = listener; // todo 第一次添加直接在这里赋值 } else if (listeners instanceof DefaultFutureListeners) { // todo 第三次添加调用这里 ((DefaultFutureListeners) listeners).add(listener); } else { // todo 第二次添加来这里复制, 由这个 DefaultFutureListeners 存放观察者 listeners = new DefaultFutureListeners((GenericFutureListener<?>) listeners, listener); } }
第二步: 为什么接着判断isDone()#
writeAndFlush()是异步执行的,而且在我们添加监听者的操作之前已经开始执行了,所以在添加完监听者之后,立即验证一把,有没有成功
思考一波:#
回顾writeAndFlush()的调用顺序,从tail开始传播两波事件,第一波write,紧接着第二波flush,一直传播到header,进入unsafe类中,由他完成把据写入jdk原生ByteBuffer的操作, 所以按理说,我们添加是listenner的回调就是在header的unsafe中完成的,这是我们的目标地
任何方法的回调都是提前设计好了的,就像pipeline中的handler中的方法的回调,就是通过遍历pipeline内部的链表实现的,这里的通知观察者,其实也是调用观察者的方法,而且他使用的一定是观察的父类及以上的引用实现的方法回调
回到我们的writeAndFlush()这个方法,在第二波事务传递完成,将数据真正写入jdk原生的ByteBuffer之前,只有进行的所有回调都是设置失败的状态,直到把数据安全发出后才可能是 回调成功的操作
此外,想要进行回调的操作,就得有被观察的对象的引用,所以一会我就回看到,Promise 一路被传递下去
我们进入的unsafe的write()就可以看到与回调相关的操作safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);,源码如下
@Override public final void write(Object msg, ChannelPromise promise) { assertEventLoop(); ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { // todo 缓存 写进来的 buffer safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION); ReferenceCountUtil.release(msg); return; }
我们继续跟进本类方法safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);, 源码如下:
protected final void safeSetFailure(ChannelPromise promise, Throwable cause) { if (!(promise instanceof VoidChannelPromise) && !promise.tryFailure(cause)) { logger.warn("Failed to mark a promise as failure because it's done already: {}", promise, cause); } }
其中重要的方法,就是回调 被观察者的 tryFailure(cause), 这个被观察者的类型是ChannelPromise, 我们去看它的实现,源码如下
@Override public boolean tryFailure(Throwable cause) { if (setFailure0(cause)) { notifyListeners(); return true; } return false; }
调用本类方法notifyListeners()
继续跟进本类方法notifyListenersNow();
接着跟进本类方法notifyListener0(this, (GenericFutureListener<?>) listeners);
继续l.operationComplete(future); 终于看到了调用了监听者的完成操作,实际上就是回调用户的方法,虽然是完成的,但是失败了
下面我们去flush()中去查看通知成功的回调过程, 方法的调用顺序如下
flush(); flush0(); doWrite(outboundBuffer);
在doWrite()方法中,就会使用自旋的方式往尝试把数据写出去, 数据被写出去后,有一个标识 done=true, 证明是成功写出了, 紧接着就是把当前的盛放ByteBuf的entry从链表上移除,源码出下
if (done) { // todo 跟进去 in.remove(); } else {
我们继续跟进remove(), 终于我们找到了成功回调的标志,在remove()的底端safeSuccess(promise);, 下一步就是用回调用户添加的监听者操作完成了,并且完成的状态是Success成功的
public boolean remove() { // todo 获取当前的 Entry Entry e = flushedEntry; if (e == null) { clearNioBuffers(); return false; } Object msg = e.msg; ChannelPromise promise = e.promise; int size = e.pendingSize; // todo 将当前的Entry进行移除 removeEntry(e); if (!e.cancelled) { // only release message, notify and decrement if it was not canceled before. ReferenceCountUtil.safeRelease(msg); safeSuccess(promise); decrementPendingOutboundBytes(size, false, true); }
