Netty实现大文件分块传输详解

文章目录

前言
Netty如何处理切分大文件的异步传输

前言

在Netty中，切分大文件进行传输主要利用ChunkedWriteHandler以及它的实现类，如ChunkedFile。这种机制允许你将大文件切分成多个小块（chunks），并通过Netty的pipeline进行异步发送。以下是使用Netty切分大文件进行传输的基本步骤：

设置Netty的ServerBootstrap或Bootstrap：
首先，你需要配置你的ServerBootstrap或Bootstrap，并添加ChunkedWriteHandler到你的pipeline中。ChunkedWriteHandler会处理所有ChunkedInput或ChunkedStream的写入。
创建ChunkedFile对象：
使用ChunkedFile类来创建一个代表大文件的ChunkedInput对象。你需要提供文件的路径，以及每个chunk的大小（或者默认使用ChunkedFile的默认chunk大小）。
写入Channel：
将ChunkedFile对象写入Channel。由于ChunkedFile实现了ChunkedInput接口，Netty会自动处理文件的切分和发送。

下面是一个简单的示例代码：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.stream.ChunkedFile;public class FileServer {private final int port;public FileServer(int port) {this.port = port;}public void run() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline();// 添加HTTP编解码器p.addLast(new HttpServerCodec());// 添加ChunkedWriteHandler，它负责处理ChunkedInput的写入p.addLast(new ChunkedWriteHandler());// 添加自定义的处理器p.addLast(new SimpleChannelInboundHandler<FullHttpRequest>() {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {// 获取请求的路径String uri = request.uri();// 假设文件都在服务器的某个目录下File file = new File("path/to/your/files" + uri);if (file.exists()) {// 创建ChunkedFile对象ChunkedFile chunkedFile = new ChunkedFile(file);// 设置响应头信息HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);HttpHeaders headers = response.headers();headers.set(HttpHeaderNames.CONTENT_TYPE, "application/octet-stream");headers.set(HttpHeaderNames.CONTENT_LENGTH, file.length());// 写入响应ctx.write(response);// 写入文件内容ctx.writeAndFlush(chunkedFile);} else {// 文件不存在时的处理HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND);ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);}}});}});// 绑定端口并开始接收传入的连接ChannelFuture f = b.bind(port).sync();// 等待服务器套接字关闭f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {int port = 8080;if (args.length > 0) {port = Integer.parseInt(args[0]);}new FileServer(port).run();}
}

Netty如何处理切分大文件的异步传输

在Netty中处理切分大文件的异步传输通常涉及以下几个步骤：

读取和切分文件：
Netty提供了ChunkedFile类，它实现了ChunkedInput接口，用于异步地读取和发送文件内容。你可以创建一个ChunkedFile对象，并指定要传输的文件路径和每个chunk的大小。
写入Chunk到Channel：
将ChunkedFile对象写入到Channel中。由于ChunkedFile实现了ChunkedInput接口，Netty会异步地处理文件的读取和发送。你可以使用ctx.writeAndFlush(chunkedFile)来将文件写入到Channel，并异步地发送数据。
ChunkedWriteHandler 处理分块写操作：
ctx.writeAndFlush(chunkedFile)->AbstractChannelHandlerContext->writeAndFlush(Object msg)->write->invokeWriteAndFlush

   private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {if (invokeHandler()) {invokeWrite0(msg, promise);invokeFlush0();} else {writeAndFlush(msg, promise);}}

invokeWrite0->ChunkedWriteHandler.write()

    @Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {queue.add(new PendingWrite(msg, promise));}

invokeFlush0()->ChunkedWriteHandler.flush()->doFlush

private void doFlush(final ChannelHandlerContext ctx) {final Channel channel = ctx.channel();if (!channel.isActive()) {discard(null);return;}boolean requiresFlush = true;ByteBufAllocator allocator = ctx.alloc();while (channel.isWritable()) {if (currentWrite == null) {currentWrite = queue.poll();}if (currentWrite == null) {break;}final PendingWrite currentWrite = this.currentWrite;final Object pendingMessage = currentWrite.msg;if (pendingMessage instanceof ChunkedInput) {final ChunkedInput<?> chunks = (ChunkedInput<?>) pendingMessage;boolean endOfInput;boolean suspend;Object message = null;try {message = chunks.readChunk(allocator);endOfInput = chunks.isEndOfInput();if (message == null) {// No need to suspend when reached at the end.suspend = !endOfInput;} else {suspend = false;}} catch (final Throwable t) {this.currentWrite = null;if (message != null) {ReferenceCountUtil.release(message);}closeInput(chunks);currentWrite.fail(t);break;}if (suspend) {// ChunkedInput.nextChunk() returned null and it has// not reached at the end of input. Let's wait until// more chunks arrive. Nothing to write or notify.break;}if (message == null) {// If message is null write an empty ByteBuf.// See https://github.com/netty/netty/issues/1671message = Unpooled.EMPTY_BUFFER;}ChannelFuture f = ctx.write(message);if (endOfInput) {this.currentWrite = null;// Register a listener which will close the input once the write is complete.// This is needed because the Chunk may have some resource bound that can not// be closed before its not written.// See https://github.com/netty/netty/issues/303f.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {closeInput(chunks);currentWrite.fail(future.cause());} else {// read state of the input in local variables before closing itlong inputProgress = chunks.progress();long inputLength = chunks.length();closeInput(chunks);currentWrite.progress(inputProgress, inputLength);currentWrite.success(inputLength);}}});} else if (channel.isWritable()) {f.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {closeInput(chunks);currentWrite.fail(future.cause());} else {currentWrite.progress(chunks.progress(), chunks.length());}}});} else {f.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {closeInput(chunks);currentWrite.fail(future.cause());} else {currentWrite.progress(chunks.progress(), chunks.length());if (channel.isWritable()) {resumeTransfer();}}}});}// Flush each chunk to conserve memoryctx.flush();requiresFlush = false;}}}