2.59 1.检测新连接

1. 打断点

有新连接过来的时候，会调用NioEventLoop中run方法的processSelectedKeys中OP_ACCEPT逻辑

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
	//...
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
        unsafe.read();
    }
    //...
}

1.1. 启动客户端连接

我们在unsafe.read打个断点，启动服务器，并用telnet或者nc链接

nc localhost 8000

2. 新连接进入

• 进入AbstractNioMessageChannel.NioMessageUnsafe#read

public void read() {
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        final ChannelPipeline pipeline = pipeline();
    	//这个allocHandler用于处理服务器端接入的速率
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.reset(config);


        boolean closed = false;
        Throwable exception = null;
        try {
            try {
            	//一直接受客户端的链接，直到没有客户端连接上来或者超过速率
                do {
                	//NioServerSocketChannel#doReadMessages
					//获取jdk底层的channel
                    int localRead = doReadMessages(readBuf);
                    if (localRead == 0) {
                        break;
                    }
                    if (localRead < 0) {
                        closed = true;
                        break;
                    }
					//对接入的连接进行计数
                    allocHandle.incMessagesRead(localRead);
            	//速率控制
                } while (allocHandle.continueReading());
            } catch (Throwable t) {
                exception = t;
            }

            int size = readBuf.size();
            for (int i = 0; i < size; i ++) {
                readPending = false;
            	//分配线程及注册selector
                pipeline.fireChannelRead(readBuf.get(i));
            }
            readBuf.clear();
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();

            if (exception != null) {
                closed = closeOnReadError(exception);

                pipeline.fireExceptionCaught(exception);
            }

            if (closed) {
                inputShutdown = true;
                if (isOpen()) {
                    close(voidPromise());
                }
            }
        } finally {
            // Check if there is a readPending which was not processed yet.
            // This could be for two reasons:
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
            //
            // See https://github.com/netty/netty/issues/2254
            if (!readPending && !config.isAutoRead()) {
                removeReadOp();
            }
        }
    }
}

客户端连接过来之后主要做三件事：accept新连接并获取jdk底层channel、对连接接入的速率进行控制、分配线程及注册selector

2.1. accept新连接并获取jdk底层channel

protected int doReadMessages(List<Object> buf) throws Exception {
	//调用accept后拿到了jdk底层的channel
    SocketChannel ch = SocketUtils.accept(javaChannel());

    try {
        if (ch != null) {
        	//创建NioSocketChannel，放到集合里
            buf.add(new NioSocketChannel(this, ch));
            return 1;//返回1说明有一个客户端的channel
        }
    } 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;
}

2.1.1. NioSocketChannel的创建

2.创建NioSocketChannel

2.2. 对连接接入的速率进行控制

DefaultMaxMessagesRecvByteBufAllocator.MaxMessageHandle#continueReading(io.netty.util.UncheckedBooleanSupplier)
public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
    return config.isAutoRead() &&
           (!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
       		//目前的连接数<16
           totalMessages < maxMessagePerRead &&
           totalBytesRead > 0;
}

2.3. 分配线程及注册selector

3.分配线程及注册selector