gzx0 2020-06-14
4.2 Netty 高性能之道
与 Socket 类和 ServerSocket 类相对应,NIO 也提供了 SocketChannel 和 ServerSocketChannel 两种不同的套接字通
客户端通信序列图如下:
public final void read() { ChannelConfig config = AbstractNioByteChannel.this.config(); ChannelPipeline pipeline = AbstractNioByteChannel.this.pipeline(); ByteBufAllocator allocator = config.getAllocator(); Handle allocHandle = this.recvBufAllocHandle(); allocHandle.reset(config); ByteBuf byteBuf = null; boolean close = false; try { do { byteBuf = allocHandle.allocate(allocator); allocHandle.lastBytesRead(AbstractNioByteChannel.this.doReadBytes(byteBuf)); if (allocHandle.lastBytesRead() <= 0) { byteBuf.release(); byteBuf = null; close = allocHandle.lastBytesRead() < 0; break; } allocHandle.incMessagesRead(1); AbstractNioByteChannel.this.readPending = false; pipeline.fireChannelRead(byteBuf); byteBuf = null; } while(allocHandle.continueReading()); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); if (close) { this.closeOnRead(pipeline); } } catch (Throwable var11) { this.handleReadException(pipeline, byteBuf, var11, close, allocHandle); } finally { if (!AbstractNioByteChannel.this.readPending && !config.isAutoRead()) { this.removeReadOp(); } } } }
public ByteBuf allocate(ByteBufAllocator alloc) { return alloc.ioBuffer(guess()); } 相当于是每次循环读取一次消息,就通过 ByteBufferAllocator 的 ioBuffer 方法获取 ByteBuf 对象,下面继续看它的接口定义 public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> { ... } 当进行 Socket IO 读写的时候,为了避免从堆内存拷贝一份副本到直接内存,Netty 的 ByteBuf 分配器直接创建非堆内 存避免缓冲区的二次拷贝,通过“零拷贝”来提升读写性能。 下面我们继续看第二种“零拷贝”的实现 CompositeByteBuf,它对外将多个 ByteBuf 封装成一个 ByteBuf,对外提供统 一封装后的 ByteBuf 接口,它的类定义如下: 通过继承关系我们可以看出 CompositeByteBuf 实际就是个 ByteBuf 的包装器,它将多个 ByteBuf 组合成一个集合,然 后对外提供统一的 ByteBuf 接口,相关定义如下: private static final ByteBuffer EMPTY_NIO_BUFFER = Unpooled.EMPTY_BUFFER.nioBuffer(); private static final Iterator<ByteBuf> EMPTY_ITERATOR = Collections.<ByteBuf>emptyList().iterator(); private final ByteBufAllocator alloc; private final boolean direct; private final List<Component> components; private final int maxNumComponents; private boolean freed; 添加 ByteBuf,不需要做内存拷贝,相关代码如下: private int addComponents0(boolean increaseWriterIndex, int cIndex, ByteBuf[] buffers, int offset, int len) { ObjectUtil.checkNotNull(buffers, "buffers"); int i = offset; boolean var16 = false; int var20; try { var16 = true; this.checkComponentIndex(cIndex); while(true) { if (i < len) { ByteBuf b = buffers[i++]; if (b != null) { cIndex = this.addComponent0(increaseWriterIndex, cIndex, b) + 1; int size = this.components.size(); if (cIndex > size) { cIndex = size; } continue; } } var20 = cIndex; var16 = false; break; } } finally { if (var16) { while(true) { if (i >= len) { ; } else { ByteBuf b = buffers[i]; if (b != null) { try { b.release(); } catch (Throwable var17) { ; } } ++i; } } } } for(; i < len; ++i) { ByteBuf b = buffers[i]; if (b != null) { try { b.release(); } catch (Throwable var18) { ; } } } return var20; } 最后,我们看下文件传输的“零拷贝”: public long transferTo(WritableByteChannel target, long position) throws IOException { long count = this.count - position; if (count >= 0L && position >= 0L) { if (count == 0L) { return 0L; } else if (this.refCnt() == 0) { throw new IllegalReferenceCountException(0); } else { this.open(); long written = this.file.transferTo(this.position + position, count, target); if (written > 0L) { this.transferred += written; } return written; } } else { throw new IllegalArgumentException("position out of range: " + position + " (expected: 0 - " + (this.count - 1L) + ‘)‘); } } Netty 文件传输 DefaultFileRegion 通过 transferTo()方法将文件发送到目标 Channel 中,下面重点看 FileChannel 的 transferTo()方法,它的 API DOC 说明如下: /** * Transfers bytes from this channel‘s file to the given writable byte * channel. * * <p> An attempt is made to read up to <tt>count</tt> bytes starting at * the given <tt>position</tt> in this channel‘s file and write them to the * target channel. An invocation of this method may or may not transfer * all of the requested bytes; whether or not it does so depends upon the * natures and states of the channels. Fewer than the requested number of * bytes are transferred if this channel‘s file contains fewer than * <tt>count</tt> bytes starting at the given <tt>position</tt>, or if the * target channel is non-blocking and it has fewer than <tt>count</tt> * bytes free in its output buffer. * * <p> This method does not modify this channel‘s position. If the given * position is greater than the file‘s current size then no bytes are * transferred. If the target channel has a position then bytes are * written starting at that position and then the position is incremented * by the number of bytes written. * * <p> This method is potentially much more efficient than a simple loop * that reads from this channel and writes to the target channel. Many * operating systems can transfer bytes directly from the filesystem cache * to the target channel without actually copying them. </p> *
public abstract long transferTo(long position, long count, WritableByteChannel target) throws IOException;
Netty 提供了多种内存管理策略,通过在启动辅助类中配置相关参数,可以实现差异化的定制。
package com.lf.io.nio; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; public class BufTest { public static void main(String[] args) { directBuf(); } public static void directBuf() { final byte[] CONTENT = new byte[1024]; int loop = 1800000; long startTime = System.currentTimeMillis(); ByteBuf poolBuffer = null; for (int i = 0; i < loop; i++) { poolBuffer = PooledByteBufAllocator.DEFAULT.directBuffer(1024); poolBuffer.writeBytes(CONTENT); poolBuffer.release(); } long endTime = System.currentTimeMillis(); System.out.println("内存池分配缓冲区耗时" + (endTime - startTime) + "ms."); long startTime2 = System.currentTimeMillis(); ByteBuf buffer = null; for (int i = 0; i < loop; i++) { buffer = Unpooled.directBuffer(1024); buffer.writeBytes(CONTENT); } endTime = System.currentTimeMillis(); System.out.println("非内存池分配缓冲区耗时" + (endTime - startTime2) + "ms."); } }
public ByteBuf directBuffer(int initialCapacity, int maxCapacity) { if (initialCapacity == 0 && maxCapacity == 0) { return this.emptyBuf; } else { validate(initialCapacity, maxCapacity); return this.newDirectBuffer(initialCapacity, maxCapacity); } }
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) { PoolThreadCache cache = (PoolThreadCache)this.threadCache.get(); PoolArena<ByteBuffer> directArena = cache.directArena; Object buf; if (directArena != null) { buf = directArena.allocate(cache, initialCapacity, maxCapacity); } else if (PlatformDependent.hasUnsafe()) { buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity); } else { buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity); } return toLeakAwareBuffer((ByteBuf)buf); }
PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) { PooledByteBuf<T> buf = this.newByteBuf(maxCapacity); this.allocate(cache, buf, reqCapacity); return buf; }
因此重点分析 DirectArena 的实现:如果没有开启使用 sun 的 unsafe,则
protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) { return (PooledByteBuf)(HAS_UNSAFE ? PooledUnsafeDirectByteBuf.newInstance(maxCapacity) : PooledDirectByteBuf.newInstance(maxCapacity)); }
static PooledDirectByteBuf newInstance(int maxCapacity) { PooledDirectByteBuf buf = (PooledDirectByteBuf)RECYCLER.get(); buf.reuse(maxCapacity); return buf; }
由于 Reactor 模式使用的是异步非阻塞 IO,所有的 IO 操作都不会导致阻塞,理论上一个线程可以独立处理所有 IO 相
Reactor 多线程模型的特点:
利用主从 NIO 线程模型,可以解决 1 个服务端监听线程无法有效处理所有客户端连接的性能不足问题。因此,在 Netty
Netty 的 NioEventLoop 读取到消息之后,直接调用 ChannelPipeline 的 fireChannelRead(Object msg),只要用户不主
从上图可以看出,Protobuf 序列化后的码流只有 Java 序列化的 1/4 左右。正是由于 Java 原生序列化性能表现太差,