Netty源碼分析之ByteBuf引用計數
引用計數是一種常用的內存管理機制，是指將資源的被引用次數保存起來，當被引用次數變為零時就將其釋放的過程。Netty在4.x版本開始使用引用計數機制進行部分對象的管理，其實現思路并不是特別復雜，它主要涉及跟蹤某個對象被引用的次數。在Netty具體代碼中需要通過引用計數進行內存管理的對象，會基于ReferenceCounted接口實現，其中引用計數大于0時則代表該對象被引用不會釋放，當引用計數減少到0時，該對象就會被釋放。通過引用計數機制，Netty可以很好的實現內存管理，引用計數減少到0時要么直接釋放內存，要么放回內存池中重復利用。
1、基本示例
下面先通過一個簡單示例看下Netty中引用計數機制的使用
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf recvBuffer = (ByteBuf) msg;// 申請ByteBuf 需要主動釋放 if(recvBuffer.isDirect()){ System.err.println(true); } PooledByteBufAllocator allocator = new PooledByteBufAllocator(true); ByteBuf sendBuffer = allocator.buffer();//申請池化直接內存 System.err.println("sendBuffer的引用計數:"+sendBuffer.refCnt()); sendBuffer.retain(); System.err.println("sendBuffer的引用計數:"+sendBuffer.refCnt()); sendBuffer.release(); System.err.println("sendBuffer的引用計數:"+sendBuffer.refCnt());
try { byte[] bytesReady = new byte[recvBuffer.readableBytes()]; recvBuffer.readBytes(bytesReady); System.out.println("channelRead收到數據："+ BytesUtils.toHexString(bytesReady)); byte[] sendBytes = new byte[] {0x7E,0x01,0x02,0x7e}; sendBuffer.writeBytes(sendBytes); ctx.writeAndFlush(sendBuffer); System.err.println("sendBuffer的引用計數:"+sendBuffer.refCnt()); }catch (Exception e) { // TODO: handle exception System.err.println(e.getMessage()); }finally { System.err.println("recvBuffer的引用計數:"+recvBuffer.refCnt()); recvBuffer.release(); //此處需要釋放 System.err.println("recvBuffer的引用計數:"+recvBuffer.refCnt()); } }
輸出結果如下，通過示例可以看出retain方法會增加計數引用，release方法會減少計數引用
true sendBuffer的引用計數:1 sendBuffer的引用計數:2 sendBuffer的引用計數:1 sendBuffer的引用計數:0 recvBuffer的引用計數:1 recvBuffer的引用計數:0
AbstractReferenceCountedByteBuf實現了對ByteBuf的內存管理，以實現內存的回收、釋放或者重復利用 ，AbstractReferenceCountedByteBuf的繼承實現關系如下圖所示
2、ReferenceCounted接口定義
首先是ReferenceCounted接口的定義
public interface ReferenceCounted { /** * Returns the reference count of this object. If {@code 0}, it means this object has been deallocated. * 返回對象的引用計數 */ int refCnt(); /** * Increases the reference count by {@code 1}. * 增加引用計數 */ ReferenceCounted retain(); /** * Increases the reference count by the specified {@code increment}. * 引用計數增加指定值 */ ReferenceCounted retain(int increment); /** * Records the current access location of this object for debugging purposes. * If this object is determined to be leaked, the information recorded by this operation will be provided to you * via {@link ResourceLeakDetector}. This method is a shortcut to {@link #touch(Object) touch(null)}. * 記錄該對象的當前訪問位置，用于調試。 * 如果確定該對象被泄露，將提供此操作記錄的信息給您 */ ReferenceCounted touch(); /** * Records the current access location of this object with an additional arbitrary information for debugging * purposes. If this object is determined to be leaked, the information recorded by this operation will be * provided to you via {@link ResourceLeakDetector}. * 記錄該對象的當前訪問位置，附加信息用于調試。 * 如果確定該對象被泄露，將提供此操作記錄的信息給您 */ ReferenceCounted touch(Object hint); /** * Decreases the reference count by {@code 1} and deallocates this object if the reference count reaches at * {@code 0}. * * @return {@code true} if and only if the reference count became {@code 0} and this object has been deallocated * 引用計數減少，如果計數變為了0，則釋放對象資源 * 如果對象資源被釋放，則返回true，否則返回false */ boolean release(); /** * Decreases the reference count by the specified {@code decrement} and deallocates this object if the reference * count reaches at {@code 0}. * * @return {@code true} if and only if the reference count became {@code 0} and this object has been deallocated * 引用計數-指定值，如果計數變為了0，則釋放對象資源或交回到對象池 * 如果對象資源被釋放，則返回true，否則返回false */ boolean release(int decrement); }
3、AbstractReferenceCountedByteBuf源碼分析
AbstractReferenceCountedByteBuf對ReferenceCounted進行了具體實現，retain與release兩個方法通過CAS方式對引用計數refcnt進行操作，下面對其源碼進行簡單分析
初始化
引用計數初始值refCnt 使用關鍵字volatile修飾，保證線程的可見性，同時使用偶數，引用增加通過位移操作實現，提高運算效率。
采用 AtomicIntegerFieldUpdater 對象，通過CAS方式更新refCnt，以實現線程安全，避免加鎖，提高效率。
private static final long REFCNT_FIELD_OFFSET; //采用 AtomicIntegerFieldUpdater 對象，CAS方式更新refCnt private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater = AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt"); //refCnt 實際值為偶數，采用位移操作提高效率 // even => "real" refcount is (refCnt >>> 1); odd => "real" refcount is 0 @SuppressWarnings("unused") private volatile int refCnt = 2;
retain操作
上面示例中每調用一次retain方法，引用計數就會累加一次，我們看下源碼中retain的具體實現
public ByteBuf retain() { return retain0(1); } @Override public ByteBuf retain(int increment) { return retain0(checkPositive(increment, "increment")); } //計數器增值操作 private ByteBuf retain0(final int increment) { // all changes to the raw count are 2x the "real" change int adjustedIncrement = increment << 1; // overflow OK here 真正的計數都是2倍遞增 int oldRef = refCntUpdater.getAndAdd(this, adjustedIncrement); //通過CAS方式遞增并獲取原值 if ((oldRef & 1) != 0) {//判斷奇偶，正常情況這里應該都是偶數 throw new IllegalReferenceCountException(0, increment); } // don't pass 0! 如果計數小于等于0，以及整型范圍越界（0x7fffffff+1）拋出異常 if ((oldRef <= 0 && oldRef + adjustedIncrement >= 0) || (oldRef >= 0 && oldRef + adjustedIncrement < oldRef)) { // overflow case refCntUpdater.getAndAdd(this, -adjustedIncrement); throw new IllegalReferenceCountException(realRefCnt(oldRef), increment); } return this; }
release操作
通過調用release方法，對引用計數做減值操作，源碼中release的具體實現要注意的是由于引用計數以2倍遞增，所以引用次數= 引用計數/2，當decrement=refcnt/2 也就是引用次數=釋放次數時，代表ByteBuf不再被引用，執行內存釋放或放回內存池的操作。
//計數器減值操作 private boolean release0(int decrement) { int rawCnt = nonVolatileRawCnt(), realCnt = toLiveRealCnt(rawCnt, decrement); //對計數器進行除以2操作,也就是引用次數 /** * /這里如注意 你傳入的減值參數decrement = realCnt 時 等同于 引用次數=釋放次數，直接進行釋放操作 */ if (decrement == realCnt) { if (refCntUpdater.compareAndSet(this, rawCnt, 1)) { //CAS方式置為1 deallocate();//內存釋放或放回內存池 return true; } return retryRelease0(decrement);//進入具體操作 } return releaseNonFinal0(decrement, rawCnt, realCnt); } private boolean releaseNonFinal0(int decrement, int rawCnt, int realCnt) { //如果decrement 小于 realCnt，通過CAS方式減去decrement*2 if (decrement < realCnt // all changes to the raw count are 2x the "real" change && refCntUpdater.compareAndSet(this, rawCnt, rawCnt - (decrement << 1))) { return false; } return retryRelease0(decrement); } private boolean retryRelease0(int decrement) { for (;;) { int rawCnt = refCntUpdater.get(this), realCnt = toLiveRealCnt(rawCnt, decrement); if (decrement == realCnt) { if (refCntUpdater.compareAndSet(this, rawCnt, 1)) { deallocate(); return true; } } else if (decrement < realCnt) {//如果decrement 小于 realCnt，通過CAS方式減去decrement*2 // all changes to the raw count are 2x the "real" change if (refCntUpdater.compareAndSet(this, rawCnt, rawCnt - (decrement << 1))) { return false; } } else { throw new IllegalReferenceCountException(realCnt, -decrement); } Thread.yield(); // this benefits throughput under high contention } } /** * Like {@link #realRefCnt(int)} but throws if refCnt == 0 */ private static int toLiveRealCnt(int rawCnt, int decrement) { if ((rawCnt & 1) == 0) { return rawCnt >>> 1; } // odd rawCnt => already deallocated throw new IllegalReferenceCountException(0, -decrement); }
4、總結
以上我們圍繞AbstractReferenceCountedByteBuf對Netty引用計數的具體實現進行了分析，可以看到Netty在實現引用計數的同時，結合CAS、位移計算等方式，保證了運算效率和線程安全，在實際項目中我們遇到類似應用場景也都可以借鑒參考，如數據發送次數，商品剩余數量等計數場景的實現。希望本文對大家能有所幫助，其中如有不足與不正確的地方還望指正與海涵，十分感謝。
關注微信公眾號，查看更多技術文章。
浙公網安備 33010602011771號