netty ------------ 如果selector检测到一个channel可以读了

时间:2021-04-06 00:16:45
-----------------一个NioEventLoopGroup 的初始化的时候,会初始化一个 NioEventLoop数组,每个NioEventLoop在初始化的时候,会open一个selector放到自己的属性中,再开启一个线程exector,然后调用Run方法,实际上调用的是NioEventLoop的run方法,在这个run
方法里面,执行的是一个for循环,不停的用那个selector来select注册到上面去的channel,然后根据channel的状态,处理key。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
return;
}
if (eventLoop != this || eventLoop == null) {
return;
}
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
--------------------如果channel可读,那么调用unsafe的read方法,下面看read方法的逻辑
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}

回顾一下netty的bootstrap的启动:

一、以serverBootStrap为例,

@Override
public final void read() {
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config); ByteBuf byteBuf = null;
boolean close = false;
try {
do {
byteBuf = allocHandle.allocate(allocator);
------byteBuf是对nio中的buffer的封装?
allocHandle.lastBytesRead(doReadBytes(byteBuf));
if (allocHandle.lastBytesRead() <= 0) {
byteBuf.release();
byteBuf = null;
close = allocHandle.lastBytesRead() < 0;
if (close) {
readPending = false;
}
break;
} allocHandle.incMessagesRead(1);
readPending = false;
            pipeline.fireChannelRead(byteBuf);
byteBuf = null;
} while (allocHandle.continueReading()); allocHandle.readComplete();
pipeline.fireChannelReadComplete(); if (close) {
closeOnRead(pipeline);
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close, allocHandle);
} finally {
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
@Override
protected int doReadBytes(final ByteBuf byteBuf) throws Exception {
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.attemptedBytesRead(byteBuf.writableBytes());
--------------再往下看
return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}
 @Override
public int writeBytes(ScatteringByteChannel in, int length) throws IOException {
ensureWritable(length);
----------再往下看
int writtenBytes = setBytes(writerIndex, in, length);
if (writtenBytes > 0) {
writerIndex += writtenBytes;
}
return writtenBytes;
}
@Override
public int setBytes(int index, ScatteringByteChannel in, int length) throws IOException {
ensureAccessible();
ByteBuffer tmpBuf = internalNioBuffer();
tmpBuf.clear().position(index).limit(index + length);
try {
-----------------------随便找了一个bytebuf的继承类,发现调用的还是nio的channel的read到一个nio的buffer的方法,而这个tmpNiobuf是netty的bytebuf里面的一个属性,是nio的bytebuffer。
return in.read(tmpNioBuf);
} catch (ClosedChannelException ignored) {
return -1;
}
}

JavaChannel把channel中的数据read到bytebuf中之后,调用pipeline.fireChannelRead,让bytebuf从headContext到tailContext流动,head是一个outbound为true的handler,也就是说,是一个为

写这个动作把最后一关的。相反的,tail是为读这个动作把关,但是正常都会在这之间加一个handler处理数据和逻辑,而不会到tail这一步。同理,写也是会在这中间加一个,而不会到head这一步。