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

一、復習

• 上次連載簡單的介紹了其他函式的作用以及功能

二、完整的LongAdder類原始碼

package com.ruigege.AtomicOperationClass4;
import java.util.function.LongBinaryOperator;

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);
 }
 
 public long sum() {
  Cell[] as = cells;
  long sum = base;
  if(as != null) {
   for (int i=0;i<as.length;i++) {
    if(Cell[i] != null) {
     sum += Cell[i].value;
    }
   }
  }
 }
 
 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;
    }
   }
  }
 }
 
 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;
     
    }
   }
  }
 }
 
 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);
 }
 
 final void longAccumulate(long x,LongBinaryOperator fn,boolean wasUncontended) {
  //初始化當前執行緒的變數threadLocalRandomProbe的值
  //代碼6
  int h;
  if(h = getProbe() == 0) {
   ThreadLocalRandom.current();//獲取一個亂數的實體
   h = getProbe();//初始化
   //這個變數在計算當前執行緒應該被分配到cells陣列的哪一個Cell元素的時候會用到
   wasUncontended = true;//contend爭奪，辯稱
  }
  
  boolean collide = false;//collide沖突，碰撞
  for(;;) {
   Cell[] as;Cell a;int n;long v;
   if((as=cells) != null && (n=as.length)>0) {//當前執行緒呼叫了add方法并且根據當前執行緒的亂數threadLocalRandomProbe和cells元素的個數計算要訪問的Cell元素下
    //下標，然后如果發現對應下標元素的值為null，則新增一個cell元素到cells陣列，并且在將其提娜佳到cells陣列之前要競爭設定cellsBusy為1
    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;
    }
    collide = false;
    continue;
   }
   h = advanceProbe(h);
  }
  //代碼14，這里進行的是cells陣列初始化
  /**
   * 1.cellsBusy是一個標示，成員變數，用來實作自旋鎖，狀態值只有0和1，當創建Cell元素，擴容Cell陣列或者
   * 初始化陣列，使用CAS操作來保證同時只有一個執行緒可以進行其中之一操作
   * 2.為0的時候說明cells陣列沒有被初始化或者擴容，也沒有新建Cell元素；為1說明cells陣列在被初始化或者擴容，或者當前新建了Cell元素；通過CAS操作來進行0或1狀態切換，這里使用了casCellBusy函式
   * 3.假設當前執行緒通過CAS設定cellsBusy為1，則當前執行緒開始初始化操作，那么這個時候其他執行緒就不能進行擴容了，
   */
  else if(cellsBusy == 0 && cells == as && casCellsBusy()){
   boolean init = false;
   try {
    //因為在多執行緒下，所以有必要再次判斷一下
    if(cells == as) {
     //初始化陣列容量為2
     Cell[] rs = new Cell[2];
     //h&l用來計算當前執行緒應該訪問cell陣列的哪個位置
     //也就是使用當前執行緒的threadLocalRandomProbe變數值和cells陣列元素個數-1做與運算
     rs[h&l] = new Cell(x);
     cells = rs;
     //標示陣列已經完成初始化
     init = true;
    }
   }finally {
    //最后重置了cellBusys標記，這里沒有使用CAS操作，但是卻是執行緒安全的，因為該變數是一個
    //volatile變數，保證了記憶體可見性，另外擴容后的cells陣列里面除了包含復制過來的元素外，還包含其他元素，這些元素的值
    //都是null，
    cellsBusy = 0;
   }
   if(init) {
    break;
   }
  }else if(casBase(v = base,((fn == null) ? v+x:fn.applyAsLong(v,x)))) {
   break; 
  }
  
 }
}

三、原始碼：

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