一般都是CAS對(duì)一個(gè)變量進(jìn)行操作,但Doug Lea大神覺得不滿足,又寫了一個(gè)LongAdder
先看下傳統(tǒng)的
再來看下LongAdder的
即將一個(gè)變量進(jìn)一步拆分到一個(gè)base數(shù)組中,減少資源競(jìng)爭(zhēng)
@sun.misc.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long valueOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> ak = Cell.class;
valueOffset = UNSAFE.objectFieldOffset
(ak.getDeclaredField("value"));
} catch (Exception e) {
throw new Error(e);
}
}
}
- 類似于AtomicLong，利用CAS來更新變量
- @Contended避免value偽共享
將多個(gè)cell數(shù)組中的值加起來的和就類似于AtomicLong中的value
public long sum() {
Cell[] as = cells; Cell a;
long sum = base;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
increment()的調(diào)用鏈
java.util.concurrent.atomic.LongAdder.increment
->java.util.concurrent.atomic.LongAdder.add
add()方法如下
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[getProbe() & m]) == null ||
!(uncontended = a.cas(v = a.value, v + x)))
longAccumulate(x, null, uncontended);
}
}
第一次Cell數(shù)組為空,進(jìn)入casBase()
final boolean casBase(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
}
即原子更新,成功則直接返回,失敗則說明出現(xiàn)并發(fā)了
if的三個(gè)判斷
- 數(shù)組為空
- 或者數(shù)組長(zhǎng)度小于1
- 或者位置上沒有Cell對(duì)象,即getProbe()&m其實(shí)相當(dāng)于hashMap里面的tab[i = (n - 1) & hash]
- 或者修改cell的值失敗
才會(huì)最終進(jìn)入到longAccumulate()方法中
小結(jié)
- 如果Cells表為空，嘗試用CAS更新base字段，成功則退出；
- 如果Cells表為空，CAS更新base字段失敗，出現(xiàn)競(jìng)爭(zhēng)，uncontended為true，調(diào)用longAccumulate；
- 如果Cells表非空，但當(dāng)前線程映射的槽為空，uncontended為true，調(diào)用longAccumulate；
- 如果Cells表非空，且前線程映射的槽非空，CAS更新Cell的值，成功則返回，否則，uncontended設(shè)為false，調(diào)用longAccumulate。
看到這里大概應(yīng)該知道為什么LongAdder會(huì)比AtomicLong更高效了，沒錯(cuò)，唯一會(huì)制約AtomicLong高效的原因是高并發(fā)，高并發(fā)意味著CAS的失敗幾率更高， 重試次數(shù)更多，越多線程重試，CAS失敗幾率又越高，變成惡性循環(huán)，AtomicLong效率降低。 那怎么解決？ LongAdder給了我們一個(gè)非常容易想到的解決方案：減少并發(fā)，將單一value的更新壓力分擔(dān)到多個(gè)value中去，降低單個(gè)value的 “熱度”，分段更新?。?！
這樣，線程數(shù)再多也會(huì)分擔(dān)到多個(gè)value上去更新，只需要增加value就可以降低 value的 “熱度” AtomicLong中的 惡性循環(huán)不就解決了嗎？ cells 就是這個(gè) “段” cell中的value 就是存放更新值的， 這樣，當(dāng)我需要總數(shù)時(shí)，把cells 中的value都累加一下不就可以了么??！
在看看add方法中的代碼，casBase方法可不可以不要，直接分段更新,上來就計(jì)算 索引位置，然后更新value？
不是不行，而是有所考慮的,因?yàn)?，casBase操作等價(jià)于AtomicLong中的CAS操作，要知道，LongAdder這樣的處理方式是有壞處的，分段操作必然帶來空間上的浪費(fèi)，可以空間換時(shí)間，但是，能不換就不換，空間時(shí)間都節(jié)約.
casBase操作保證了在低并發(fā)時(shí)，不會(huì)立即進(jìn)入分支做分段更新操作，因?yàn)榈筒l(fā)時(shí)，casBase操作基本都會(huì)成功，只有并發(fā)高到一定程度了，才會(huì)進(jìn)入分支
longAccumulate()方法如下
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
int h;
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) {
if ((a = as[(n - 1) & h]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create
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;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
else if (n >= NCPU || cells != as)
collide = false; // At max size or stale
else if (!collide)
collide = true;
else if (cellsBusy == 0 && casCellsBusy()) {
try {
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 {
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
h = advanceProbe(h);
}
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
- 如果Cells表為空，嘗試獲取鎖之后初始化表（初始大小為2）；
- 如果Cells表非空，對(duì)應(yīng)的Cell為空，自旋鎖未被占用，嘗試獲取鎖，添加新的Cell；
- 如果Cells表非空，找到線程對(duì)應(yīng)的Cell，嘗試通過CAS更新該值；
- 如果Cells表非空，線程對(duì)應(yīng)的Cell CAS更新失敗，說明存在競(jìng)爭(zhēng)，嘗試獲取自旋鎖之后擴(kuò)容,將cells數(shù)組擴(kuò)大，降低每個(gè)cell的并發(fā)量后再試
