下面來(lái)具體看一下AbstractUnsafe的主要方法實(shí)現(xiàn)。 AbstractUnsafe的重要實(shí)現(xiàn)register實(shí)現(xiàn) @Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } if (isRegistered()) { //檢查是否已經(jīng)注冊(cè), 避免重復(fù)只需注冊(cè)動(dòng)作。 promise.setFailure(new IllegalStateException("registered to an event loop already")); return; } if (!isCompatible(eventLoop)) {//檢查eventloop是否滿足Channel的要求，由子類實(shí)現(xiàn) promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; } //設(shè)置Channel的EventLoop實(shí)例 AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop()) { //檢查是否在當(dāng)前線程中，如果是，直接調(diào)用 register0(promise); } else { //如果不是，把register0包裝到runnable中放到eventloop中調(diào)用。 try { eventLoop.execute(new Runnable() { @Override public 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); } } } 這個(gè)方法的實(shí)現(xiàn)為我們展示了netty使用I/O線程的一般套路 if(eventLoop.inEventLoop()){ doSomething(); }else{ eventLoop.execute(new Runnable(){ @Override public void run() { doSomething(); } }); } 對(duì)于某個(gè)需要放到I/O線性中執(zhí)行的方法，先檢查當(dāng)前線程是不是I/O線程，是就直接執(zhí)行，不是就把它包裝到Ruannable中放到eventLoop中執(zhí)行。 register的功能總結(jié)一句話就是調(diào)用register0, 下面看看register0的實(shí)現(xiàn)。 private void register0(ChannelPromise promise) { try { // check if the channel is still open as it could be closed in the mean time when the register // call was outside of the eventLoop //確保promise沒(méi)有被取消同時(shí)Channel沒(méi)有被關(guān)閉才能執(zhí)行后面的動(dòng)作 if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; doRegister(); //執(zhí)行真正的register操作，留改子類實(shí)現(xiàn) neverRegistered = false; registered = true; //設(shè)置Channel已經(jīng)處于registed狀態(tài) // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the // user may already fire events through the pipeline in the ChannelFutureListener. //觸發(fā)handlerAdded事件 pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); pipeline.fireChannelRegistered(); //觸發(fā)channelRegistered事件 // Only fire a channelActive if the channel has never been registered. This prevents firing // multiple channel actives if the channel is deregistered and re-registered. if (isActive()) { if (firstRegistration) {//確保Channel只有在第一次register 的時(shí)候被觸發(fā) pipeline.fireChannelActive(); } else if (config().isAutoRead()) { // This channel was registered before and autoRead() is set. This means we need to begin read // again so that we process inbound data. // // See https://github.com/netty/netty/issues/4805 //對(duì)于設(shè)置了autoRead的Channel執(zhí)行beginRead(); beginRead(); } } } catch (Throwable t) { // Close the channel directly to avoid FD leak. closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } } register語(yǔ)義:
bind實(shí)現(xiàn) @Override public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { assertEventLoop(); if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } // See: https://github.com/netty/netty/issues/576 if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) && localAddress instanceof InetSocketAddress && !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() && !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) { // Warn a user about the fact that a non-root user can't receive a // broadcast packet on *nix if the socket is bound on non-wildcard address. logger.warn( "A non-root user can't receive a broadcast packet if the socket " + "is not bound to a wildcard address; binding to a non-wildcard " + "address (" + localAddress + ") anyway as requested."); } //先保存是否active的狀態(tài) boolean wasActive = isActive(); try { doBind(localAddress); //調(diào)用doBind, 需要子類實(shí)現(xiàn)這個(gè)方法完成真正的bind操作 } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { //如果執(zhí)行完doBind后從非active狀態(tài)變成active裝，則觸發(fā)channelActive事件 invokeLater(new Runnable() { @Override public void run() { pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); } bind語(yǔ)義:
disconnect實(shí)現(xiàn) disconnect和bind的實(shí)現(xiàn)類型，不同的是他調(diào)用的是doDisconnect方法，這個(gè)方法同樣是抽象方法需要子類實(shí)現(xiàn)。當(dāng)channel的狀態(tài)從非active變成active狀態(tài)時(shí)，調(diào)用pipeline.fireChannelInactive()觸發(fā)channelInactive事件。 close實(shí)現(xiàn) @Override public final void close(final ChannelPromise promise) { assertEventLoop(); close(promise, CLOSE_CLOSED_CHANNEL_EXCEPTION, CLOSE_CLOSED_CHANNEL_EXCEPTION, false); } private void close(final ChannelPromise promise, final Throwable cause, final ClosedChannelException closeCause, final boolean notify) { if (!promise.setUncancellable()) { return; } if (closeInitiated) { //這段代碼的作用就是防止多次執(zhí)行close操作 if (closeFuture.isDone()) { // Closed already. safeSetSuccess(promise); } else if (!(promise instanceof VoidChannelPromise)) { // Only needed if no VoidChannelPromise. // This means close() was called before so we just register a listener and return closeFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { promise.setSuccess(); } }); } return; } closeInitiated = true; final boolean wasActive = isActive(); final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; //把outboundBuffer置空，在這之后無(wú)法進(jìn)行write或flush操作 this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer. Executor closeExecutor = prepareToClose(); //這個(gè)方法默認(rèn)實(shí)現(xiàn)是return null. 如果有些可以在子類中覆蓋這個(gè)方法添加關(guān)閉前的準(zhǔn)備代 //下面的if..else執(zhí)行的是相同的操作，不同的是如果closeExecutor可以用，就在這個(gè)executor中執(zhí)行，否則在當(dāng)前線程總執(zhí)行 if (closeExecutor != null) { closeExecutor.execute(new Runnable() { @Override public void run() { try { // Execute the close. doClose0(promise); //執(zhí)行close操作 } finally { // Call invokeLater so closeAndDeregister is executed in the EventLoop again! // close完成之后的操作, 在eventLoop中執(zhí)行 invokeLater(new Runnable() { @Override public void run() { if (outboundBuffer != null) { // Fail all the queued messages //對(duì)outboundBuffer中的數(shù)據(jù)進(jìn)行錯(cuò)誤處理 outboundBuffer.failFlushed(cause, notify); outboundBuffer.close(closeCause); } //執(zhí)行deregister操作， 如果channel由active變成非active狀態(tài)就觸發(fā)channelInactive事件 fireChannelInactiveAndDeregister(wasActive); } }); } } }); } else { try { // Close the channel and fail the queued messages in all cases. doClose0(promise); } finally { if (outboundBuffer != null) { // Fail all the queued messages. outboundBuffer.failFlushed(cause, notify); outboundBuffer.close(closeCause); } } if (inFlush0) { //如果正在執(zhí)行flush操作，把deregister操作放在eventLoop中執(zhí)行 invokeLater(new Runnable() { @Override public void run() { fireChannelInactiveAndDeregister(wasActive); } }); } else { fireChannelInactiveAndDeregister(wasActive); } } } private void doClose0(ChannelPromise promise) { try { doClose(); //調(diào)用doClose執(zhí)行真正的close操作，它是一個(gè)抽象方法，需要在子類中實(shí)現(xiàn)。 closeFuture.setClosed(); safeSetSuccess(promise); } catch (Throwable t) { closeFuture.setClosed(); safeSetFailure(promise, t); } } close實(shí)現(xiàn)的代碼雖然比較多，但做的事情比較簡(jiǎn)單:首先執(zhí)行close操作，然后實(shí)現(xiàn)deregister操作，觸發(fā)channelInactive事件。 在close的實(shí)現(xiàn)中，先調(diào)用assertEventLoop方法確保當(dāng)前方法是在eventLoop中執(zhí)行，然后多次使用invokeLater方法吧一系列操作放在放在Runnable中執(zhí)行，這樣做的目的是事為了保證接下來(lái)的操作一定在當(dāng)前操作完成之后才會(huì)執(zhí)行，這一點(diǎn)是有eventLoop來(lái)保證的，eventLoop執(zhí)行Runnable的順序和調(diào)用execute的順序一致，相關(guān)實(shí)現(xiàn)會(huì)在后面eventLoop章節(jié)具體討論。 deregister實(shí)現(xiàn) @Override public final void deregister(final ChannelPromise promise) { assertEventLoop(); deregister(promise, false); } private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) { if (!promise.setUncancellable()) { return; } if (!registered) { //避免多次執(zhí)行deregister操作 safeSetSuccess(promise); return; } // As a user may call deregister() from within any method while doing processing in the ChannelPipeline, // we need to ensure we do the actual deregister operation later. This is needed as for example, // we may be in the ByteToMessageDecoder.callDecode(...) method and so still try to do processing in // the old EventLoop while the user already registered the Channel to a new EventLoop. Without delay, // the deregister operation this could lead to have a handler invoked by different EventLoop and so // threads. // // See: // https://github.com/netty/netty/issues/4435 invokeLater(new Runnable() { @Override public void run() { try { doDeregister(); //執(zhí)行真正的deregister操作，這方法默認(rèn)沒(méi)做任何事情，子類可以根據(jù)需要覆蓋實(shí)現(xiàn) } catch (Throwable t) { logger.warn("Unexpected exception occurred while deregistering a channel.", t); } finally { if (fireChannelInactive) { pipeline.fireChannelInactive(); // 觸發(fā)channelInactive事件 } // Some transports like local and AIO does not allow the deregistration of // an open channel. Their doDeregister() calls close(). Consequently, // close() calls deregister() again - no need to fire channelUnregistered, so check // if it was registered. if (registered) { registered = false; pipeline.fireChannelUnregistered(); //觸發(fā)channelUnregistered事件 } safeSetSuccess(promise); } } }); } 語(yǔ)義:
調(diào)用doDeregister執(zhí)行真正的deregister操作
根據(jù)參數(shù)可能需要觸發(fā)channelInactive事件
觸發(fā)channelUnregistered事件
write實(shí)現(xiàn) @Override public final void write(Object msg, ChannelPromise promise) { assertEventLoop(); ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { //如果outboundBuffer是null, 意味著這個(gè)channel已經(jīng)被close掉了，需要使用promise返回錯(cuò)誤，然后釋放掉msg // If the outboundBuffer is null we know the channel was closed and so // need to fail the future right away. If it is not null the handling of the rest // will be done in flush0() // See https://github.com/netty/netty/issues/2362 safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION); // release message now to prevent resource-leak ReferenceCountUtil.release(msg); return; } int size; try { msg = filterOutboundMessage(msg); //過(guò)濾msg, 默認(rèn)實(shí)現(xiàn)中沒(méi)有做任何操作，把msg原樣返回, 資料可以根據(jù)需要覆蓋實(shí)現(xiàn)。 size = pipeline.estimatorHandle().size(msg); //計(jì)算msg序列化之后的長(zhǎng)度 if (size < 0) { size = 0; } } catch (Throwable t) { safeSetFailure(promise, t); ReferenceCountUtil.release(msg); return; } outboundBuffer.addMessage(msg, size, promise); //把msg放入outboundBuffer中 } write的操作比較簡(jiǎn)單，他只是把消息放到outboundBuffer中，并沒(méi)有做實(shí)際的寫(xiě)操作。 flush實(shí)現(xiàn) @Override public final void flush() { assertEventLoop(); //確保在eventLoop中執(zhí)行 ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { return; } outboundBuffer.addFlush(); //如果outboundBuffer不是null才可以進(jìn)入真正的write階段 flush0(); } protected void flush0() { if (inFlush0) { //確保不被多個(gè)線程同時(shí)執(zhí)行 // Avoid re-entrance return; } final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null || outboundBuffer.isEmpty()) { //確保outboundBuffer有數(shù)據(jù)是才執(zhí)行下面的步驟 return; } inFlush0 = true; // Mark all pending write requests as failure if the channel is inactive. if (!isActive()) { //如果channel不是active狀態(tài)，返回錯(cuò)誤 try { if (isOpen()) { outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true); } else { // Do not trigger channelWritabilityChanged because the channel is closed already. outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } } finally { inFlush0 = false; } return; } try { doWrite(outboundBuffer); //執(zhí)行真正的寫(xiě)操作，這是一個(gè)抽象方法，需要子類實(shí)現(xiàn)。 } catch (Throwable t) { if (t instanceof IOException && config().isAutoClose()) { /** * Just call {@link #close(ChannelPromise, Throwable, boolean)} here which will take care of * failing all flushed messages and also ensure the actual close of the underlying transport * will happen before the promises are notified. * * This is needed as otherwise {@link #isActive()} , {@link #isOpen()} and {@link #isWritable()} * may still return {@code true} even if the channel should be closed as result of the exception. */ //如是I/O異常，并且channel配置允許自動(dòng)關(guān)閉，則關(guān)閉channel close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } else { try { shutdownOutput(voidPromise(), t); //關(guān)閉output通道，不允許執(zhí)行write操作。 } catch (Throwable t2) { close(voidPromise(), t2, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } } } finally { inFlush0 = false; } } 語(yǔ)義:
