6.2 Channel實現

![Netty_Channel類圖][2]

Channel的類圖比較清晰。咱們主要分析NioSocketChannel和NioServerSocketChannel這兩條線。

6.2.1 AbstractChannel

首先看其中的字段：

    private final Channel parent;   // 父Channel
    private final Unsafe unsafe;    
    private final DefaultChannelPipeline pipeline;  // 處理通道
    private final ChannelFuture succeededFuture = new SucceededChannelFuture(this, null);
    private final VoidChannelPromise voidPromise = new VoidChannelPromise(this, true);
    private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
    private final CloseFuture closeFuture = new CloseFuture(this);

    private volatile SocketAddress localAddress;    // 本地地址
    private volatile SocketAddress remoteAddress;   // 遠端地址
    private volatile EventLoop eventLoop;   // EventLoop線程
    private volatile boolean registered;    // 是否註冊到EventLoop

而後，咱們看其中的構造方法：

    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }

newUnsafe()和newChannelPipeline()可由子類覆蓋實現。在Netty的實現中每個Channel都有一個對應的Unsafe內部類：AbstractChannel--AbstractUnsafe，AbstractNioChannel--AbstractNioUnsafe等等，newUnsafe()方法正好用來生成這樣的對應關係。ChannelPipeline將在以後講解，這裏先了解它的功能：做爲用戶處理器Handler的容器爲用戶提供自定義處理I/O事件的能力即爲用戶提供業務邏輯處理。AbstractChannel中對I/O事件的處理，都委託給ChannelPipeline處理，代碼都一模一樣：

    public ChannelFuture bind(SocketAddress localAddress) {
        return pipeline.bind(localAddress);
    }

AbstractChannel其餘方法都比較簡單，主要關注狀態斷定的方法：

    public boolean isRegistered() {
        return registered;
    }

    public boolean isWritable() {
        ChannelOutboundBuffer buf = unsafe.outboundBuffer();
        return buf != null && buf.isWritable(); // 寫緩衝區不爲null且可寫
    }

對於Channel的實現來講，其中的內部類Unsafe纔是關鍵，由於其中含有I/O事件處理的細節。AbstractUnsafe做爲AbstractChannel的內部類，定義了I/O事件處理的基本框架，其中的細節留給子類實現。咱們將依次對各個事件框架進行分析。

register事件框架

    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        if (isRegistered()) {
            promise.setFailure(...);    // 已經註冊則失敗
            return;
        }
        if (!isCompatible(eventLoop)) { // EventLoop不兼容當前Channel
            promise.setFailure(...);
            return;
        }
        AbstractChannel.this.eventLoop = eventLoop;
        // 當前線程爲EventLoop線程直接執行；不然提交任務給EventLoop線程
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try {
                eventLoop.execute(() -> { register0(promise); });
            } catch (Throwable t) {
                closeForcibly();    // 異常時關閉Channel
                closeFuture.setClosed();    
                safeSetFailure(promise, t);
            }
        }
    }

12-22行相似的代碼結構，Netty使用了不少次，這是爲了保證I/O事件以及用戶定義的I/O事件處理邏輯（業務邏輯）在一個線程中處理。咱們看提交的任務register0()：

    private void register0(ChannelPromise promise) {
        try {
            // 確保Channel沒有關閉
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            boolean firstRegistration = neverRegistered;
            doRegister();   // 模板方法，細節由子類完成
            neverRegistered = false;
            registered = true;
            pipeline.invokeHandlerAddedIfNeeded();  // 將用戶Handler添加到ChannelPipeline
            safeSetSuccess(promise);
            pipeline.fireChannelRegistered();   // 觸發Channel註冊事件
            if (isActive()) {
                // ServerSocketChannel接受的Channel此時已被激活
                if (firstRegistration) {
                    // 首次註冊且激活觸發Channel激活事件
                    pipeline.fireChannelActive();   
                } else if (config().isAutoRead()) {
                    beginRead();   // 可視爲模板方法 
                }
            }
        } catch (Throwable t) {
            closeForcibly();     // 可視爲模板方法
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

register0()方法定義了註冊到EventLoop的總體框架，整個流程以下：
(1).註冊的具體細節由doRegister()方法完成，子類中實現。
(2).註冊後將處理業務邏輯的用戶Handler添加到ChannelPipeline。
(3).異步結果設置爲成功，觸發Channel的Registered事件。
(4).對於服務端接受的客戶端鏈接，若是首次註冊，觸發Channel的Active事件，若是已設置autoRead，則調用beginRead()開始讀取數據。
對於(4)的是由於fireChannelActive()中也根據autoRead配置，調用了beginRead()方法。beginRead()方法其實也是一個框架，細節由doBeginRead()方法在子類中實現：

    public final void beginRead() {
        assertEventLoop();
        if (!isActive()) {
            return;
        }
        try {
            doBeginRead();
        } catch (final Exception e) {
            invokeLater(() -> { pipeline.fireExceptionCaught(e); });
            close(voidPromise());
        }
    }

異常處理的closeForcibly()方法也是一個框架，細節由doClose()方法在子類中實現：

    public final void closeForcibly() {
        assertEventLoop();
        try {
            doClose();
        } catch (Exception e) {
            logger.warn("Failed to close a channel.", e);
        }
    }

register框架中有一對safeSetXXX()方法，將未完成的Promise標記爲完成且成功或失敗，其實現以下：

    protected final void safeSetSuccess(ChannelPromise promise) {
        if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
            logger.warn(...);
        }
    }

至此，register事件框架分析完畢。

bind事件框架

    public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
        assertEventLoop();
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return; // 確保Channel沒有關閉
        }
        boolean wasActive = isActive();
        try {
            doBind(localAddress);   // 模板方法，細節由子類完成
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
        if (!wasActive && isActive()) { 
            invokeLater(() -> { pipeline.fireChannelActive(); });   // 觸發Active事件
        }
        safeSetSuccess(promise);
    }

bind事件框架較爲簡單，主要完成在Channel綁定完成後觸發Channel的Active事件。其中的invokeLater()方法向Channel註冊到的EventLoop提交一個任務：

    private void invokeLater(Runnable task) {
        try {
            eventLoop().execute(task);
        } catch (RejectedExecutionException e) {
            logger.warn("Can't invoke task later as EventLoop rejected it", e);
        }
    }

closeIfClosed()方法當Channel再也不打開時關閉Channel，代碼以下：

    protected final void closeIfClosed() {
        if (isOpen()) {
            return;
        }
        close(voidPromise());
    }

close()也是一個框架，以後會進行分析。

disconnect事件框架

    public final void disconnect(final ChannelPromise promise) {
        assertEventLoop();
        if (!promise.setUncancellable()) {
            return;
        }
        boolean wasActive = isActive();
        try {
            doDisconnect(); // 模板方法，細節由子類實現
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            closeIfClosed();
            return;
        }
        if (wasActive && !isActive()) {
            invokeLater(() ->{ pipeline.fireChannelInactive(); });  // 觸發Inactive事件
        }
        safeSetSuccess(promise);
        closeIfClosed(); // disconnect框架可能會調用close框架
    }

close事件框架

    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;
        }
        final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;   
        if (outboundBuffer == null) {   // outboundBuffer做爲一個標記，爲空表示Channel正在關閉
            if (!(promise instanceof VoidChannelPromise)) {
                // 當Channel關閉時，將這次close異步請求結果也設置爲成功
                closeFuture.addListener( (future) -> { promise.setSuccess(); });
            }
            return;
        }
        if (closeFuture.isDone()) {
            safeSetSuccess(promise);    // 已經關閉，保證底層close只執行一次
            return;
        }
        final boolean wasActive = isActive();
        this.outboundBuffer = null; // 設置爲空禁止write操做，同時做爲標記字段表示正在關閉
        Executor closeExecutor = prepareToClose();
        if (closeExecutor != null) {
            closeExecutor.execute(() -> {
                try {
                    doClose0(promise);  // prepareToClose返回的executor執行
                } finally {
                    invokeLater( () -> { // Channel註冊的EventLoop執行
                        // 寫緩衝隊列中的數據所有設置失敗
                        outboundBuffer.failFlushed(cause, notify);
                        outboundBuffer.close(closeCause);
                        fireChannelInactiveAndDeregister(wasActive);
                    });
                }
            });
        } else {    // 當前調用線程執行
            try {
                doClose0(promise);
            } finally {
                outboundBuffer.failFlushed(cause, notify);
                outboundBuffer.close(closeCause);
            }
            if (inFlush0) {
                invokeLater( () -> { fireChannelInactiveAndDeregister(wasActive); });
            } else {
                fireChannelInactiveAndDeregister(wasActive);
            }
        }
    }
    
    private void doClose0(ChannelPromise promise) {
        try {
            doClose();  // 模板方法，細節由子類實現
            closeFuture.setClosed();
            safeSetSuccess(promise);
        } catch (Throwable t) {
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

close事件框架保證只有一個線程執行了真正關閉的doClose()方法，prepareToClose()作一些關閉前的清除工做並返回一個Executor，若是不爲空，須要在該Executor裏執行doClose0()方法；爲空，則在當前線程執行（爲何這樣設計？）。寫緩衝區outboundBuffer同時也做爲一個標記字段，爲空表示Channel正在關閉此時禁止寫操做。fireChannelInactiveAndDeregister()方法須要invokeLater()使用EventLoop執行，是由於其中會調用deRegister()方法觸發Inactive事件，而事件執行須要在EventLoop中執行。

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

deregister事件框架

    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) {
            safeSetSuccess(promise);    // 已經deregister
            return;
        }
        invokeLater( () -> {
            try {
                doDeregister(); // 模板方法，子類實現具體細節
            } catch (Throwable t) {
                logger.warn(...);
            } finally {
                if (fireChannelInactive) {
                    pipeline.fireChannelInactive(); // 根據參數觸發Inactive事件
                }
                if (registered) {
                    registered = false;
                    pipeline.fireChannelUnregistered(); // 首次調用觸發Unregistered事件
                }
                safeSetSuccess(promise);
            }
        });
    }

deregister事件框架的處理流程很清晰，其中，使用invokeLater()方法是由於：用戶可能會在ChannlePipeline中將當前Channel註冊到新的EventLoop，確保ChannelPipiline事件和doDeregister()在同一個EventLoop完成（[?][3]）。

須要注意的是：事件之間可能相互調用，好比：disconnect->close->deregister。

write事件框架

    public final void write(Object msg, ChannelPromise promise) {
        assertEventLoop();
        ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null) {
            // 聯繫close操做，outboundBuffer爲空表示Channel正在關閉，禁止寫數據
            safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);
            ReferenceCountUtil.release(msg); // 釋放msg 防止泄露
            return;
        }
        int size;
        try {
            msg = filterOutboundMessage(msg);
            size = pipeline.estimatorHandle().size(msg);
            if (size < 0) {
                size = 0;
            }
        } catch (Throwable t) {
            safeSetFailure(promise, t);
            ReferenceCountUtil.release(msg);
            return;
        }
        outboundBuffer.addMessage(msg, size, promise);
    }

事實上，這是Netty定義的write操做的所有代碼，完成的功能是將要寫的消息Msg加入到寫緩衝區。其中的filterOutboundMessage()可對消息進行過濾整理，例如把HeapBuffer轉爲DirectBuffer，具體實現由子類負責：

    protected Object filterOutboundMessage(Object msg) throws Exception {
        return msg; // 默認實現
    }

flush事件框架

    public final void flush() {
        assertEventLoop();
        ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null) {
            return; // Channel正在關閉直接返回
        }
        outboundBuffer.addFlush();  // 添加一個標記
        flush0();
    }

    protected void flush0() {
        if (inFlush0) {
            return;     // 正在flush返回防止屢次調用
        }
        final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
        if (outboundBuffer == null || outboundBuffer.isEmpty()) {
            return; // Channel正在關閉或者已沒有須要寫的數據
        }
        inFlush0 = true;
        if (!isActive()) {
            // Channel已經非激活，將全部進行中的寫請求標記爲失敗
            try {
                if (isOpen()) {
                    outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true);
                } else {
                    outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
                }
            } finally {
                inFlush0 = false;
            }
            return;
        }
        try {
            doWrite(outboundBuffer);    // 模板方法，細節由子類實現
        } catch (Throwable t) {
            if (t instanceof IOException && config().isAutoClose()) {
                close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
            } else {
                outboundBuffer.failFlushed(t, true);
            }
        } finally {
            inFlush0 = false;
        }
    }

flush事件中執行真正的底層寫操做，Netty對於寫的處理引入了一個寫緩衝區ChannelOutboundBuffer，由該緩衝區控制Channel的可寫狀態，其具體實現，將會在緩衝區一章中分析。

至此，Unsafe中的事件方法已經分析完7個，但還有connect和read沒有引入，下一節將進行分析。

6.2.2 AbstractNioChannel

Netty的實現中，Unsafe的I/O事件框架中的細節實現方法doXXX()放到了Channel子類中而不是Unsafe子類中，因此咱們先分析Unsafe，而後分析Channel。
AbstractNioChannel從名字能夠看出是對NIO的抽象，咱們自頂向下一步一步深刻細節，該類中定義了一個NioUnsafe接口：

    public interface NioUnsafe extends Unsafe {
        SelectableChannel ch(); // 對應NIO中的JDK實現的Channel
        void finishConnect();   // 鏈接完成
        void read();    // 從JDK的Channel中讀取數據
        void forceFlush(); 
    }

回憶NIO的三大概念：Channel、Buffer、Selector，Netty的Channel包裝了JDK的Channel從而實現更爲複雜的功能。Unsafe中可使用ch()方法，NioChannel中可使用javaChannel()方法得到JDK的Channel。接口中定義了finishConnect()方法是由於：SelectableChannel設置爲非阻塞模式時，connect()方法會當即返回，此時鏈接操做可能沒有完成，若是沒有完成，則須要調用JDK的finishConnect()方法完成鏈接操做。也許你已經注意到，AbstractUnsafe中並無connect事件框架，這是由於並非全部鏈接都有標準的connect過程，好比Netty的LocalChannel和EmbeddedChannel。可是NIO中的鏈接操做則有較爲標準的流程，在介紹Connect事件框架前，先介紹一下其中使用到的相關字段，這些字段定義在AbstractNioChannel中：

    private ChannelPromise connectPromise;  // 鏈接異步結果
    private ScheduledFuture<?> connectTimeoutFuture;    // 鏈接超時檢測任務異步結果
    private SocketAddress requestedRemoteAddress;   // 鏈接的遠端地址

Connect事件框架：

    public final void connect(final SocketAddress remoteAddress, final SocketAddress localAddress,
                                            final ChannelPromise promise) {
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return; // Channel已被關閉
        }
        try {
            if (connectPromise != null) {
                throw new ConnectionPendingException(); // 已有鏈接操做正在進行
            }
            boolean wasActive = isActive();
            // 模板方法，細節子類完成
            if (doConnect(remoteAddress, localAddress)) {
                fulfillConnectPromise(promise, wasActive);  // 鏈接操做已完成
            } else {
                // 鏈接操做還沒有完成
                connectPromise = promise;
                requestedRemoteAddress = remoteAddress;
                // 這部分代碼爲Netty的鏈接超時機制
                int connectTimeoutMillis = config().getConnectTimeoutMillis();
                if (connectTimeoutMillis > 0) {
                    connectTimeoutFuture = eventLoop().schedule(() -> {
                        ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                        ConnectTimeoutException cause = new ConnectTimeoutException("...");
                        if (connectPromise != null && connectPromise.tryFailure(cause)) {
                            close(voidPromise());
                        }
                    }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                }

                promise.addListener((ChannelFutureListener) (future) -> {
                    if (future.isCancelled()) {
                        // 鏈接操做取消則鏈接超時檢測任務取消
                        if (connectTimeoutFuture != null) {
                            connectTimeoutFuture.cancel(false);
                        }
                        connectPromise = null;
                        close(voidPromise());
                    }
                });
            }
        } catch (Throwable t) {
            promise.tryFailure(annotateConnectException(t, remoteAddress));
            closeIfClosed();
        }
    }

Connect事件框架中包含了Netty的鏈接超時檢測機制：向EventLoop提交一個調度任務，設定的超時時間已到則向鏈接操做的異步結果設置失敗而後關閉鏈接。fulfillConnectPromise()設置異步結果爲成功並觸發Channel的Active事件：

    private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
        if (promise == null) {
            return; // 操做已取消或Promise已被通知？
        }
        boolean active = isActive();
        boolean promiseSet = promise.trySuccess();  // False表示用戶取消操做
        if (!wasActive && active) { // 此時用戶沒有取消Connect操做
            pipeline().fireChannelActive(); // 觸發Active事件
        }
        if (!promiseSet) {
            close(voidPromise()); // 操做已被用戶取消，關閉Channel
        }
    }

FinishConnect事件框架：

    public final void finishConnect() {
        assert eventLoop().inEventLoop();
        try {
            boolean wasActive = isActive();
            doFinishConnect();  // 模板方法
            fulfillConnectPromise(connectPromise, wasActive);   // 首次Active觸發Active事件
        } catch (Throwable t) {
            fulfillConnectPromise(connectPromise, annotateConnectException(...));
        } finally {
            if (connectTimeoutFuture != null) {
                connectTimeoutFuture.cancel(false); // 鏈接完成，取消超時檢測任務
            }
            connectPromise = null;
        }
    }

finishConnect()只由EventLoop處理就緒selectionKey的OP_CONNECT事件時調用，從而完成鏈接操做。注意：鏈接操做被取消或者超時不會使該方法被調用。

Flush事件細節：

    protected final void flush0() {
        if (isFlushPending()) {
            return; // 已經有flush操做，返回
        }
        super.flush0(); // 調用父類方法
    }

    public final void forceFlush() {
        super.flush0(); // 調用父類方法
    }

    private boolean isFlushPending() {
        SelectionKey selectionKey = selectionKey();
        return selectionKey.isValid() && 
                    (selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0;
    }

forceFlush()方法由EventLoop處理就緒selectionKey的OP_WRITE事件時調用，將緩衝區中的數據寫入Channel。isFlushPending()方法容易致使困惑：爲何selectionKey關心OP_WRITE事件表示正在Flush呢？OP_WRITE表示通道可寫，而通常狀況下通道均可寫，若是selectionKey一直關心OP_WRITE事件，那麼將不斷從select()方法返回從而致使死循環。Netty使用一個寫緩衝區，write操做將數據放入緩衝區中，flush時設置selectionKey關心OP_WRITE事件，完成後取消關心OP_WRITE事件。因此，若是selectionKey關心OP_WRITE事件表示此時正在Flush數據。

AbstractNioUnsafe還有最後一個方法removeReadOp()：

    protected final void removeReadOp() {
        SelectionKey key = selectionKey();
        if (!key.isValid()) {
            return; // selectionKey已被取消
        }
        int interestOps = key.interestOps();
        if ((interestOps & readInterestOp) != 0) {
            key.interestOps(interestOps & ~readInterestOp); // 設置爲再也不感興趣
        }
    }

Netty中將服務端的OP_ACCEPT和客戶端的Read統一抽象爲Read事件，在NIO底層I/O事件使用bitmap表示，一個二進制位對應一個I/O事件。當一個二進制位爲1時表示關心該事件，readInterestOp的二進制表示只有1位爲1，因此體會interestOps & ~readInterestOp的含義，可知removeReadOp()的功能是設置SelectionKey再也不關心Read事件。相似的，還有setReadOp()、removeWriteOp()、setWriteOp()等等。

分析完AbstractNioUnsafe，咱們再分析AbstractNioChannel，首先看其中還沒講解的字段：

    private final SelectableChannel ch; // 包裝的JDK Channel
    protected final int readInterestOp; // Read事件，服務端OP_ACCEPT，其餘OP_READ
    volatile SelectionKey selectionKey; // JDK Channel對應的選擇鍵
    private volatile boolean inputShutdown; // Channel的輸入關閉標記
    private volatile boolean readPending;   // 底層讀事件進行標記

再看一下構造方法：

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false);    // 設置非阻塞模式
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
                // log
            }
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }

其中的ch.configureBlocking(false)方法設置Channel爲非阻塞模式，從而爲Netty提供非阻塞處理I/O事件的能力。

對於AbstractNioChannel的方法，咱們主要分析它實現I/O事件框架細節部分的doXXX()方法。

    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                return;
            } catch (CancelledKeyException e) {
                if (!selected) {
                    // 選擇鍵取消從新selectNow()，清除因取消操做而緩存的選擇鍵
                    eventLoop().selectNow();
                    selected = true;
                } else {
                    throw e;
                }
            }
        }
    }
    
    protected void doDeregister() throws Exception {
        eventLoop().cancel(selectionKey()); // 設置取消選擇鍵
    }

對於Register事件，當Channel屬於NIO時，已經能夠肯定註冊操做的所有細節：將Channel註冊到給定NioEventLoop的selector上便可。注意，其中第二個參數0表示註冊時不關心任何事件，第三個參數爲Netty的NioChannel對象自己。對於Deregister事件，選擇鍵執行cancle()操做，選擇鍵表示JDK Channel和selctor的關係，調用cancle()終結這種關係，從而實現從NioEventLoop中Deregister。須要注意的是：cancle操做調用後，註冊關係不會當即生效，而會將cancle的key移入selector的一個取消鍵集合，當下次調用select相關方法或一個正在進行的select調用結束時，會從取消鍵集合中移除該選擇鍵，此時註銷才真正完成。一個Cancle的選擇鍵爲無效鍵，調用它相關的方法會拋出CancelledKeyException。

    protected void doBeginRead() throws Exception {
        if (inputShutdown) {
            return; // Channel的輸入關閉？什麼狀況下發生？
        }
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return; // 選擇鍵被取消而再也不有效
        }
        readPending = true; // 設置底層讀事件正在進行
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            // 選擇鍵關心Read事件
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

對於NioChannel的beginRead事件，只需將Read事件設置爲選擇鍵所關心的事件，則以後的select()調用若是Channel對應的Read事件就緒，便會觸發Netty的read()操做。

    protected void doClose() throws Exception {
        ChannelPromise promise = connectPromise;
        if (promise != null) {
            // 鏈接操做還在進行，但用戶調用close操做
            promise.tryFailure(DO_CLOSE_CLOSED_CHANNEL_EXCEPTION);
            connectPromise = null;
        }
        ScheduledFuture<?> future = connectTimeoutFuture;
        if (future != null) {
            // 若是有鏈接超時檢測任務，則取消
            future.cancel(false);
            connectTimeoutFuture = null;
        }
    }

此處的doClose操做主要處理了鏈接操做相關的後續處理。並無實際關閉Channel，因此須要子類繼續增長細節實現。AbstractNioChannel中還有關於建立DirectBuffer的方法，將在之後必要時進行分析。其餘的方法則較爲簡單，不在列出。最後提一下isCompatible()方法，說明NioChannel只在NioEventLoop中可用。

    protected boolean isCompatible(EventLoop loop) {
        return loop instanceof NioEventLoop;
    }

AbstractNioChannel的子類實現分爲服務端AbstractNioMessageChannel和客戶端AbstractNioByteChannel，咱們將首先分析服務端AbstractNioMessageChannel。

6.2.3 AbstractNioMessageChannel

AbstractNioMessageChannel是底層數據爲消息的NioChannel。在Netty中，服務端Accept的一個Channel被認爲是一條消息，UDP數據報也是一條消息。該類主要完善flush事件框架的doWrite細節和實現read事件框架（在內部類NioMessageUnsafe完成）。首先看read事件框架：

    public void read() {
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        if (!config.isAutoRead() && !isReadPending()) {
            // 此時讀操做不被容許，既沒有配置autoRead也沒有底層讀事件進行
            removeReadOp(); // 清除read事件，再也不關心
            return;
        }
        
        final int maxMessagesPerRead = config.getMaxMessagesPerRead();
        final ChannelPipeline pipeline = pipeline();
        boolean closed = false;
        Throwable exception = null;
        try {
            try {
                for (;;) {
                    int localRead = doReadMessages(readBuf); // 模板方法，讀取消息
                    if (localRead == 0) { // 沒有數據可讀
                        break;  
                    }
                    if (localRead < 0) { // 讀取出錯
                        closed = true;  
                        break;
                    }
                    if (!config.isAutoRead()) { //沒有設置AutoRead
                        break;
                    }
                    if (readBuf.size() >= maxMessagesPerRead) { // 達到最大可讀數
                        break;
                    }
                }
            } catch (Throwable t) {
                exception = t;
            }
            
            setReadPending(false);  // 已沒有底層讀事件
            int size = readBuf.size();
            for (int i = 0; i < size; i ++) {
                pipeline.fireChannelRead(readBuf.get(i));   //觸發ChannelRead事件，用戶處理
            }
            readBuf.clear(); // ChannelReadComplete事件中若是配置autoRead則會調用beginRead，從而不斷進行讀操做
            pipeline.fireChannelReadComplete(); // 觸發ChannelReadComplete事件，用戶處理

            if (exception != null) {
                if (exception instanceof IOException && !(exception instanceof PortUnreachableException)) {
                    // ServerChannel異常也不能關閉，應該恢復讀取下一個客戶端
                    closed = !(AbstractNioMessageChannel.this instanceof ServerChannel);
                }
                pipeline.fireExceptionCaught(exception);
            }

            if (closed) {
                if (isOpen()) {
                    close(voidPromise());   // 非serverChannel且打開則關閉
                }
            }
        } finally {
            if (!config.isAutoRead() && !isReadPending()) {
                // 既沒有配置autoRead也沒有底層讀事件進行
                removeReadOp();
            }
        }
    }

read事件框架的流程已在代碼中註明，須要注意的是讀取消息的細節doReadMessages(readBuf)方法由子類實現。
咱們主要分析NioServerSocketChannel，它不支持doWrite()操做，因此咱們再也不分析本類的flush事件框架的doWrite細節方法，直接轉向下一個目標：NioServerSocketChannel。

6.2.4 NioServerSocketChannel

你確定已經使用過NioServerSocketChannel，做爲處於Channel最底層的子類，NioServerSocketChannel會實現I/O事件框架的底層細節。首先須要注意的是：NioServerSocketChannel只支持bind、read和close操做。

   protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) { // JDK版本1.7以上
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }
    
    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = javaChannel().accept();
        try {
            if (ch != null) {
                // 一個NioSocketChannel爲一條消息
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            logger.warn("Failed to create a new channel from an accepted socket.", t);
            try {
                ch.close();
            } catch (Throwable t2) {
                logger.warn("Failed to close a socket.", t2);
            }
        }
        return 0;
    }
    
    protected void doClose() throws Exception {
        javaChannel().close();
    }

其中的實現，都是調用JDK的Channel的方法，從而實現了最底層的細節。須要注意的是：此處的doReadMessages()方法每次最多返回一個消息（客戶端鏈接），由此可知NioServerSocketChannel的read操做一次至多處理的鏈接數爲config.getMaxMessagesPerRead()，也就是參數值MAX_MESSAGES_PER_READ。此外doClose()覆蓋了AbstractNioChannel的實現，由於NioServerSocketChannel不支持connect操做，因此不須要鏈接超時處理。

最後，咱們再看關鍵構造方法：

    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

其中的SelectionKey.OP_ACCEPT最爲關鍵，Netty正是在此處將NioServerSocketChannel的read事件定義爲NIO底層的OP_ACCEPT，統一完成read事件的抽象。

至此，咱們已分析完兩條線索中的服務端部分，下面分析客戶端部分。首先是AbstractNioChannel的另外一個子類AbstractNioByteChannel。

6.2.5 AbstractNioByteChannel

從字面可推知，AbstractNioByteChannel的底層數據爲Byte字節。首先看構造方法：

    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ);
    }

其中的SelectionKey.OP_READ，說明AbstractNioByteChannel的read事件爲NIO底層的OP_READ事件。
而後咱們看read事件框架：

    public final void read() {
        final ChannelConfig config = config();
        if (!config.isAutoRead() && !isReadPending()) {
            // 此時讀操做不被容許，既沒有配置autoRead也沒有底層讀事件進行
            removeReadOp();
            return;
        }

        final ChannelPipeline pipeline = pipeline();
        final ByteBufAllocator allocator = config.getAllocator();
        final int maxMessagesPerRead = config.getMaxMessagesPerRead();
        RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
        if (allocHandle == null) {
            this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
        }

        ByteBuf byteBuf = null;
        int messages = 0;
        boolean close = false;
        try {
            int totalReadAmount = 0;
            boolean readPendingReset = false;
            do {
                byteBuf = allocHandle.allocate(allocator);  // 建立一個ByteBuf
                int writable = byteBuf.writableBytes(); 
                int localReadAmount = doReadBytes(byteBuf); // 模板方法，子類實現細節
                if (localReadAmount <= 0) { // 沒有數據可讀
                    byteBuf.release();
                    byteBuf = null;
                    close = localReadAmount < 0; // 讀取數據量爲負數表示對端已經關閉
                    break;
                }
                if (!readPendingReset) {
                    readPendingReset = true;
                    setReadPending(false);  // 沒有底層讀事件進行
                    // 此時，若autoRead關閉則必須調用beginRead，read操做纔會讀取數據
                }
                pipeline.fireChannelRead(byteBuf);  // 觸發ChannelRead事件，用戶處理
                byteBuf = null;

                if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {   // 防止溢出
                    totalReadAmount = Integer.MAX_VALUE;
                    break;
                }
                totalReadAmount += localReadAmount;

                if (!config.isAutoRead()) { // 沒有配置AutoRead
                    break;
                }
                if (localReadAmount < writable) {   // 讀取數小於可寫數，可能接受緩衝區已徹底耗盡
                    break;
                }
            } while (++ messages < maxMessagesPerRead);

            // ReadComplete結束時，若是開啓autoRead則會調用beginRead，從而能夠繼續read
            pipeline.fireChannelReadComplete();
            allocHandle.record(totalReadAmount);

            if (close) {
                closeOnRead(pipeline);
                close = false;
            }
        } catch (Throwable t) {
            handleReadException(pipeline, byteBuf, t, close);
        } finally {
            if (!config.isAutoRead() && !isReadPending()) {
                // 既沒有配置autoRead也沒有底層讀事件進行
                removeReadOp(); 
            }
        }
    }

AbstractNioByteChannel的read事件框架處理流程與AbstractNioMessageChannel的稍有不一樣：AbstractNioMessageChannel依次讀取Message，最後統一觸發ChannelRead事件；而AbstractNioByteChannel每讀取到必定字節就觸發ChannelRead事件。這是由於，AbstractNioMessageChannel需求高吞吐量，特別是ServerSocketChannel須要儘量多地接受鏈接；而AbstractNioByteChannel需求快響應，要儘量快地響應遠端請求。

read事件的具體流程請參考代碼和代碼註釋進行理解，再也不分析。注意到代碼中有關於接收緩衝區的代碼，這一部分咱們單獨使用一節講述，以後會分析。當讀取到的數據小於零時，表示遠端鏈接已關閉，這時會調用closeOnRead(pipeline)方法：

    private void closeOnRead(ChannelPipeline pipeline) {
        SelectionKey key = selectionKey();
        setInputShutdown(); // 遠端關閉此時設置Channel的輸入源關閉
        if (isOpen()) {
            if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
                // 取消關心Read事件並觸發UserEvent事件ChannelInputShutdownEvent
                key.interestOps(key.interestOps() & ~readInterestOp);   
                pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
            } else {
                close(voidPromise());   // 直接關閉
            }
        }
    }

這段代碼正是Channel參數ALLOW_HALF_CLOSURE的意義描述，該參數爲True時，會觸發用戶事件ChannelInputShutdownEvent，不然，直接關閉該Channel。

拋出異常時，會調用handleReadException(pipeline, byteBuf, t, close)方法：

    private void handleReadException(ChannelPipeline pipeline,
                                         ByteBuf byteBuf, Throwable cause, boolean close) {
        if (byteBuf != null) {  // 已讀取到數據
            if (byteBuf.isReadable()) { // 數據可讀
                setReadPending(false);
                pipeline.fireChannelRead(byteBuf);  
            } else {    // 數據不可讀
                byteBuf.release();
            }
        }
        pipeline.fireChannelReadComplete();
        pipeline.fireExceptionCaught(cause);
        if (close || cause instanceof IOException) {
            closeOnRead(pipeline);
        }
    }

可見，拋出異常時，若是讀取到可用數據和正常讀取同樣觸發ChannelRead事件，只是最後會統一觸發ExceptionCaught事件由用戶進行處理。

至此，read事件框架分析完畢，下面咱們分析write事件的細節實現方法doWrite()。在此以前，先看filterOutboundMessage()方法對須要寫的數據進行過濾。

    protected final Object filterOutboundMessage(Object msg) {
        if (msg instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) msg;
            if (buf.isDirect()) {
                return msg;
            }
            return newDirectBuffer(buf); // 非DirectBuf轉爲DirectBuf
        }
        if (msg instanceof FileRegion) {
            return msg;
        }
        throw new UnsupportedOperationException("...");
    }

可知，Netty支持的寫數據類型只有兩種：DirectBuffer和FileRegion。咱們再看這些數據怎麼寫到Channel上，也就是doWrite()方法：

    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        int writeSpinCount = -1;
        boolean setOpWrite = false;
        for (;;) {
            Object msg = in.current();
            if (msg == null) {  // 數據已所有寫完
                clearOpWrite();     // 清除OP_WRITE事件
                return;
            }

            if (msg instanceof ByteBuf) {
                ByteBuf buf = (ByteBuf) msg;
                int readableBytes = buf.readableBytes();
                if (readableBytes == 0) {
                    in.remove();
                    continue;
                }

                boolean done = false;
                long flushedAmount = 0;
                if (writeSpinCount == -1) {
                    writeSpinCount = config().getWriteSpinCount();
                }
                for (int i = writeSpinCount - 1; i >= 0; i --) {
                    int localFlushedAmount = doWriteBytes(buf); // 模板方法，子類實現細節
                    if (localFlushedAmount == 0) {
                        // NIO在非阻塞模式下寫操做可能返回0表示未寫入數據
                        setOpWrite = true;
                        break;
                    }

                    flushedAmount += localFlushedAmount;
                    if (!buf.isReadable()) {
                        // ByteBuf不可讀，此時數據已寫完
                        done = true;
                        break;
                    }
                }
                
                in.progress(flushedAmount); // 記錄進度
                if (done) {
                    in.remove();    // 完成時，清理緩衝區
                } else {
                    break;  // 跳出循環執行incompleteWrite()
                }
            } else if (msg instanceof FileRegion) {
                // ....
            } else {
                throw new Error();  // 其餘類型不支持
            }
        }
        incompleteWrite(setOpWrite);
    }

代碼中省略了對FileRegion的處理，FileRegion是Netty對NIO底層的FileChannel的封裝，負責將File中的數據寫入到WritableChannel中。FileRegion的默認實現是DefaultFileRegion，若是你很感興趣它的實現，能夠自行查閱。

咱們主要分析對ByteBuf的處理。doWrite的流程簡潔明瞭，核心操做是模板方法doWriteBytes(buf)，將ByteBuf中的數據寫入到Channel，因爲NIO底層的寫操做返回已寫入的數據量，在非阻塞模式下該值可能爲0，此時會調用incompleteWrite()方法：

    protected final void incompleteWrite(boolean setOpWrite) {
        if (setOpWrite) {
            setOpWrite();   // 設置繼續關心OP_WRITE事件
        } else {
            // 此時已進行寫操做次數writeSpinCount，但並無寫完
            Runnable flushTask = this.flushTask;
            if (flushTask == null) {
                flushTask = this.flushTask = (Runnable) () -> { flush(); };
            }
            // 再次提交一個flush()任務
            eventLoop().execute(flushTask);
        }
    }

該方法分兩種狀況處理，在上文提到的第一種狀況（實際寫0數據）下，設置SelectionKey繼續關心OP_WRITE事件從而繼續進行寫操做；第二種狀況下，也就是寫操做進行次數達到配置中的writeSpinCount值但還沒有寫完，此時向EventLoop提交一個新的flush任務，此時能夠響應其餘請求，從而提交響應速度。這樣的處理，不會使大數據的寫操做佔用所有資源而使其餘請求得不到響應，可見這是一個較爲公平的處理。這裏引出一個問題：使用Netty如何搭建高性能文件服務器？
至此，已分析完對於Byte數據的read事件和doWrite細節的處理，接下里，繼續分析NioSocketChannel，從而完善各事件框架的細節部分。

6.2.6 NioSocketChannel

NioSocketChannel做爲Channel的最末端子類，實現了NioSocket相關的最底層細節實現，首先看doBind()：

    protected void doBind(SocketAddress localAddress) throws Exception {
        doBind0(localAddress);
    }

    private void doBind0(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            javaChannel().bind(localAddress);   // JDK版本1.7以上
        } else {
            javaChannel().socket().bind(localAddress);
        }
    }

這部分代碼與NioServerSocketChannel中相同，委託給JDK的Channel進行綁定操做。
接着再看doConnect()和doFinishConnect()方法：

    protected boolean doConnect(SocketAddress remoteAddress, 
                                        SocketAddress localAddress) throws Exception {
        if (localAddress != null) {
            doBind0(localAddress);
        }

        boolean success = false;
        try {
            boolean connected = javaChannel().connect(remoteAddress);
            if (!connected) {
                // 設置關心OP_CONNECT事件，事件就緒時調用finishConnect()
                selectionKey().interestOps(SelectionKey.OP_CONNECT);
            }
            success = true;
            return connected;
        } finally {
            if (!success) {
                doClose();
            }
        }
    }

    protected void doFinishConnect() throws Exception {
        if (!javaChannel().finishConnect()) {
            throw new Error();
        }
    }

JDK中的Channel在非阻塞模式下調用connect()方法時，會當即返回結果：成功創建鏈接返回True，操做還在進行時返回False。返回False時，須要在底層OP_CONNECT事件就緒時，調用finishConnect()方法完成鏈接操做。
再看doDisconnect()和doClose()方法：

    protected void doDisconnect() throws Exception {
        doClose();
    }

    protected void doClose() throws Exception {
        super.doClose();    // AbstractNioChannel中關於鏈接超時的處理
        javaChannel().close();
    }

而後看核心的doReadBytes()和doWriteXXX()方法：

    protected int doReadBytes(ByteBuf byteBuf) throws Exception {
        return byteBuf.writeBytes(javaChannel(), byteBuf.writableBytes());
    }

    protected int doWriteBytes(ByteBuf buf) throws Exception {
        final int expectedWrittenBytes = buf.readableBytes();
        return buf.readBytes(javaChannel(), expectedWrittenBytes);
    }

    protected long doWriteFileRegion(FileRegion region) throws Exception {
        final long position = region.transfered();
        return region.transferTo(javaChannel(), position);
    }

對於read和write操做，委託給ByteBuf處理，咱們將使用專門的一章，對這一部分細節進行完善，將在後面介紹。
NioSocketChannel最重要的部分是覆蓋了父類的doWrite()方法，使用更高效的方式進行寫操做，其代碼以下：

    protected void doWrite(ChannelOutboundBuffer in) throws Exception {
        for (;;) {
            int size = in.size();
            if (size == 0) {
                clearOpWrite(); // 全部數據已寫完，再也不關心OP_WRITE事件
                break;
            }
            long writtenBytes = 0;
            boolean done = false;
            boolean setOpWrite = false;

            ByteBuffer[] nioBuffers = in.nioBuffers();
            int nioBufferCnt = in.nioBufferCount();
            long expectedWrittenBytes = in.nioBufferSize();
            SocketChannel ch = javaChannel();

            switch (nioBufferCnt) {
                case 0: // 沒有ByteBuffer，也就是隻有FileRegion
                    super.doWrite(in);  // 使用父類方法進行普通處理
                    return;
                case 1: // 只有一個ByteBuffer，此時的處理等效於父類方法的處理
                    ByteBuffer nioBuffer = nioBuffers[0];
                    for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
                        final int localWrittenBytes = ch.write(nioBuffer);
                        if (localWrittenBytes == 0) {
                            setOpWrite = true;
                            break;
                        }
                        expectedWrittenBytes -= localWrittenBytes;
                        writtenBytes += localWrittenBytes;
                        if (expectedWrittenBytes == 0) {
                            done = true;
                            break;
                        }
                    }
                    break;
                default: // 多個ByteBuffer，採用gathering方法處理
                    for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
                        // gathering方法，此時一次寫多個ByteBuffer
                        final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
                        if (localWrittenBytes == 0) {
                            setOpWrite = true;
                            break;
                        }
                        expectedWrittenBytes -= localWrittenBytes;
                        writtenBytes += localWrittenBytes;
                        if (expectedWrittenBytes == 0) {
                            done = true;
                            break;
                        }
                    }
                    break;
            }
            in.removeBytes(writtenBytes);   // 清理緩衝區
            if (!done) {
                incompleteWrite(setOpWrite);    // 寫操做並無完成
                break;
            }
        }
    }

在明白了父類的doWrite方法後，這段代碼便容易理解，本段代碼作的優化是：當輸出緩衝區中有多個buffer時，採用Gathering Writes將數據從這些buffer寫入到同一個channel。
在AbstractUnsafe對close事件框架的分析中，有一個prepareToClose()方法，進行關閉的必要處理並在必要時返回一個Executor執行doClose()操做，默認方法返回null，NioSocketChannelUnsafe覆蓋了父類的實現，代碼以下：

    protected Executor prepareToClose() {
            try {
                if (javaChannel().isOpen() && config().getSoLinger() > 0) {
                    doDeregister(); // 取消選擇鍵selectionKey
                    return GlobalEventExecutor.INSTANCE;
                }
            } catch (Throwable ignore) {
                //
            }
            return null;
        }

SO_LINGER表示Socket關閉的延時時間，在此時間內，內核將繼續把TCP緩衝區的數據發送給對端且執行close操做的線程將阻塞直到數據發送完成。Netty的原則是I/O線程不能被阻塞，因此此時返回一個Executor用於執行阻塞的doClose()操做。doDeregister()取消選擇鍵selectionKey是由於：延遲關閉期間，若是selectionKey仍然關心OP_WRITE事件，而輸出緩衝區又爲null，這樣write操做直接返回，不會再執行clearOpWrite()操做取消關心OP_WRITE事件，而Channel通常是可寫的，這樣OP_WRITE事件會不斷就緒從而耗盡CPU，因此須要取消選擇鍵刪除註冊的事件。
[1]: //upload-images.jianshu.io/upload_images/3288959-5a4be2f31620177d.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240
[2]: http://img.blog.csdn.net/20160928165809260
[3]: https://github.com/netty/netty/issues/4435

自頂向下深刻分析Netty（六）--Channel總述

自頂向下深刻分析Netty（六）--Channel總述

自頂向下深刻分析Netty（六）--Channel源碼實現

6.1 總述

6.1.1 Channel

6.1.2 Channel配置參數

(1).通用參數

(2).SocketChannel參數

(3).ServerSocketChannel參數

(4).DatagramChannel參數

6.1.3 Channel接口

6.1.4 Unsafe

6.2 Channel實現

6.2.1 AbstractChannel

register事件框架

bind事件框架

disconnect事件框架

close事件框架

deregister事件框架

write事件框架