NOTE
2.59 1.检测新连接
1. 打断点 有新连接过来的时候,会调用NioEventLoop中run方法的processSelectedKeys中OP ACCEPT逻辑 1.1. 启动客户端连接 我们在unsafe.read打个断点,启动服务器,并用telnet或者nc链接 2. 新连接进入 - 进入AbstractNioMe
这是历史学习笔记,可能存在过时或不完整的理解。
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.2. 对连接接入的速率进行控制
DefaultMaxMessagesRecvByteBufAllocator.MaxMessageHandle#continueReading(io.netty.util.UncheckedBooleanSupplier)
public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
return config.isAutoRead() &&
(!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
//目前的连接数<16
totalMessages < maxMessagePerRead &&
totalBytesRead > 0;
}