JUC中的原子操作類及其原理
昨天簡單的看了看Unsafe的使用，今天我們看看JUC中的原子類是怎麼使用Unsafe的，以及分析一下其中的原理！
一.簡單使用AtomicLong
還記的上一篇部落格中我們使用了volatile關鍵字修飾了一個int型別的變數，然後兩個執行緒，分別對這個變數進行10000次+1操作，最後結果不是20000，現在我們改成AtomicLong之後，你會發現結果始終都是20000了！有興趣的可以試試，程式碼如下
package com.example.demo.study;
import java.util.concurrent.atomic.AtomicLong;
public class Study0127 {
//這是一個全域性變數,注意，這裡使用了一個原子類AtomicLong
public AtomicLong num = new AtomicLong();
//每次呼叫這個方法，都會對全域性變數加一操作，執行10000次
public void sum() {
for (int i = 0; i < 10000; i++) {
//使用了原子類的incrementAndGet方法，其實就是把num++封裝成原子操作
num.incrementAndGet();
System.out.println("當前num的值為num= "+ num);
}
}
public static void main(String[] args) throws InterruptedException {
Study0127 demo = new Study0127();
//下面就是新建兩個執行緒,分別呼叫一次sum方法
new Thread(new Runnable() {
@Override
public void run() {
demo.sum();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
demo.sum();
}
}).start();
}
}
二.走近AtomicLong類
在java中JDK 1.5之後，就出現了一個包，簡稱JUC併發包，全稱就是java.util .concurrent，其中我們應該聽說過一個類ConcurrentHashMap，這個map挺有意思的，有興趣可以看看原始碼！還有很多併發時候需要使用的類比如AtomicInteger，AtomicLong，AtomicBoolean等等，其實都差不多，這次我們就簡單看看AtomicLong，其他的幾個類也差不多
public class AtomicLong extends Number implements java.io.Serializable {
//獲取Unsafe物件，上篇部落格說了我們自己的類中不能使用這種方式的原因，但是官方的這個類為什麼可以這樣獲取呢？因為本類AtomicLong
//就是在rt.jar包下面，本類就是用Bootstrap類載入的，所以就可以用這種方式
private static final Unsafe unsafe = Unsafe.getUnsafe();
//value這個欄位的偏移量
private static final long valueOffset;
//判斷jvm是否支援long型別的CAS操作
static final boolean VM_SUPPORTS_LONG_CAS = VMSupportsCS8();
private static native boolean VMSupportsCS8();
static {
try {
valueOffset = unsafe.objectFieldOffset
(AtomicLong.class.getDeclaredField("value"));
} catch (Exception ex) { throw new Error(ex); }
}
//這裡用了volatile使的多執行緒下可見性，一定要分清楚原子性和可見性啊
private volatile long value;
//兩個構造器不多說
public AtomicLong() {
}
public AtomicLong(long initialValue) {
value = initialValue;
}
然後我們看看AtomicLong的+1操作，可以看到使用的還是unsafe這個類，只需要看看getAndAddLong方法就可以了
方法getAndAddLong裡面就是進行了CAS操作，可以看成如果同時有多個執行緒都呼叫incrementAndGet方法進行+1，那麼同一時間只有一個執行緒會去進行操作，而其他的會不斷的使用CAS去嘗試+1，每次嘗試的時候都會去主記憶體中獲取最新的值；
public final long getAndAddLong(Object o, long offset, long delta) {
long v;
do {
//當某個執行緒CAS成功就跳出這個迴圈，否則就一直在迴圈不斷的嘗試，這也是CAS和執行緒阻塞的區別
} while (!compareAndSwapLong(o, offset, v, v + delta));
return v;
}
//這個CAS方法看不到，c實現的
public final native boolean compareAndSwapLong(Object o, long offset,long expected,long x);
有興趣的可以看看AtomicLong的其他方法，很多都一樣，CAS是核心
三.CAS的不足以及認識LongAdder
從上面的例子中，我們可以知道在多執行緒下使用AtomicLong類的時候，同一個時刻使用那個共享變數的只能是一個執行緒，其他的執行緒都是在無限迴圈，這種迴圈也是需要消耗效能的，如果執行緒比較多，很多的執行緒都在各自的無限迴圈中，或者叫做多個執行緒都在自旋；每個執行緒都在自旋無數次真的是比較坑，比較消耗效能，我們可以想辦法自旋一定的次數，執行緒就結束運行了，有興趣的可以瞭解一下自旋鎖，其實就是這麼一個原理，很容易，哈哈哈！
在JDK8之後，提供了一個更好的類取代AtomicLong，那就是LongAdder，上面說過同一時間只有一個執行緒在使用那個共享變數，其他的執行緒都在自旋，那麼如果可以把這個共享變數拆開成多個部分，那麼是不是可以多個執行緒同時可以去操作呢？然後操作完之後再綜合起來，有點分治法的思想，分而治之，最後綜合起來。
那麼我們怎麼把那個共享變數拆成多個部分呢？
在LongAdder中是這樣處理的，把那個變數拆成一個base（這個是long型別的，初始值為0）和一個Cell（這個裡面封裝了一個long型別的值，初始值為0），每個執行緒只會去競爭很多Cell就行了，最後把多個Cell中的值和base累加起來就是最終結果；而且一個執行緒如果沒有競爭到Cell之後不會傻傻的自旋，直接想辦法去競爭下一個Cell；
下圖所示
四.簡單使用LongAdder
用法其實和AtomicLong差不多，有興趣的可以試試，最後的結果始終都是20000
package com.example.demo.study;
import java.util.concurrent.atomic.LongAdder;
public class Study0127 {
//這裡使用LongAdder類
public LongAdder num = new LongAdder();
//每次呼叫這個方法，都會對全域性變數加一操作，執行10000次
public void sum() {
for (int i = 0; i < 10000; i++) {
//LongAdder類的自增操作，相當於i++
num.increment();
System.out.println("當前num的值為num= "+ num);
}
}
public static void main(String[] args) throws InterruptedException {
Study0127 demo = new Study0127();
//下面就是新建兩個執行緒,分別呼叫一次sum方法
new Thread(new Runnable() {
@Override
public void run() {
demo.sum();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
demo.sum();
}
}).start();
}
}
五.走進LongAdder
從上面可以看到base只能是一個，而Cell可能有多個，而且Cell太多了也是很佔記憶體的，所以一開始的時候不會建立Cell，只有在需要時才建立，也叫做惰性載入。
我們可以知道LongAdder是繼承自Striped64這個類的
而Striped64類中有三個欄位，cells陣列用於存放多個Cell，一個是base不多說，還有一個cellsBusy用來實現自旋鎖，狀態只能是0或1(0表示Cell陣列沒有被初始化和擴容，也沒有正在建立Cell元素，反之則為1)，在建立Cell，初始化Cell陣列或者擴容Cell陣列的時候，就會用到這個欄位，保證同一時刻只有一個執行緒可以進行其中之一的操作。
1.我們簡單看看Cell的結構
從下面程式碼中可以很清楚的看到所謂的Cell就是對一個long型別變數的CAS操作
@sun.misc.Contended //這個註解的作用是為了避免偽共享，至於什麼偽共享，後面有機會再說說
static final class Cell {
//每個Cell類中就是這個宣告的變數後期要進行累加的
volatile long value;
//建構函式
Cell(long x) { value = x; }
//Unsafe物件
private static final sun.misc.Unsafe UNSAFE;
//value的偏移量
private static final long valueOffset;
//這個靜態程式碼塊中就是獲取Unsafe物件和偏移量的
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> ak = Cell.class;
valueOffset = UNSAFE.objectFieldOffset
(ak.getDeclaredField("value"));
} catch (Exception e) {
throw new Error(e);
}
}
//CAS操作，沒什麼好說的
final boolean cas(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
}
}
2.LongAdder類自增方法increment()
我們可以看到increment()方法其實就是呼叫了add方法，我們需要關注add方法幹了一些什麼；
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
//這裡的cells是父類Striped64中的，不為空的話就儲存到as中，然後呼叫casBase方法，就是CAS給base更新為base+x,也就是每次都新增x，
//在這裡由於add(1L)傳入的引數是1，也就是每次就是加一
//如果CAS成功之後就不說了，就完成操作了，如果CAS失敗，則進入到裡面去
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
//這個if判斷條件賊長，我們把這幾個條件分為1，2，3，4部分，前三部分都是用於決定執行緒應該訪問Cell陣列中哪一個Cell元素，最後一個部分用於更新Cell的值
//如果第1，2，3部分都不滿足，也就是說Cell陣列存在而且已經找到了確定的Cell元素，那就到第四部分，更新對應的Cell中的值（在Cell類中的cas方法已經看過了）
//如果第1，2，3部分滿足其中一個，那也就是說Cell陣列根本就不存在或者執行緒找不到對應的Cell，就執行longAccumulate方法
if (as == null || (m = as.length - 1) < 0 || (a = as[getProbe() & m]) == null || !(uncontended = a.cas(v = a.value, v + x)))
//後面仔細看看這個方法，這是對Cell陣列的初始化和擴容，很有意思
longAccumulate(x, null, uncontended);
}
}
//一個簡單的CAS操作
final boolean casBase(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
}
對於上面的，有興趣的可以看看是怎麼找到指定的Cell的，在上面的a = as[getProbe() & m]中，其中m=陣列的長度-1，其實這裡也是一個取餘的運算，而getProbe()這個方法是用於獲取當前執行緒的threadLocalRandomProb(當前本地執行緒探測值，初始值為0)，其實也就是一個隨機數啊，然後對陣列的長度取餘得到的就是對應的陣列的索引，首次呼叫這個方法是陣列的第一個元素，如果陣列的第一個元素為null，那麼就說明沒有找到對應的Cell；
對於取餘運算，舉個簡單的例子吧，我也有點忘記了，比如隨機數9要對4進行取餘，我們可以9&(4-1)=9&3=1001&0011=1，利用位運算取餘瞭解一下；
現在我們重點看看longAccumulate方法，程式碼比較長，單獨提取出來看看
3.longAccumulate方法
//此方法是對Cell陣列的初始化和擴容，注意有個形參LongBinaryOperator，這是JDK8新增的函數語言程式設計的介面，函式簽名為(T,T)->T,這裡傳進來的是null
final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
int h;
//初始化當前執行緒的threadLocalRandomProbd的值，也就是生成一個隨機數
if ((h = getProbe()) == 0) {
ThreadLocalRandom.current(); // force initialization
h = getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
for (;;) {
Cell[] as; Cell a; int n; long v;
//這裡表示初始化完畢了
if ((as = cells) != null && (n = as.length) > 0) {
//這裡表示隨機數和陣列大小取餘，得到的結果就是當前執行緒要匹配到的Cell元素的索引，如果索引對應在Cell陣列中的元素為null，就新增一個Cell物件扔進去
if ((a = as[(n - 1) & h]) == null) {
//cellsBusy為0，表示當前Cell沒有進行擴容、初始化操作或者正在建立Cell等操作，那麼當前執行緒可以對這個Cell陣列為所欲為
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create
//看下面的Cell陣列初始化，說的很清楚，主要是設定cellsBusy為1，然後將當前執行緒匹配到的Cell設定為新建立的Cell物件
if (cellsBusy == 0 && casCellsBusy()) {
boolean created = false;
try { // Recheck under lock
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，表示此時其他執行緒又可以對Cell陣列為所欲為了
cellsBusy = 0;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
//Cell元素存在就執行CAS更新Cell中的值，這裡fn是形參為null
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
//當Cell陣列元素個數大於CPU的個數
else if (n >= NCPU || cells != as)
collide = false; // At max size or stale
//是否有衝突
else if (!collide)
collide = true;
//擴容Cell陣列，和上面兩個else if一起看
//如果當前Cell陣列元素沒有達到CPU個數而且有衝突就新型擴容，擴容的數量是原來的兩倍Cell[] rs = new Cell[n << 1];，為什麼要和CPU個數比較呢？
//因為當Cell陣列元素和CPU個數相同的時候，效率是最高的，因為每一個執行緒都是一個CPU來執行，再來修改其中其中一個Cell中的值
//這裡還是利用cellsBusy這個欄位，在下面初始化Cell陣列中的用法一樣，就不多說了
else if (cellsBusy == 0 && casCellsBusy()) {
try {
//這裡就是新建一個數組是原來的兩倍，然後將原來陣列的元素複製到新的陣列，再改變原來的cells的引用指向新的陣列
if (cells == as) { // Expand table unless stale
Cell[] rs = new Cell[n << 1];
for (int i = 0; i < n; ++i)
rs[i] = as[i];
cells = rs;
}
} finally {
//使用完就重置為0
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
//這裡的作用是當執行緒找了好久，發現所有Cell個數已經和CPU個數相同了，然後匹配到的Cell正在被其他執行緒使用
//於是為了找到一個空閒的Cell，於是要重新計算hash值
h = advanceProbe(h);
}
//初始化Cell陣列
//記得上面好像說過cellsBusy這個欄位是能是0或者是1，當時0的時候，說明Cell陣列沒有初始化和擴容，也沒有正在建立Cell元素，
//反之則為1，而casCellsBusy()方法就是用CAS將cellsBusy的值從0修改為1，表示當前執行緒正在初始化Cell陣列，其他執行緒就不能進行擴容操作了
//如果一個執行緒在初始化這個Cell陣列，其他執行緒在擴容的時候，看上面擴容，也會執行casCellsBusy()方法進行CAS操作，會失敗，因為期望的值是1，而不是0
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
//這裡首先新建一個容量為2的陣列，然後用隨機數h&1，也就是隨機數對陣列的容量取餘的方式得到索引，然後初始化陣列中每個Cell元素
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
//初始化完成之後要把這個欄位重置為0，表示此時其他執行緒就又可以對這個Cell進行擴容了
cellsBusy = 0;
}
if (init)
break;
}
//將base更新為base+x，表示base會逐漸累加Cell陣列中每一個Cell中的值
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
其實longAccumulate方法就是表示多執行緒的時候對Cell陣列的初始化，新增Cell元素還有擴容操作，還有就是當一個執行緒匹配到了Cell元素，發現其他執行緒正在使用就會重新計算隨機數，然後繼續匹配其他的Cell元素去了，沒什麼特別難的吧！別看這個方法很長，就是做這幾個操作
六.總結
這一篇核心就是CAS，我們簡單的說了一下原子操作類AtomicLong的自增，但是當執行緒很多的情況下，使用CAS有很大的缺點，就是同一時間是會有一個執行緒在執行，其他所有執行緒都在自旋，自旋會消耗效能，於是可以使用JDK提供的一個LongAdder類代替，這個類的作用就是將AtomicLong中的值優化為了一個base和一個Cell陣列，多執行緒去競爭的時候，假設執行緒個數個CPU個數相同，那麼此時每一個執行緒都有單獨的一個CPU去執行，然後單獨的匹配到Cell陣列中的某個元素，如果沒有匹配到那麼會對這個Cell陣列進行初始化操作；如果匹配到的Cell陣列中的元素正在使用，那麼久判斷是否可以新建一個Cell丟數組裡面去，如果陣列已經滿了，而且陣列數量小於CPU個數，那麼久進行擴容；擴容結束後，還是匹配到的Cell陣列中的位置正在使用，那麼就是衝突，就會重新計算，通過一個新的隨機數和陣列的取餘，得到一個新的索引，再去訪問該對應的Cell陣列的位置。。。。
仔細看看還是挺有意思的