5.接入客戶端連接

• 服務端發現新連接
  • 循環讀取read事件
  • 將新連接交給pipeline處理
  • 收尾工作
  • 清除感興趣事件
• 兩種Channel的類比關系
• 感想

服務端發現新連接

在服務端啟動過程中，ServerBootstrap通過反射的方式創建了一個NioServerSocketChannel，並且綁定了OP_ACCEPT感興趣事件。啟動之后，bossGroup中的NioEventLoop線程不斷輪詢這些事件，並進行處理。
前一節已經簡述了一下processSelectedKey方法的代碼，這里單獨抽出處理OP_ACCEPT的相關代碼

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

netty將OP_ACCEPT和OP_READ統一視為read事件，所以這里兩種事件一起判斷。繼續跟進unsafe.read，這里是NioMessageUnsafe對象

private final List<Object> readBuf = new ArrayList<Object>();

public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);
    try {
        do {
            int localRead = doReadMessages(readBuf);
            if (localRead == 0) {
                break;
            }
            if (localRead < 0) {
                break;
            }
            allocHandle.incMessagesRead(localRead);
        } while (allocHandle.continueReading());
        int size = readBuf.size();
        for (int i = 0; i < size; i++) {
            readPending = false;
            pipeline.fireChannelRead(readBuf.get(i));
        }
        readBuf.clear();
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();                
    } finally {                
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

read代碼比較長，可以分為如下幾部分

1. 檢測是否為nioEventLoop線程
2. 拿到config和pipeline
3. 創建並初始化一個ByteBufAllocator.Handle，這個Handle可以動態變更緩沖區的容量。
4. 循環讀取read事件
5. 將新連接交給pipeline處理
6. 收尾工作
7. 若不滿足條件，移除感興趣事件

循環讀取read事件

前兩個步驟無需多言，步驟3不是本章重點，略過。於是來到步驟4，先看一下doReadMessage方法

protected int doReadMessages(List<Object> buf) throws Exception {
    SocketChannel ch = SocketUtils.accept(javaChannel());
    if (ch != null) {
        buf.add(new NioSocketChannel(this, ch));
        return 1;
    }
    return 0;
}

在這個方法里,netty調用底層的ServerSocketChannel去接受一個新連接，並將它放入臨時的readBuf容器內。出方法后，若接收到了新連接，allocateHandle記錄一下連接數。循環的判斷出口是一個比較復雜的邏輯關系,在接入新連接時，totalMessages為1，而maxMessagePerRead默認為16，所以這里返回false，也就不會繼續循環。

// 繼承於MaxMessageHandle的方法
public boolean continueReading() {
    return continueReading(defaultMaybeMoreSupplier);
}

public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
    return config.isAutoRead() && (!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
        totalMessages < maxMessagePerRead && totalBytesRead > 0;
}

將新連接交給pipeline處理

對每一個readBuf容器內的新連接事件，都調用pipeline的fireChannelRead事件。pipeline首先會交給headContext，由headContext調用channelRead方法向下傳遞。在服務端啟動過程中，ServerBootStrap將一個ServerBootstrapAcceptor設置在了pipeline中，先來回顧一下它。

// ServerBootstrap的init方法截取
pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));

ServerBootstrapAcceptor將ServerSocketChannel及用戶設置的多個屬性都添加了進來，並將自身添加到pipeline中。也因此，head傳播read事件時，會調用ServerBootstrapAcceptor的channelRead方法，channelRead方法將客戶端相關的childHandler、childOptions、childAttributes都設置到NioSocketChannel上，除了handler外，都是使用netty時用戶自定義的部分。而handler是一個HandlerInitializer，在完成它的initChannel方法后，會將自身從pipeline刪除，而用戶在initChannel方法中添加的handler則會保留，從而在新連接接入時服務。

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
    child.pipeline().addLast(childHandler);
    setChannelOptions(child, childOptions, logger);
    setAttributes(child, childAttrs);
    childGroup.register(child).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                forceClose(child, future.cause());
            }
        }
    });
}

在添加完成后，從workerGroup中挑選一個NioEventLoop，將NioSocketChannel注冊到上面。這部分代碼與服務端幾乎相同。
時序圖如下

與服務端稍有不同的是unsafe的register0方法中。在服務端啟動時，因為還只是注冊而沒有綁定端口，isActive為false，從而不會啟動，而到了bind完成后，服務端才會調用pipeline.fireChannelActive，從而開始各自的讀事件。

if (isActive()) {
    if (firstRegistration) {
        pipeline.fireChannelActive();
    } else if (config().isAutoRead()) {
        beginRead();
    }
}

收尾工作

從代碼上可以看出收尾工作一共3步

1. 清空readBuf容器
2. 調用Allocator.Handle的readComplete方法
3. 調用pipeline.fireChannelReadComplete方法

步驟1很簡單，步驟2調用了AdaptiveRecvByteBufAllocator.Handle的record(int actualReadBytes)方法，它可以根據本次接受的數據量動態預測下一個緩沖區容量大小。步驟3傳播readComplete事件，若沒有進行重寫的話，會再次調用channel.read方法, 進而調用unsafe.read方法。

清除感興趣事件

當readPengind為false且channel設置了非自動讀時，清除感興趣事件，具體實現是將readInterestOp取反后，與selectionKey的interestop取與。

兩種Channel的類比關系

在這張圖上可以看出來NioServerSocketChannel與NioSocketChannel呈現出軸對稱的傾向。這種良好的類繼承關系使得netty服務端與客戶端的代碼復用性很高。我們看一下AbstractChannel與AbstractNioChannel的構造方法

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

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) {
            logger.warn(
                        "Failed to close a partially initialized socket.", e2);
        }
        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}    

AbstractChannel中的newUnsafe方法是抽象的，其主要實現類就是圖中的NioMessageUnsafe和NioByteUnsafe，分別對應服務端和客戶端。Channel對於IO事件的讀寫在底層都會委托給Unsafe對象。netty將OP_ACCEPT和OP_READ都視為讀事件也是為了盡可能復用代碼。關於pipeline，將在下一小節講述。AbstractNioChannel的入參也會根據服務端與客戶端有所不同，其中NioServerSocketChannel的parent為null，NioSocketChannel的parent為創建它的NioServetSocketChannel。channel分別是jdk的ServerSocketchannel和SocketChannel，readInterestOp分別是OP_ACCEPT和OP_READ。此外兩者均設置了非阻塞模式。

感想

在服務端接入新連接這里，可以看到netty通過良好的繼承關系實現了多態，這帶來了良好的代碼復用性，但也給源碼閱讀帶來了很大的困擾。可以與服務端啟動一節多多類比，才能更好的體會netty封裝的目的。