Java高并發16-LongAdder類原始碼決議（上）

一、復習

• AtomicLong類

二、LongAdder原始碼分析

1.繼承與實作關系

• LongAdder類繼承自Striped64這個類，同時實作了Serializable介面

abtract class Strped64{
    transient volatile Cell[] cells;
    transient volatile long base;
    transient volatile int cellsBusy;
    ........
}

• 繼承了三個變數，其中cells變數就是Cell實體陣列，用于存放共享變數，base值是一個基礎值，cellsBusy用來實作自旋鎖，其值只有0和1
• 當創建Cell元素，擴容Cell陣列或者初始化Cell陣列的時候，使用CAS操作該變數來保證同時只有一個執行緒可以進行其中之一的操作，

2.關鍵字transient

• 講這個關鍵字之前不得不提一下Seriable這個介面，我們以LongAdder這里類為例，由于它實作了序列化這個介面，也就意味著這個類可以序列化，并且保存到磁盤中，而不僅僅是只在記憶體中使用，然后對于有一些資料是不應該序列化的，比如密碼，關鍵資訊等等，這些資料保存到變數之后，然后被transient這個關鍵字修飾，那么將來序列化這個實體的時候，就不會保存這個變數，保證資訊安全，

3.Cell原始碼

package com.ruigege.AtomicOperationClass4;
import sun.misc.Unsafe;

@sun.misc.Contended public class LongAdderTest {
	
	volatile long value;
	
	public LongAdderTest(long value) {
		this.value = value;
	}

	private static final Unsafe unsafe;
	private static final long valueOffset;
	static {
		try {
			unsafe = Unsafe.getUnsafe();
			Class<?> ak = Cell.class;
			valueOffset = unsafe.objectFieldOffset(ak.getDeclaredField("value"));
		}catch(Exception e) {
			throw new Error(e);
		}
	}
	
	final long cas(long cmp,long val) {
		return unsafe.compareAndSwapLong(this,valueOffset,cmp,val);
	}
}

• cas函式使用了CAS操作，保證了當前執行緒更新時被分配的Cell元素中的value值的原子性，

	public long sum() {
		Cell[] as = cells;
		long sum = base;
		if(as != null) {
			for (int i=0;i<as.length;i++) {
				if(as[i] != null) {
					sum += as[i].value;
				}
			}
		}
	}

• 上面這個函式就是求總值，但是由于沒有加鎖，所以在計算的程序中，有可能Cell實體的值變化了，所以得到的值可能不準，

	public void reset() {
		Cell[] as = cells;
		base = 0L;
		if(as != null) {
			for (int i=0;i<as.length;i++) {
				if(as[i] != null) {
					as[i].value = 0L;
				}
			}
		}
	}

• 該函式用于重置該LongAdder實體

	public long sunThenReset() {
		Cell[] as = cells;
		int sum = base;
		if(as != null) {
			for (int i=0;i<as.length;i++) {
				if(as[i] != null) {
					sum += as[i];
					as[i] = 0L;
				}
			}
		}
	}

• 先求和，然后Cell陣列置空，但是這個方法沒有加鎖，容易造成資料不一致，

	public void add(long x) {
		Cell[] as;
		long b,v;
		int m;
		Cell a;
		if((as = cells) != null) || !caseBase(b=base,b+x)){
			//如果as是一個空陣列，也就是第一個是false，那么會去判斷第二個，第二個函式，其實就是一個CAS操作，
			//空陣列就意味著沒有初始化陣列，這個LongAdder實體，里面就一個base變數，也就是說并發量太小了，不足以
			//初始化陣列，這時這個實體就是一個AtomicLong實體，沒有區別，第二個判斷就是給base賦值為最新的增長量
			//如果cas失敗了，那么就進入到下面的判斷陳述句中
			boolean uncontended = true;
			if(as==null || (m = as.length-1)<0 || (a = as[getProbe() & m]) == null || !(uncontended=a.cas(v=a.value,v+x))){
				/**
				 * 四個判斷條件，逐步深入
				 * （1）如果陣列為空，繼續執行判斷體內；不為空，看第二個條件
				 * （2）如果陣列中元素為0，繼續執行判斷體內；不少于1個，看第三個條件
				 * （3）getProbe()函式用于獲取獲取當前執行緒中變數threadLocalRandomProbe的值，這個值一開始為0，在接下
				 * 來的判斷體中會進行初始化，并且當前執行緒通過分配的Cell元素的cas函式來保證對Cell元素value值更新的原子性，
				 * 這第三個就是為了找到一個陣列中的Cell變數，我們暫且看到是0&m,那就是0了，也就是啊as[0]賦值給a，并且如果是空的，則直接
				 * 進入到判斷執行體；如果不為空，接著看第四個判斷條件
				 * （4）第四個判斷條件就是剛才的那個a使用CAS操作來更新值，如果更新失敗了就進入到執行體
				 */
				longAccumulate(x,null,uncontended);
			}
		}
	}
	
	final boolean casBase(long cmp,long val) {
		return UNSAFE.compareAndSwapLong(this,BASE,cmp,val);
	}

• 以上都是add函式，內部具體實作邏輯，

final void longAccumulate(long x,LongBinaryOperator fn,boolean wasUncontended) {
		//初始化當前執行緒的變數threadLocalRandomProbe的值
		int h;
		if(h = getProbe() == 0) {
			ThreadLocalRandom.current();
			h = getProbe();
			wasUncontended = true;
		}
		boolean collide = false;
		for(;;) {
			Cell[] as;Cell a;int n;long v;
			if((as=cells) != null && (n=as.length)>0) {
				if((a=as[(n-1) &h]) == null) {
					if(cellsBusy == 0) {
						Cell r = new Cell(x);
						if(cellsBusy == 0 && casCellsBusy()) {
							boolean created = false;
						}
						try {
							Cell[] rs;int m,j;
							if((rs=cells) != null && (m=rs.length)>0 && rs[j = (m-1) &h] == null) {
								rs[j] = r;
								created = true;
							}
						}finally {
							cellsBusy = 0;
						}
						if(created) {
							break;
						}
						continue;
					}
				}
				collide = false;
			}else if(!wasUntended) {
				wasUncontended = true;
			}else if(a.cas(v=a.value,((fn==null)?v+x : fn.applyAsLong(v,x)))) {
				break;
			}else if(n >= NCPU || cells != as) {
				collide = false;
			}else if(!collide) {
				collide = ture;
			}else if(cellsBusy == 0 && casCellsBusy()) {
				try {
					if(cells == as) {
						Cell[] rs = new Cell[n<<1];
						for(int i=0;i<n;++i) {
							rs[i] = as[i];
						}
						cells = rs;
					}
				}finally {
					cellsBusy = 0;
				}
			}
		}
	}

三、原始碼：

• 所在包：com.ruigege.AtomicOperationClass4
• https://github.com/ruigege66/ConcurrentJava
• CSDN：https://blog.csdn.net/weixin_44630050
• 博客園：https://www.cnblogs.com/ruigege0000/
• 歡迎關注微信公眾號：傅里葉變換，個人賬號，僅用于技術交流