Netty源码分析-Channel对Buffer的堆内存和直接内存的使用

痛定思痛。 2022-03-31 19:12 424阅读 0赞

### Channel的Unsafe对Buffer的使用 ###

*  在Channel的接口体系设计中，Channel只是与ChannelHandler，ChannelHandlerContext，ChannelPipeline完成对数据的处理，而不直接参与和底层socket进行数据交互，是通过一个内部Unsafe的接口设计来完成这项工作。
 *  如对于数据读取，在Channel的实现中拿到的是Unsafe已经读取回来的数据，Unsafe将socket的字节数据读取后转为了对象Object msg的数据，或者将ServerSocket接收到的客户端socket连接，将socket对象交给Channel。具体为交给Channel所绑定的ChannelPipeline，从而开始Channel体系结构的数据流处理。
 *  故Buffer包中相关缓存实现类主要是在Channel的Unsafe接口的实现类中引用，读取底层Java NIO的socket的数据，如SocketChannel的bytes：  
    如下为AbstractNioByteChannel的NioByteUnsafe的read方法：  
    AbstractNioByteChannel是NioSocketChannel的基类，主要是读取客户端发送过来的字节数据

@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();
            }
        }
    }

*  从以上代码可以看出是ByteBuf的allocator是从ChannelConfig中获取：

final ByteBufAllocator allocator = config.getAllocator();

### Channel使用堆内存还是直接内存 ###

*  allocator的实际类型代表了Channel具体是使用堆内存还是直接内存。这个可以通过用户代码通过childOption设置或者使用默认的：
    
    1.  用户代码传入：主要是通过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.  默认实现，一般不需要通过用户代码传入，使用默认的即可，默认实现为：如果当前的运行平台是android，则使用unpooled，即不使用池化机制，因为android移动端，内存资源有限，不适合做缓存；其他平台则默认使用池化机制，提高性能，以空间换时间。其次是使用堆内存还是直接内存的问题，主要是根据：
        
        （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);
            }
        }