概述
Netty 传输文件的时候没有使用 ByteBuf 进行向 Channel 中写入数据，而使用的 FileRegion。下面通过示例了解下 FileRegion 的用法，然后深入源码分析 为什么不使用 ByteBuf 而使用 FileRegion。
示例 （Netty example 中的示例）
public final class FileServer {
public static void main(String[] args) throws Exception {
// Configure the server.
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(
new StringEncoder(CharsetUtil.UTF_8),
new LineBasedFrameDecoder(8192),
new StringDecoder(CharsetUtil.UTF_8),
new ChunkedWriteHandler(),
// 自定义 Handler
new FileServerHandler());
}
});
// 起动服务
ChannelFuture f = b.bind(8080).sync();
// 等待服务关闭
f.channel().closeFuture().sync();
} finally {
// Shut down all event loops to terminate all threads.
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
从示例中可以看出 ChannelPipeline 中添加了自定义的 FileServerHandler()。
下面看下 FileServerHandler 的源码，其它几个 Handler 的都是 Netty 中自带的，以后会分析这些 Handler 的具体实现原理。
public class FileServerHandler extends SimpleChannelInboundHandler<String> {
@Override
public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
RandomAccessFile raf = null;
long length = -1;
try {
raf = new RandomAccessFile(msg, "r");
length = raf.length();
} catch (Exception e) {
ctx.writeAndFlush("ERR: " + e.getClass().getSimpleName() + ": " + e.getMessage() + '\n');
return;
} finally {
if (length < 0 && raf != null) {
raf.close();
}
}
ctx.write("OK: " + raf.length() + '\n');
if (ctx.pipeline().get(SslHandler.class) == null) {
// 传输文件使用了 DefaultFileRegion 进行写入到 NioSocketChannel 中
ctx.write(new DefaultFileRegion(raf.getChannel(), 0, length));
} else {
// SSL enabled - cannot use zero-copy file transfer.
ctx.write(new ChunkedFile(raf));
}
ctx.writeAndFlush("\n");
}
}
从 FileServerHandler 中可以看出，传输文件使用了 DefaultFileRegion 进行写入到 NioSocketChannel 里。
我们知道向 NioSocketChannel 里写数据，都是使用的 ByteBuf 进行写入。这里为啥使用 DefaultFileRegion 呢？
DefaultFileRegion 源码
public class DefaultFileRegion extends AbstractReferenceCounted implements FileRegion {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultFileRegion.class);
// 传输的文件
private final File f;
// 文件的其实坐标
private final long position;
// 传输的字节数
private final long count;
// 已经写入的字节数
private long transferred;
// 传输文件对应的 FileChannel
private FileChannel file;
/**
* Create a new instance
*
* @param file 要传输的文件
* @param position 传输文件的其实位置
* @param count 传输文件的字节数
*/
public DefaultFileRegion(FileChannel file, long position, long count) {
if (file == null) {
throw new NullPointerException("file");
}
if (position < 0) {
throw new IllegalArgumentException("position must be >= 0 but was " + position);
}
if (count < 0) {
throw new IllegalArgumentException("count must be >= 0 but was " + count);
}
this.file = file;
this.position = position;
this.count = count;
f = null;
}
....
}
transferTo() 方法
DefaultFileRegion 中有一个很重要的方法 transferTo() 方法
@Override
public long transferTo(WritableByteChannel target, long position) throws IOException {
long count = this.count - position;
if (count < 0 || position < 0) {
throw new IllegalArgumentException(
"position out of range: " + position +
" (expected: 0 - " + (this.count - 1) + ')');
}
if (count == 0) {
return 0L;
}
if (refCnt() == 0) {
throw new IllegalReferenceCountException(0);
}
// Call open to make sure fc is initialized. This is a no-oop if we called it before.
open();
long written = file.transferTo(this.position + position, count, target);
if (written > 0) {
transferred += written;
}
return written;
}
这里可以看出 文件 通过 FileChannel.transferTo 方法直接发送到 WritableByteChannel 中。
通过 Nio 的 FileChannel 可以使用 map 文件映射的方式，直接发送到 SocketChannel中，这样可以减少两次 IO 的复制。
第一次 IO：读取文件的时间从系统内存中拷贝到 jvm 内存中。
第二次 IO：从 jvm 内存中写入 Socket 时，再 Copy 到系统内存中。
这就是所谓的零拷贝技术。
写入 FileRegion
public abstract class AbstractNioByteChannel extends AbstractNioChannel {
private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
......
} else if (msg instanceof FileRegion) {
FileRegion region = (FileRegion) msg;
if (region.transferred() >= region.count()) {
in.remove();
return 0;
}
long localFlushedAmount = doWriteFileRegion(region);
if (localFlushedAmount > 0) {
in.progress(localFlushedAmount);
if (region.transferred() >= region.count()) {
in.remove();
}
return 1;
}
} else {
throw new Error();
}
return WRITE_STATUS_SNDBUF_FULL;
}
从 ChannelOutboundBuffer 中获取 FileRegion 类型的节点。
然后调用 NioSocketChannel.doWriteFileRegion() 方法进行写入。
NioSocketChannel.doWriteFileRegion()
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
@Override
protected long doWriteFileRegion(FileRegion region) throws Exception {
final long position = region.transferred();
return region.transferTo(javaChannel(), position);
}
这里调用 FileRegion.transferTo() 方法，使用 基于文件内存映射技术进行文件发送。
