有时候会有一系列的处理in的handler，使用fireChannelRead处理传递
转载自https://blog.csdn.net/u011702633/article/details/82051329
Netty源码解析(八) —— channel的read操作
客户端channel在建立连接之后会关注read事件，那么read事件在哪触发的呢？
NioEventLoop中
/**
* 读事件和 accept事件都会经过这里,但是拿到的unsafe对象不同 所以后续执行的read操作也不一样
* NioServerChannel进行accept操作
* NioChannel进行read操作
*/
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read
@Override
public final void read() {
final ChannelConfig config = config();
//若 inputClosedSeenErrorOnRead = true ，移除对 SelectionKey.OP_READ 事件的感兴趣。
if (shouldBreakReadReady(config)) {
//清楚读事件
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
//获取并重置allocHandle对象
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;//是否关闭
try {
do {
//申请bytebuf
byteBuf = allocHandle.allocate(allocator);
//读取数据，设置最后读取字节数
allocHandle.lastBytesRead(doReadBytes(byteBuf));
//没读到数据
if (allocHandle.lastBytesRead() <= 0) {
// 梅毒到数据 释放buf
byteBuf.release();
byteBuf = null;
//如果最后读取的字节为小于 0 ，说明对端已经关闭
close = allocHandle.lastBytesRead() < 0;
if (close) {
// There is nothing left to read as we received an EOF.
readPending = false;
}
break;
}
//读到了数据
allocHandle.incMessagesRead(1);
readPending = false;
//通知pipline read事件
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
} while (allocHandle.continueReading());//循环判断是否继续读取
//读取完成
allocHandle.readComplete();
//通知pipline读取完成
pipeline.fireChannelReadComplete();
if (close) {//关闭连接
closeOnRead(pipeline);
}
} catch (Throwable t) {
handleReadException(pipeline, byteBuf, t, close, allocHandle);
} finally {
.....
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
}
- 若 inputClosedSeenErrorOnRead = true ，移除对 SelectionKey.OP_READ 事件的感兴趣。
- 获取并重置allocHandle对象（代理alloctor的一些功能）
- 读取数据，设置最后读取字节数
- 通知pipline read事件
- 通知pipline读取完成
这次先跟着主线走，然后再回头看细节，直接定位到事件通知
io.netty.channel.DefaultChannelPipeline#fireChannelRead
/**
* 有数据读入的时候会调用(InBound) 也可以手动调用
* @param msg 客户端连接的channel
* @return
*/
@Override
public final ChannelPipeline fireChannelRead(Object msg) {
//pipline节点类的静态方法 穿进去的head
AbstractChannelHandlerContext.invokeChannelRead(head, msg);
return this;
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
最后调用到io.netty.channel.DefaultChannelPipeline.HeadContext#channelRead
/**
* 被调用read 可以向下传递
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ctx.fireChannelRead(msg);
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
ctx.fireChannelRead(msg);向下个节点传递，如果自定义了handler来处理就可以拦截channelRead的bytebuf数据来进行处理，负责一直向下传递到TailContext节点处理
io.netty.channel.DefaultChannelPipeline.TailContext#channelRead
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//如果不去自定义handler处理byteBuf 最终会到TailContext 来处理
onUnhandledInboundMessage(msg);
}
protected void onUnhandledInboundMessage(Object msg) {
try {
//日志提醒没有处理
logger.debug(
"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
} finally {
//释放byteBuf内存
ReferenceCountUtil.release(msg);
}
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
数据到达TailContext节点之后，再onUnhandledInboundMessage方法中打印数据未处理日志，然后释放bytebuf内存
io.netty.channel.nio.AbstractNioByteChannel#doReadBytes读取操作的方法
@Override
/**
* 读取数据
*/
protected int doReadBytes(ByteBuf byteBuf) throws Exception {
//获取handle对象
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
//设置读索引=写索引
allocHandle.attemptedBytesRead(byteBuf.writableBytes());
//读取无数据到buf
return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- allocHandle.lastBytesRead()小于0证明没有读取到数据，要释放bytebuf
- 读取到数据，allocHandle.incMessagesRead(1);
- allocHandle.continueReading() 循环判断是否继续读取
@Override
public boolean continueReading() {
return continueReading(defaultMaybeMoreSupplier);
}
@Override
public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
// Keep reading if we are allowed to read more bytes, and our last read filled up the buffer we provided.
//最后读取的字节数大于0 并且等于最大可写入的字节数
return bytesToRead > 0 && maybeMoreDataSupplier.get();
}
private final UncheckedBooleanSupplier defaultMaybeMoreSupplier = new UncheckedBooleanSupplier() {
@Override
public boolean get() {
//最后读取的字节数 是否等于最大可写入字节数
return attemptedBytesRead == lastBytesRead;
}
};
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
最后读取的字节数大于0，并且最后读取的数据==尝试可写入的大小，即证明可以继续读取
出现异常时候调用io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#handleReadException
private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
RecvByteBufAllocator.Handle allocHandle) {
if (byteBuf != null) {
if (byteBuf.isReadable()) {
readPending = false;
//把已经读取到的数据 通知到pipline中
pipeline.fireChannelRead(byteBuf);
} else {
//释放bytebuf
byteBuf.release();
}
}
//读取完成
allocHandle.readComplete();
//通知pipline读取完成
pipeline.fireChannelReadComplete();
//通知pipline异常
pipeline.fireExceptionCaught(cause);
if (close || cause instanceof IOException) {
closeOnRead(pipeline);
}
}- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
io.netty.channel.DefaultChannelPipeline#fireExceptionCaught
@Override
public final ChannelPipeline fireExceptionCaught(Throwable cause) {
AbstractChannelHandlerContext.invokeExceptionCaught(head, cause);
return this;
}最终调用到TailContext的io.netty.channel.DefaultChannelPipeline.TailContext#exceptionCaught
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
onUnhandledInboundException(cause);
}
protected void onUnhandledInboundException(Throwable cause) {
try {
logger.warn(
"An exceptionCaught() event was fired, and it reached at the tail of the pipeline. " +
"It usually means the last handler in the pipeline did not handle the exception.",
cause);
} finally {
ReferenceCountUtil.release(cause);
}
}和读取操作一样，最终也是要再发生异常的时候释放buf的内存