吃透Netty源码系列五十八之Unsafe和AbstractUnsafe

这个Unsafe可不是JDK原生的Unsafe哦&＃xff0c;主要就是一些直接跟IO底层直接相关的通用操作&＃xff1a;

interface Unsafe {// 接受数据的时候用于分配字节缓冲区的处理器RecvByteBufAllocator.Handle recvBufAllocHandle();// 本地地址SocketAddress localAddress();// 远程地址SocketAddress remoteAddress();//向事件循环注册通道&＃xff0c;完成后回调void register(EventLoop eventLoop, ChannelPromise promise);// 绑定本地地址&＃xff0c;完成后回调void bind(SocketAddress localAddress, ChannelPromise promise);// 连接void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);// 断开连接&＃xff0c;完成回调void disconnect(ChannelPromise promise);// 关闭连接&＃xff0c;完成回调void close(ChannelPromise promise);// 立即关闭&＃xff0c;不触发任何事件void closeForcibly();// 注销&＃xff0c;完成回调void deregister(ChannelPromise promise);// 开始读操作void beginRead();// 写操作void write(Object msg, ChannelPromise promise);// 冲刷所有的出站数据void flush();// 特殊的占位符&＃xff0c;不接受通知ChannelPromise voidPromise();//写操作的出站缓冲区ChannelOutboundBuffer outboundBuffer();}


AbstractUnsafe基本抽象实现

属性

这些就是一些基本的属性&＃xff0c;要进行数据的读写&＃xff0c;需要有接收缓冲区&＃xff0c;所以有了recvHandle处理器&＃xff0c;写出去的时候需要有写缓冲区ChannelOutboundBuffer &＃xff0c;注意ChannelOutboundBuffer是初始化的时候就会创建&＃xff0c;就创建一次。

//出站字节缓冲区private volatile ChannelOutboundBuffer outboundBuffer &＃61; new ChannelOutboundBuffer(AbstractChannel.this);private RecvByteBufAllocator.Handle recvHandle;//接受数据缓冲分配器的处理器private boolean inFlush0;//是否正在缓冲/** true if the channel has never been registered, false otherwise */private boolean neverRegistered &＃61; true;//通道没注册过private void assertEventLoop() {//断言还没注册&＃xff0c;或者当前线程是IO线程assert !registered || eventLoop.inEventLoop();}


recvBufAllocHandle接受缓冲区处理器

缓冲区分配上次说过了&＃xff0c;出站缓冲区以前文章也有讲过。

&＃64;Overridepublic RecvByteBufAllocator.Handle recvBufAllocHandle() {if (recvHandle &＃61;&＃61; null) {recvHandle &＃61; config().getRecvByteBufAllocator().newHandle();}return recvHandle;}

register注册到事件循环

注册方法其实就是判断是否当前线程就是IO线程&＃xff0c;是的话就直接执行&＃xff0c;不是就包装成一个任务提交给IO线程&＃xff0c;这样就避免多线程的问题&＃xff0c;始终是单线程操作。

&＃64;Overridepublic final void register(EventLoop eventLoop, final ChannelPromise promise) {ObjectUtil.checkNotNull(eventLoop, "eventLoop");if (isRegistered()) {//是否已经注册人到一个eventLooppromise.setFailure(new IllegalStateException("registered to an event loop already"));return;}if (!isCompatible(eventLoop)) {//是否是NioEventLoop类型promise.setFailure(new IllegalStateException("incompatible event loop type: " &＃43; eventLoop.getClass().getName()));return;}AbstractChannel.this.eventLoop &＃61; eventLoop;
//只能当前线程是eventLoop的线程才可以注册&＃xff0c;防止多线程并发问题&＃xff0c;所以即使多线程来操作&＃xff0c;也是安全的&＃xff0c;会按照一定顺序提交到任务队列里if (eventLoop.inEventLoop()) {register0(promise);} else {//否则就当做任务提交给eventLoop的任务队列try {eventLoop.execute(new Runnable() {&＃64;Overridepublic void run() {register0(promise);}});} catch (Throwable t) {logger.warn("Force-closing a channel whose registration task was not accepted by an event loop: {}",AbstractChannel.this, t);closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}}


register0执行注册逻辑

这里是注册过程要做的事&＃xff0c;进行真正的注册逻辑doRegister&＃xff0c;其实就是将NIO通道注册到Selector上&＃xff0c;然后进行处理器的待添加事件的处理&＃xff0c;注册回调成功&＃xff0c;管道传递注册事件&＃xff0c;如果是第一次注册&＃xff0c;管道传递通道激活事件&＃xff0c;否则是设置自动读的话就注册读监听。

private void register0(ChannelPromise promise) {try {if (!promise.setUncancellable() || !ensureOpen(promise)) {//确保是不可取消和通道打开着&＃xff0c;否则就返回return;}boolean firstRegistration &＃61; neverRegistered;//设置注册标记doRegister();//进行注册逻辑neverRegistered &＃61; false;//AbstractUnsafe的已注册标记registered &＃61; true;//channel的已注册标记pipeline.invokeHandlerAddedIfNeeded();//如果在注册前有处理器添加&＃xff0c;还没进行HandlerAdded回调&＃xff0c;注册成功后要回调safeSetSuccess(promise);//回调注册成功pipeline.fireChannelRegistered();//通道注册事件传递if (isActive()) {//通道激活的话if (firstRegistration) {//第一次注册要进行激活事件传递pipeline.fireChannelActive();} else if (config().isAutoRead()) {//否则如果设置了自动读&＃xff0c;就进行读监听beginRead();}}} catch (Throwable t) {closeForcibly();//强制关闭closeFuture.setClosed();//关闭回调safeSetFailure(promise, t);//设置失败}}


bind绑定地址

省略了部分。看主逻辑&＃xff0c;做具体的doBind&＃xff0c;如果通道开始没激活&＃xff0c;绑定后激活的话&＃xff0c;就开一个延时的任务&＃xff0c;进行激活事件传递&＃xff0c;最后回调绑定成功。

&＃64;Overridepublic final void bind(final SocketAddress localAddress, final ChannelPromise promise) {...boolean wasActive &＃61; isActive();try {doBind(localAddress);} catch (Throwable t) {safeSetFailure(promise, t);closeIfClosed();return;}if (!wasActive && isActive()) {//绑定前没激活&＃xff0c;绑定后激活了invokeLater(new Runnable() {&＃64;Overridepublic void run() {pipeline.fireChannelActive();}});}safeSetSuccess(promise);}


disconnect断开连接

调用doDisconnect&＃xff0c;断开连接&＃xff0c;如果开始激活的&＃xff0c;断开后失效了&＃xff0c;就传递失效事件。如果通道关闭了&＃xff0c;还要处理关闭事件closeIfClosed。

&＃64;Overridepublic final void disconnect(final ChannelPromise promise) {assertEventLoop();if (!promise.setUncancellable()) {return;}boolean wasActive &＃61; isActive();try {doDisconnect();// Reset remoteAddress and localAddressremoteAddress &＃61; null;localAddress &＃61; null;} catch (Throwable t) {safeSetFailure(promise, t);closeIfClosed();return;}if (wasActive && !isActive()) {invokeLater(new Runnable() {&＃64;Overridepublic void run() {pipeline.fireChannelInactive();}});}safeSetSuccess(promise);closeIfClosed(); // doDisconnect() might have closed the channel}


close关闭通道和出站缓冲区

进行通道的关闭&＃xff0c;主要还是出站缓冲区的处理和传递通道失效和注销事件。

&＃64;Overridepublic final void close(final ChannelPromise promise) {assertEventLoop();ClosedChannelException closedChannelException &＃61; new ClosedChannelException();close(promise, closedChannelException, closedChannelException, false);}private void close(final ChannelPromise promise, final Throwable cause,final ClosedChannelException closeCause, final boolean notify) {if (!promise.setUncancellable()) {return;}if (closeInitiated) {//如果已经发起关闭了if (closeFuture.isDone()) {//判断是否关闭完成// Closed already.safeSetSuccess(promise);//回调} else if (!(promise instanceof VoidChannelPromise)) { closeFuture.addListener(new ChannelFutureListener() {//如果不是VoidChannelPromise&＃xff0c;添加关闭监听&＃64;Overridepublic void operationComplete(ChannelFuture future) throws Exception {promise.setSuccess();}});}return;}closeInitiated &＃61; true;//已经开始关闭了//处理出站缓冲区关闭final boolean wasActive &＃61; isActive();final ChannelOutboundBuffer outboundBuffer &＃61; this.outboundBuffer;this.outboundBuffer &＃61; null; // Disallow adding any messages and flushes to outboundBuffer.Executor closeExecutor &＃61; prepareToClose();if (closeExecutor !&＃61; null) {closeExecutor.execute(new Runnable() {&＃64;Overridepublic void run() {try {// Execute the close.doClose0(promise);} finally {invokeLater(new Runnable() {&＃64;Overridepublic void run() {if (outboundBuffer !&＃61; null) {// Fail all the queued messagesoutboundBuffer.failFlushed(cause, notify);outboundBuffer.close(closeCause);}fireChannelInactiveAndDeregister(wasActive);}});}}});} else {try {doClose0(promise);} finally {if (outboundBuffer !&＃61; null) {// Fail all the queued messages.outboundBuffer.failFlushed(cause, notify);outboundBuffer.close(closeCause);}}if (inFlush0) {invokeLater(new Runnable() {&＃64;Overridepublic void run() {fireChannelInactiveAndDeregister(wasActive);}});} else {fireChannelInactiveAndDeregister(wasActive);}}}


doClose0关闭通道

具体的关闭逻辑和回调&＃xff0c;具体逻辑是在通道中实现的&＃xff0c;后面会讲。

private void doClose0(ChannelPromise promise) {try {doClose();closeFuture.setClosed();safeSetSuccess(promise);} catch (Throwable t) {closeFuture.setClosed();safeSetFailure(promise, t);}}


fireChannelInactiveAndDeregister传递通道失效和注销事件

传递通道失效和注销事件。

private void fireChannelInactiveAndDeregister(final boolean wasActive) {deregister(voidPromise(), wasActive && !isActive());}


doDeregister注销事件

提交一个任务&＃xff0c;进行注销doDeregister&＃xff0c;然后根据情况传递通道失效和注销事件。

private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {if (!promise.setUncancellable()) {return;}if (!registered) {safeSetSuccess(promise);return;}invokeLater(new Runnable() {&＃64;Overridepublic void run() {try {doDeregister();} catch (Throwable t) {logger.warn("Unexpected exception occurred while deregistering a channel.", t);} finally {if (fireChannelInactive) {pipeline.fireChannelInactive();}if (registered) {registered &＃61; false;pipeline.fireChannelUnregistered();}safeSetSuccess(promise);}}});}


shutdownOutput出站缓冲区关闭处理

清理出站缓冲区ChannelOutboundBuffer &＃xff0c;并传递fireUserEventTriggered事件。

&＃64;UnstableApipublic final void shutdownOutput(final ChannelPromise promise) {assertEventLoop();shutdownOutput(promise, null);}private void shutdownOutput(final ChannelPromise promise, Throwable cause) {if (!promise.setUncancellable()) {return;}final ChannelOutboundBuffer outboundBuffer &＃61; this.outboundBuffer;if (outboundBuffer &＃61;&＃61; null) {//如果出站缓冲区为null的话&＃xff0c;就回调失败promise.setFailure(new ClosedChannelException());return;}this.outboundBuffer &＃61; null; // Disallow adding any messages and flushes to outboundBuffer.禁止添加数据到出站缓冲区了final Throwable shutdownCause &＃61; cause &＃61;&＃61; null ?//根据异常创建ChannelOutputShutdownExceptionnew ChannelOutputShutdownException("Channel output shutdown") :new ChannelOutputShutdownException("Channel output shutdown", cause);Executor closeExecutor &＃61; prepareToClose();//有关闭执行器if (closeExecutor !&＃61; null) {//提交一个任务closeExecutor.execute(new Runnable() {&＃64;Overridepublic void run() {try {// Execute the shutdown.doShutdownOutput();promise.setSuccess();} catch (Throwable err) {promise.setFailure(err);} finally {//出站缓冲区事件任务// Dispatch to the EventLoopeventLoop().execute(new Runnable() {&＃64;Overridepublic void run() {//出站缓冲区事件处理closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);}});}}});} else {try {//直接处理关闭// Execute the shutdown.doShutdownOutput();promise.setSuccess();} catch (Throwable err) {promise.setFailure(err);} finally {closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);}}}private void closeOutboundBufferForShutdown(ChannelPipeline pipeline, ChannelOutboundBuffer buffer, Throwable cause) {buffer.failFlushed(cause, false);//不能冲刷buffer.close(cause, true);//关闭出站缓冲区pipeline.fireUserEventTriggered(ChannelOutputShutdownEvent.INSTANCE);//传递事件}

好像有点长了&＃xff0c;下一篇继续吧。

好了&＃xff0c;今天就到这里了&＃xff0c;希望对学习理解有帮助&＃xff0c;大神看见勿喷&＃xff0c;仅为自己的学习理解&＃xff0c;能力有限&＃xff0c;请多包涵。