文章13 | 阅读 5460 | 点赞0
@Override
public final void read() {
final ChannelConfig config = config();
if (shouldBreakReadReady(config)) {
clearReadPending();
return;
}
final ChannelPipeline pipeline = pipeline();
// 获取allocator,allocator是创建ByteBuf对象
final ByteBufAllocator allocator = config.getAllocator();
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
allocHandle.reset(config);
ByteBuf byteBuf = null;
boolean close = false;
try {
do {
// 从allocator获取一个ByteBuf对象实例
byteBuf = allocHandle.allocate(allocator);
// doReadBytes从底层socket读取数据到byteBuf
allocHandle.lastBytesRead(doReadBytes(byteBuf));
if (allocHandle.lastBytesRead() <= 0) {
// nothing was read. release the buffer.
byteBuf.release();
byteBuf = null;
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;
// 将byteBuf传给pipeline,从而开始数据处理
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 {
// 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();
}
}
}
final ByteBufAllocator allocator = config.getAllocator();
allocator的实际类型代表了Channel具体是使用堆内存还是直接内存。这个可以通过用户代码通过childOption设置或者使用默认的:
用户代码传入:主要是通过childOption传入,则ServerBootstrap在接收到客户端连接并创建SocketChannel时,会根据初始化ServerBootstrap时的childOption设置,对该SocketChannel的config进行配置。
// 处理客户端请求的配置
serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
// 客户端SocketChannel所使用的allocator的实现类
serverBootstrap.childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
serverBootstrap.childHandler(new NettyServerInitializer(webSocketService));
(1)当前运行平台是否支持使用Java的unsafe来进行本地方法调用;
(2)程序的系统参数是否设置了io.netty.noPreferDirect=true。如果当前平台支持unsafe且io.netty.noPreferDirect=false或者没有设置,默认为false,则使用直接内存;否则使用堆内存。
这两个参数的默认值:
1. 是否运行通过底层api直接访问直接内存,默认:允许
-Dio.netty.noPreferDirect
2. 是否允许使用sun.misc.Unsafe,默认:允许;注意:使用sun的私有类库存在平台移植问题,另外sun.misc.Unsafe类是不安全的,如果操作失败,不是抛出异常,而是虚拟机core dump,不建议使用Unsafe
-Dio.netty.noUnsafe
源码分析如下:
1.
public class DefaultChannelConfig implements ChannelConfig {
...
private volatile ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
...
}
2.
public interface ByteBufAllocator {
ByteBufAllocator DEFAULT = ByteBufUtil.DEFAULT_ALLOCATOR;
}
3. DEFAULT_ALLOCATOR
public final class ByteBufUtil {
...
static final ByteBufAllocator DEFAULT_ALLOCATOR;
static {
// android则是unpooled,其他为pooled
String allocType = SystemPropertyUtil.get(
"io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
allocType = allocType.toLowerCase(Locale.US).trim();
ByteBufAllocator alloc;
if ("unpooled".equals(allocType)) {
alloc = UnpooledByteBufAllocator.DEFAULT;
logger.debug("-Dio.netty.allocator.type: {}", allocType);
} else if ("pooled".equals(allocType)) {
alloc = PooledByteBufAllocator.DEFAULT;
logger.debug("-Dio.netty.allocator.type: {}", allocType);
} else {
alloc = PooledByteBufAllocator.DEFAULT;
logger.debug("-Dio.netty.allocator.type: pooled (unknown: {})", allocType);
}
DEFAULT_ALLOCATOR = alloc;
...
}
...
}
4. UnpooledByteBufAllocator.DEFAULT:非池化机制默认alloctor
/** * Default instance which uses leak-detection for direct buffers. */
public static final UnpooledByteBufAllocator DEFAULT =
new UnpooledByteBufAllocator(PlatformDependent.directBufferPreferred());
5. PooledByteBufAllocator.DEFAULT:池化机制默认allocator
public static final PooledByteBufAllocator DEFAULT =
new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());
6. 在4,5中,都调用了PlatformDependent.directBufferPreferred(),如果返回true,则使用直接内存,否则使用堆内存。PlatformDependent.directBufferPreferred()的底层实现如下:
private static final Throwable UNSAFE_UNAVAILABILITY_CAUSE = unsafeUnavailabilityCause0();
private static final boolean DIRECT_BUFFER_PREFERRED =
UNSAFE_UNAVAILABILITY_CAUSE == null && !SystemPropertyUtil.getBoolean("io.netty.noPreferDirect", false);
unsafeUnavailabilityCause0的实现:判断当前平台是否支持使用Java的unsafe
private static Throwable unsafeUnavailabilityCause0() {
if (isAndroid()) {
logger.debug("sun.misc.Unsafe: unavailable (Android)");
return new UnsupportedOperationException("sun.misc.Unsafe: unavailable (Android)");
}
Throwable cause = PlatformDependent0.getUnsafeUnavailabilityCause();
if (cause != null) {
return cause;
}
try {
boolean hasUnsafe = PlatformDependent0.hasUnsafe();
logger.debug("sun.misc.Unsafe: {}", hasUnsafe ? "available" : "unavailable");
return hasUnsafe ? null : PlatformDependent0.getUnsafeUnavailabilityCause();
} catch (Throwable t) {
logger.trace("Could not determine if Unsafe is available", t);
// Probably failed to initialize PlatformDependent0.
return new UnsupportedOperationException("Could not determine if Unsafe is available", t);
}
}
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://xieyizun.blog.csdn.net/article/details/85731161
内容来源于网络,如有侵权,请联系作者删除!