ByteBuffer和netty.ByteBuf詳解
前言
數(shù)據(jù)序列化存儲，或者數(shù)據(jù)通過網(wǎng)絡(luò)傳輸時，會遇到不可避免將數(shù)據(jù)轉(zhuǎn)成字節(jié)數(shù)組的場景。字節(jié)數(shù)組的讀寫不會太難，但又有點繁瑣，為了避免重復(fù)造輪子，jdk推出了ByteBuffer來幫助我們操作字節(jié)數(shù)組；而netty是一款當(dāng)前流行的java網(wǎng)絡(luò)IO框架，它內(nèi)部定義了一個ByteBuf來管理字節(jié)數(shù)組，和ByteBuffer大同小異
ByteBuffer 零拷貝之MappedByteBuffer DirectByteBuffer堆外內(nèi)存回收機制 netty之ByteBuf
Buffer結(jié)構(gòu)
public abstract class Buffer {
//關(guān)系: mark <= position <= limit <= capacity
private int mark = -1;
private int position = 0;
private int limit;
private int capacity;
long address; // Used only by direct buffers，直接內(nèi)存的地址
mark：調(diào)用mark()方法的話，mark值將存儲當(dāng)前position的值，等下次調(diào)用reset()方法時，會設(shè)定position的值為之前的標(biāo)記值 position：是下一個要被讀寫的byte元素的下標(biāo)索引 limit：是緩沖區(qū)中第一個不能讀寫的元素的數(shù)組下標(biāo)索引，也可以認(rèn)為是緩沖區(qū)中實際元素的數(shù)量 capacity：是緩沖區(qū)能夠容納元素的最大數(shù)量，這個值在緩沖區(qū)創(chuàng)建時被設(shè)定，而且不能夠改變
Buffer.API
Buffer(int mark, int pos, int lim, int cap)
//Buffer創(chuàng)建時設(shè)置的最大數(shù)組容量值
public final int capacity()
//當(dāng)前指針的位置
public final int position()
//限制可讀寫大小
public final Buffer limit(int newLimit)
//標(biāo)記當(dāng)前position的位置
public final Buffer mark()
//配合mark使用，position成之前mark()標(biāo)志的位置。先前沒調(diào)用mark則報錯
public final Buffer reset()
//寫->讀模式翻轉(zhuǎn)，單向的
//position變成了初值位置0，而limit變成了寫模式下position位置
public final Buffer flip()
//重置position指針位置為0，mark為-1；相對flip方法是limit不變
public final Buffer rewind() //復(fù)位
//和rewind一樣，多出一步是limit會被設(shè)置成capacity
public final Buffer clear()
//返回剩余未讀字節(jié)數(shù)
public final int remaining()
ByteBuffer結(jié)構(gòu)
public abstract class ByteBuffer extends Buffer
implements Comparable<ByteBuffer>{
final byte[] hb; //僅限堆內(nèi)內(nèi)存使用
final int offset;
boolean isReadOnly;
ByteBuffer.API
//申請堆外內(nèi)存
public static ByteBuffer allocateDirect(int capacity)
//申請堆內(nèi)內(nèi)存
public static ByteBuffer allocate(int capacity)
//原始字節(jié)包裝成ByteBuffer
public static ByteBuffer wrap(byte[] array, int offset, int length)
//原始字節(jié)包裝成ByteBuffer
public static ByteBuffer wrap(byte[] array)
//創(chuàng)建共享此緩沖區(qū)內(nèi)容的新字節(jié)緩沖區(qū)
public abstract ByteBuffer duplicate()
//分片，創(chuàng)建一個新的字節(jié)緩沖區(qū)
//新ByteBuffer的開始位置是此緩沖區(qū)的當(dāng)前位置position
public abstract ByteBuffer slice()
//獲取字節(jié)內(nèi)容
public abstract byte get()
//從ByteBuffer偏移offset的位置，獲取length長的字節(jié)數(shù)組，然后返回當(dāng)前ByteBuffer對象
public ByteBuffer get(byte[] dst, int offset, int length)
//設(shè)置byte內(nèi)存
public abstract ByteBuffer put(byte b);
//以offset為起始位置設(shè)置length長src的內(nèi)容，并返回當(dāng)前ByteBuffer對象
public ByteBuffer put(byte[] src, int offset, int length長)
//將沒有讀完的數(shù)據(jù)移到到緩沖區(qū)的初始位置，position設(shè)置為最后一沒讀字節(jié)數(shù)據(jù)的下個索引，limit重置為為capacity
//讀->寫模式，相當(dāng)于flip的反向操作
public abstract ByteBuffer compact()
//是否是直接內(nèi)存
public abstract boolean isDirect()
ByteBuffer bf = ByteBuffer.allocate(10);`，創(chuàng)建大小為10的ByteBuffer對象 
寫入數(shù)據(jù)
ByteBuffer buf ByteBuffer.allocate(10);
buf.put("csc".getBytes());
調(diào)用flip轉(zhuǎn)換緩沖區(qū)為讀模式; buf.flip();
讀取緩沖區(qū)中到內(nèi)容：get(); System.out.println((char) buf.get());
零拷貝之MappedByteBuffer
共享內(nèi)存映射文件，對應(yīng)的ByteBuffer子操作類，MappedByteBuffer是基于mmap實現(xiàn)的。關(guān)于零拷貝的mmap的底層原理可以看看:框架篇：小白也能秒懂的Linux零拷貝原理[1]。MappedByteBuffer需要FileChannel調(diào)用本地map函數(shù)映射。C++代碼可以查閱下FileChannelImpl.c-Java_sun_nio_ch_FileChannelImpl_map0方法[2] 使用MappedByteBuffer和文件映射，其讀寫可以減少內(nèi)存拷貝次數(shù)
FileChannel readChannel = FileChannel.open(Paths.get("./cscw.txt"), StandardOpenOption.READ);
MappedByteBuffer data = readChannel.map(FileChannel.MapMode.READ_ONLY, 0, 1024 * 1024 * 40);
DirectByteBuffer堆外內(nèi)存回收機制Cleaner
下面我們看看直接內(nèi)存的回收機制（java8）；DirectByteBuffer內(nèi)部存在一個Cleaner對象，并且委托內(nèi)部類Deallocator對象進(jìn)行內(nèi)存回收
class DirectByteBuffer extends MappedByteBuffer implements DirectBuffer
{
//構(gòu)造函數(shù)
DirectByteBuffer(int cap) {
.... //內(nèi)存分配
cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
...
}
private static class Deallocator implements Runnable{
...
public void run() {
if (address == 0) {
// Paranoia
return;
}
unsafe.freeMemory(address); //回收內(nèi)存
address = 0;
Bits.unreserveMemory(size, capacity);
}
}
細(xì)看下Cleaner，繼承于PhantomReference，并且在 public void clean()方法會調(diào)用Deallocator進(jìn)行清除操作
public class Cleaner extends PhantomReference<Object> {
//如果DirectByteBuffer對象被回收，相應(yīng)的Cleaner會被放入dummyQueue隊列
private static final ReferenceQueue<Object> dummyQueue = new ReferenceQueue();
//構(gòu)造函數(shù)
public static Cleaner create(Object var0, Runnable var1) {
return var1 == null ? null : add(new Cleaner(var0, var1));
}
private Cleaner(Object var1, Runnable var2) {
super(var1, dummyQueue);
this.thunk = var2;
}
private final Runnable thunk;
public void clean() {
if (remove(this)) {
try {
this.thunk.run();
} catch (final Throwable var2) {
....
在Reference內(nèi)部存在一個守護(hù)線程，循環(huán)獲取Reference，并判斷是否Cleaner對象，如果是則調(diào)用其clean方法
public abstract class Reference<T>
static {
ThreadGroup tg = Thread.currentThread().getThreadGroup();
for (ThreadGroup tgn = tg; tgn != null; g = tgn, tgn = tg.getParent());
Thread handler = new ReferenceHandler(tg, "Reference Handler");
...
handler.setDaemon(true);
handler.start();
...
}
...
//內(nèi)部類調(diào)用 tryHandlePending
private static class ReferenceHandler extends Thread {
public void run() {
while (true) {
tryHandlePending(true);
}
}
...
static boolean tryHandlePending(boolean waitForNotify) {
Cleaner c;
.... //從鏈表獲取對象被回收的引用
// 判斷Reference是否Cleaner，如果是則調(diào)用其clean方法
if (c != null) {
c.clean(); //調(diào)用Cleaner的clean方法
return true;
}
ReferenceQueue<? super Object> q = r.queue;
if (q != ReferenceQueue.NULL) q.enqueue(r);
return true;
netty之ByteBuf
ByteBuf原理 Bytebuf通過兩個位置指針來協(xié)助緩沖區(qū)的讀寫操作，分別是readIndex和writerIndex
* +-------------------+------------------+------------------+
* | discardable bytes | readable bytes | writable bytes |
* | | (CONTENT) | |
* +-------------------+------------------+------------------+
* | | | |
* 0 <= readerIndex <= writerIndex <= capacity
ByteBuf.API
//獲取ByteBuf分配器
public abstract ByteBufAllocator alloc()
//丟棄可讀字節(jié)
public abstract ByteBuf discardReadBytes()
//返回讀指針
public abstract int readerIndex()
//設(shè)置讀指針
public abstract ByteBuf readerIndex(int readerIndex);
//標(biāo)志當(dāng)前讀指針位置，配合resetReaderIndex使用
public abstract ByteBuf markReaderIndex()
public abstract ByteBuf resetReaderIndex()
//返回可讀字節(jié)數(shù)
public abstract int readableBytes()
//返回寫指針
public abstract int writerIndex()
//設(shè)置寫指針
public abstract ByteBuf writerIndex(int writerIndex);
//標(biāo)志當(dāng)前寫指針位置，配合resetWriterIndex使用
public abstract ByteBuf markWriterIndex()
public abstract ByteBuf resetWriterIndex()
//返回可寫字節(jié)數(shù)
public abstract int writableBytes()
public abstract ByteBuf clear();
//設(shè)置讀寫指針
public abstract ByteBuf setIndex(int readerIndex, int writerIndex)
//指針跳過length
public abstract ByteBuf skipBytes(int length)
//以當(dāng)前位置切分ByteBuf todo
public abstract ByteBuf slice();
//切割起始位置為index，長度為length的ByteBuf todo
public abstract ByteBuf slice(int index, int length);
//Returns a copy of this buffer's readable bytes. //復(fù)制ByteBuf todo
public abstract ByteBuf copy()
//是否可讀
public abstract boolean isReadable()
//是否可寫
public abstract boolean isWritable()
//字節(jié)編碼順序
public abstract ByteOrder order()
//是否在直接內(nèi)存申請的ByteBuf
public abstract boolean isDirect()
//轉(zhuǎn)為jdk.NIO的ByteBuffer類
public abstract ByteBuffer nioBuffer()
使用示例
public static void main(String[] args) {
//分配大小為10的內(nèi)存
ByteBuf buf = Unpooled.buffer(10);
//寫入
buf.writeBytes("csc".getBytes());
//讀取
byte[] b = new byte[3];
buf.readBytes(b);
System.out.println(new String(b));
System.out.println(buf.writerIndex());
System.out.println(buf.readerIndex());
}
－－－－result－－－－
csc
3
3
ByteBuf初始化時，readIndex和writerIndex等于0，調(diào)用 writeXXX()方法寫入數(shù)據(jù)，writerIndex會增加(setXXX方法無作用)；調(diào)用readXXX()方法讀取數(shù)據(jù)，則會使readIndex增加(getXXX方法無作用)，但不會超過writerIndex在讀取數(shù)據(jù)之后，0-readIndex之間的byte數(shù)據(jù)被視為discard，調(diào)用discardReadBytes()，釋放這部分空間，作用類似于ByteBuffer的compact方法
參考文章
java.nio.ByteBuffer用法小結(jié)[3] Netty系列-一分鐘了解ByteBuffer和ByteBuf結(jié)構(gòu)[4] Netty之有效規(guī)避內(nèi)存泄漏[5]
Reference
框架篇：小白也能秒懂的Linux零拷貝原理: https://juejin.cn/post/6887469050515947528
[2]FileChannelImpl.c-Java_sun_nio_ch_FileChannelImpl_map0方法: https://github.com/unofficial-openjdk/openjdk/blob/jdk8u/jdk8u/jdk/src/solaris/native/sun/nio/ch/FileChannelImpl.c
[3]java.nio.ByteBuffer用法小結(jié): https://blog.csdn.net/mrliuzhao/article/details/89453082
[4]Netty系列-一分鐘了解ByteBuffer和ByteBuf結(jié)構(gòu): https://www.jianshu.com/p/3930150bf7f0
[5]Netty之有效規(guī)避內(nèi)存泄漏: https://www.jianshu.com/p/cec977b28079?from=timeline
評論
圖片
表情