Netty中使用零拷贝方式写大数据
2021-02-07 06:15
因为网络饱和的可能性，如何在异步框架中高效地写大块的数据是一个特殊的问题。由于写操作是非阻塞的，所以即使没有写出所有的数据，写操作也会在完成时返回并通知ChannelFuture。当这种情况发生时，如果仍然不停地写入，就有内存耗尽的风险。所以在写大型数据时，需要准备好处理到远程节点的连接是慢速连接的情况，这种情况会导致内存释放的延迟。让我们考虑下将一个文件内容写出到网络的情况。
我们知道在传输的过程中，由于NIO的零拷贝特性，这种特性消除了将文件的内容从文件系统移动到网络栈的复制过程。所有的这一切都发生在Netty的核心中，所以应用程序所有需要做的就是使用一个FileRegion接口的实现，其在Netty的API文档中的定义是：“通过支持零拷贝的文件传输的Channel来发送的文件区域。”
下面的代码清单展示了如何通过从FileInputStream创建一个DefaultFileRegion，并将其写入Channel，从而利用零拷贝特性来传输一个文件的内容。
使用FileRegion传输文件的内容
FileInputStream in = new FileInputStream(file); // 创建一个FileInputStream FileRegion region = new DefaultFileRegion(in.getChannel(), 0, // 以该文件的完整长度创建一个新的DefaultFileRegion file.length());channel.writeAndFlush(region).addListener(// 发送该DefaultFileRegion，并注册一个ChannelFutureListener new ChannelFutureListener(){ @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { Throwable cause = future.cause();// 处理失败 // Do something } } });
这个示例只适用于文件内容的直接传输，不包括应用程序对数据的任何处理。在需要将数据从文件系统复制到用户内存中时，可以使用ChunkedWriteHandler，它支持异步写大型数据流，而又不会导致大量的内存消耗。
关键是interfaceChunkedInput，其中类型参数B是readChunk()方法返回的类型。Netty预置了该接口的4个实现，如下表中所列出的。每个都代表了一个将由ChunkedWriteHandler处理的不定长度的数据流。
ChunkedInput 的实现
ChunkedFile 从文件中逐块获取数据，当你的平台不支持零拷贝或者你需要转换数据时使用
ChunkedNioFile 和ChunkedFile 类似，只是它使用了FileChannel
ChunkedStream 从InputStream 中逐块传输内容
ChunkedNioStream 从ReadableByteChannel 中逐块传输内容
下面代码清单说明了ChunkedStream的用法，它是实践中最常用的实现。所示的类使用了一个File以及一个SslContext进行实例化。当initChannel()方法被调用时，它将使用所示的ChannelHandler 链初始化该Channel。
当Channel 的状态变为活动的时，WriteStreamHandler 将会逐块地把来自文件中的数据作为ChunkedStream 写入。数据在传输之前将会由SslHandler加密。
使用ChunkedStream 传输文件内容
public class ChunkedWriteHandlerInitializer extends ChannelInitializer{ private final File file; private final SslContext sslCtx; public ChunkedWriteHandlerInitializer(File file, SslContext sslCtx) { this.file = file; this.sslCtx = sslCtx; } @Override protected void initChannel(Channel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new SslHandler(sslCtx.newEngine(ch.alloc());// 将SslHandler添加到ChannelPipeline中 pipeline.addLast(new ChunkedWriteHandler());// 添加ChunkedWriteHandler以处理作为ChunkedInput传入的数据 pipeline.addLast(new WriteStreamHandler());// 一旦连接建立，WriteStreamHandler就开始写文件数据 } public final class WriteStreamHandler extends ChannelInboundHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception {// 当连接建立时，channelActive()方法将使用ChunkedInput写文件数据 super.channelActive(ctx); ctx.writeAndFlush(new ChunkedStream(new FileInputStream(file))); } } }
逐块输入要使用你自己的ChunkedInput实现，请在ChannelPipeline中安装一个ChunkedWriteHandler。在本节中，我们讨论了如何通过使用零拷贝特性来高效地传输文件，以及如何通过使用ChunkedWriteHandler来写大型数据而又不必冒着导致OutOfMemoryError的风险。