下面來具體看一下AbstractUnsafe的主要方法實現。 AbstractUnsafe的重要實現register實現 @Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } if (isRegistered()) { //檢查是否已經注冊, 避免重復只需注冊動作。 promise.setFailure(new IllegalStateException("registered to an event loop already")); return; } if (!isCompatible(eventLoop)) {//檢查eventloop是否滿足Channel的要求，由子類實現 promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; } //設置Channel的EventLoop實例 AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop()) { //檢查是否在當前線程中，如果是，直接調用 register0(promise); } else { //如果不是，把register0包裝到runnable中放到eventloop中調用。 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); } } } 這個方法的實現為我們展示了netty使用I/O線程的一般套路 if(eventLoop.inEventLoop()){ doSomething(); }else{ eventLoop.execute(new Runnable(){ @Override public void run() { doSomething(); } }); } 對于某個需要放到I/O線性中執行的方法，先檢查當前線程是不是I/O線程，是就直接執行，不是就把它包裝到Ruannable中放到eventLoop中執行。 register的功能總結一句話就是調用register0, 下面看看register0的實現。 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沒有被取消同時Channel沒有被關閉才能執行后面的動作 if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; doRegister(); //執行真正的register操作，留改子類實現 neverRegistered = false; registered = true; //設置Channel已經處于registed狀態 // 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. //觸發handlerAdded事件 pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); pipeline.fireChannelRegistered(); //觸發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 的時候被觸發 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 //對于設置了autoRead的Channel執行beginRead(); beginRead(); } } } catch (Throwable t) { // Close the channel directly to avoid FD leak. closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } } register語義:
bind實現 @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的狀態 boolean wasActive = isActive(); try { doBind(localAddress); //調用doBind, 需要子類實現這個方法完成真正的bind操作 } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive && isActive()) { //如果執行完doBind后從非active狀態變成active裝，則觸發channelActive事件 invokeLater(new Runnable() { @Override public void run() { pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); } bind語義:
調用抽象方法doBind, 它需要子類實現。
如果channel的狀態從非active變成active狀態，則觸發channelActive事件
disconnect實現 disconnect和bind的實現類型，不同的是他調用的是doDisconnect方法，這個方法同樣是抽象方法需要子類實現。當channel的狀態從非active變成active狀態時，調用pipeline.fireChannelInactive()觸發channelInactive事件。 close實現 @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) { //這段代碼的作用就是防止多次執行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置空，在這之后無法進行write或flush操作 this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer. Executor closeExecutor = prepareToClose(); //這個方法默認實現是return null. 如果有些可以在子類中覆蓋這個方法添加關閉前的準備代 //下面的if..else執行的是相同的操作，不同的是如果closeExecutor可以用，就在這個executor中執行，否則在當前線程總執行 if (closeExecutor != null) { closeExecutor.execute(new Runnable() { @Override public void run() { try { // Execute the close. doClose0(promise); //執行close操作 } finally { // Call invokeLater so closeAndDeregister is executed in the EventLoop again! // close完成之后的操作, 在eventLoop中執行 invokeLater(new Runnable() { @Override public void run() { if (outboundBuffer != null) { // Fail all the queued messages //對outboundBuffer中的數據進行錯誤處理 outboundBuffer.failFlushed(cause, notify); outboundBuffer.close(closeCause); } //執行deregister操作， 如果channel由active變成非active狀態就觸發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) { //如果正在執行flush操作，把deregister操作放在eventLoop中執行 invokeLater(new Runnable() { @Override public void run() { fireChannelInactiveAndDeregister(wasActive); } }); } else { fireChannelInactiveAndDeregister(wasActive); } } } private void doClose0(ChannelPromise promise) { try { doClose(); //調用doClose執行真正的close操作，它是一個抽象方法，需要在子類中實現。 closeFuture.setClosed(); safeSetSuccess(promise); } catch (Throwable t) { closeFuture.setClosed(); safeSetFailure(promise, t); } } close實現的代碼雖然比較多，但做的事情比較簡單:首先執行close操作，然后實現deregister操作，觸發channelInactive事件。 在close的實現中，先調用assertEventLoop方法確保當前方法是在eventLoop中執行，然后多次使用invokeLater方法吧一系列操作放在放在Runnable中執行，這樣做的目的是事為了保證接下來的操作一定在當前操作完成之后才會執行，這一點是有eventLoop來保證的，eventLoop執行Runnable的順序和調用execute的順序一致，相關實現會在后面eventLoop章節具體討論。 deregister實現 @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) { //避免多次執行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(); //執行真正的deregister操作，這方法默認沒做任何事情，子類可以根據需要覆蓋實現 } catch (Throwable t) { logger.warn("Unexpected exception occurred while deregistering a channel.", t); } finally { if (fireChannelInactive) { pipeline.fireChannelInactive(); // 觸發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(); //觸發channelUnregistered事件 } safeSetSuccess(promise); } } }); } 語義:
調用doDeregister執行真正的deregister操作
根據參數可能需要觸發channelInactive事件
觸發channelUnregistered事件
write實現 @Override public final void write(Object msg, ChannelPromise promise) { assertEventLoop(); ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { //如果outboundBuffer是null, 意味著這個channel已經被close掉了，需要使用promise返回錯誤，然后釋放掉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); //過濾msg, 默認實現中沒有做任何操作，把msg原樣返回, 資料可以根據需要覆蓋實現。 size = pipeline.estimatorHandle().size(msg); //計算msg序列化之后的長度 if (size < 0) { size = 0; } } catch (Throwable t) { safeSetFailure(promise, t); ReferenceCountUtil.release(msg); return; } outboundBuffer.addMessage(msg, size, promise); //把msg放入outboundBuffer中 } write的操作比較簡單，他只是把消息放到outboundBuffer中，并沒有做實際的寫操作。 flush實現 @Override public final void flush() { assertEventLoop(); //確保在eventLoop中執行 ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null) { return; } outboundBuffer.addFlush(); //如果outboundBuffer不是null才可以進入真正的write階段 flush0(); } protected void flush0() { if (inFlush0) { //確保不被多個線程同時執行 // Avoid re-entrance return; } final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; if (outboundBuffer == null || outboundBuffer.isEmpty()) { //確保outboundBuffer有數據是才執行下面的步驟 return; } inFlush0 = true; // Mark all pending write requests as failure if the channel is inactive. if (!isActive()) { //如果channel不是active狀態，返回錯誤 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); //執行真正的寫操作，這是一個抽象方法，需要子類實現。 } 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配置允許自動關閉，則關閉channel close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } else { try { shutdownOutput(voidPromise(), t); //關閉output通道，不允許執行write操作。 } catch (Throwable t2) { close(voidPromise(), t2, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false); } } } finally { inFlush0 = false; } } 語義:
