逅弈 歡迎轉(zhuǎn)載丰捷，注明原創(chuàng)出處即可，謝謝寂汇！
要實(shí)現(xiàn)一個(gè)網(wǎng)站訪問量的計(jì)數(shù)器病往，可以通過一個(gè)Long類型的對(duì)象，并加上synchronized內(nèi)置鎖的方式骄瓣。但是這種方式使得多線程的訪問變成了串行的停巷，同一時(shí)刻只能有一個(gè)線程可以更改long的值，那么為了能夠使多線程并發(fā)的更新long的值榕栏，我們可以使用J.U.C包中的Atomic原子類畔勤。這些類的更新是原子的，不需要加鎖即可實(shí)現(xiàn)并發(fā)的更新扒磁，并且是線程安全的庆揪。
可是Atomic原子類是怎么保證并發(fā)更新的線程安全的呢？讓我們看一下AtomicLong的自增方法incrementAndGet()：
public final long incrementAndGet() {
// 無限循環(huán)妨托，即自旋
for (;;) {
// 獲取主內(nèi)存中的最新值
long current = get();
long next = current + 1;
// 通過CAS原子更新缸榛，若能成功則返回，否則繼續(xù)自旋
if (compareAndSet(current, next))
return next;
}
}
private volatile long value;
public final long get() {
return value;
}
可以發(fā)現(xiàn)其內(nèi)部保持著一個(gè)volatile修飾的long變量兰伤，volatile保證了long的值更新后内颗，其他線程能立即獲得最新的值。
在incrementAndGet中首先是一個(gè)無限循環(huán)（自旋）敦腔，然后獲取long的最新值均澳，將long加1，然后通過compareAndSet()方法嘗試將long的值有current更新為next会烙。如果能更新成功负懦，則說明當(dāng)前還沒有其他線程更新該值，則返回next柏腻，如果更新失敗纸厉，則說明有其他線程提前更新了該值五嫂，則當(dāng)前線程繼續(xù)自旋嘗試更新。
CAS的基本思想是認(rèn)為當(dāng)前環(huán)境中的并發(fā)并沒有那么高沃缘，比較樂觀的看待整個(gè)并發(fā)槐臀，只需要在更新某個(gè)值時(shí)先檢查下該值有沒有發(fā)生變化水慨，如果沒有發(fā)生變化則更新敬扛，否則放棄更新朝抖。
CAS的操作其底層是通過調(diào)用sun.misc.Unsafe類中的CompareAndSwap的方法保證線程安全的。Unsafe類中主要有下面三種CompareAndSwap方法：
public final native boolean compareAndSwapObject(Object obj, long offset, Object expect, Object update);
public final native boolean compareAndSwapInt(Object obj, long offset, int expect, int update);
public final native boolean compareAndSwapLong(Object obj, long offset, long expect, long update);
可以看到這些方法都是native的急侥，需要調(diào)用JNI接口坏怪，也即通過操作系統(tǒng)來保證這些方法的執(zhí)行陕悬。
以上原子更新操作中除了CAS之外還有一個(gè)自旋（無限循環(huán)）捉超，那么什么是自旋呢拼岳？為什么要用自旋呢？下面我們來了解一下自旋
自旋鎖
- 簡(jiǎn)述
跟互斥鎖一樣惜纸，一個(gè)線程要想訪問被自旋鎖保護(hù)的共享資源耐版，必須先得到鎖粪牲，在訪問完共享資源后腺阳，必須釋放鎖亭引。
如果在獲取自旋鎖時(shí)焙蚓，沒有線程保持該鎖，那么將立即得到鎖赵哲；如果在獲取自旋鎖時(shí)鎖已經(jīng)有保持者，那么獲取鎖的操作將自旋在那里将宪，直到該自旋鎖的保持者釋放了鎖较坛。
- 優(yōu)點(diǎn)
自旋鎖比較適用于鎖使用者保持鎖時(shí)間比較短的情況丑勤，比如執(zhí)行一個(gè)變量的自增操作法竞。
正是由于自旋鎖使用者一般保持鎖時(shí)間非常短，因此選擇自旋而不是睡眠是非常必要的薛躬，自旋鎖的效率遠(yuǎn)高于互斥鎖型宝，因?yàn)榫€程的睡眠趴酣、喚醒需要操作系統(tǒng)的支持岖寞，開銷比較大渊涝，因此當(dāng)一個(gè)操作保持鎖的時(shí)間非常短時(shí)跨释，不需要將線程掛起或睡眠，而是讓線程執(zhí)行一個(gè)忙循環(huán)岁疼，等到自旋鎖的持有者釋放了鎖之后瑰排，當(dāng)前線程將會(huì)獲得鎖暖侨。
- 缺點(diǎn)
遞歸死鎖
試圖遞歸地獲得自旋鎖必然會(huì)引起死鎖：遞歸程序的持有實(shí)例在第二個(gè)實(shí)例循環(huán)，以試圖獲得相同自旋鎖時(shí)京郑，不會(huì)釋放此自旋鎖
過多占用cpu資源
如果不加限制些举，由于申請(qǐng)者一直在循環(huán)等待户魏，因此自旋鎖在鎖定的時(shí)候,如果不成功,不會(huì)睡眠,會(huì)持續(xù)的嘗試叼丑。
單cpu的時(shí)候自旋鎖會(huì)讓其它process動(dòng)不了幢码。因此症副，一般自旋鎖實(shí)現(xiàn)會(huì)有一個(gè)參數(shù)限定最多持續(xù)嘗試次數(shù)政基，超出后, 自旋鎖放棄當(dāng)前time slice辕坝。 等下一次機(jī)會(huì)
雖然CAS可以高效的對(duì)某些共享變量進(jìn)行并發(fā)的更改酱畅，但是他也是有缺點(diǎn)的纺酸，其中之一就是ABA問題餐蔬。當(dāng)要更改的值從A變?yōu)锽樊诺，之后又變?yōu)锳词爬，則檢查時(shí)可能會(huì)發(fā)現(xiàn)沒有發(fā)生變化顿膨，實(shí)際上已經(jīng)發(fā)生了變化。解決方法是變更之前加上版本號(hào)，如1A芽唇，2B取劫，3A炮捧∨乜危可通過AtomicStampedReference來解決ABA問題书蚪，這個(gè)類的compareAndSet方法殊校，將首先檢查當(dāng)前引用是否等于預(yù)期引用为流，并且當(dāng)前標(biāo)志是否等于預(yù)期標(biāo)志敬察，如果全部相等，則以原子方式將該引用和該標(biāo)志的值設(shè)置為給定的更新值虫给，否則不予更新。
除此之外缠黍，在并發(fā)量非常高的情況下瓷式，CAS失敗的幾率將變得非常高语泽，重試的次數(shù)也會(huì)跟著增加贸典，越多線程重試，CAS失敗的幾率就越高踱卵，變成惡性循環(huán)廊驼。因此在并發(fā)量非常高的環(huán)境中，如果仍然想通過原子類來更新的話惋砂，可以使用AtomicLong的替代類：LongAdder妒挎。
將單一value的更新壓力分擔(dān)到多個(gè)value中去，降低單個(gè)value的“熱度”西饵，分段更新酝掩，這樣，線程數(shù)再多也會(huì)分擔(dān)到多個(gè)value上去更新眷柔，只需要增加value的個(gè)數(shù)就可以降低value的 “熱度”,這樣AtomicLong中的惡性循環(huán)就可以解決了。
在LongAdder中cells就是這個(gè)“段”,cell中的value就是存放更新值的，這樣，當(dāng)我需要總數(shù)時(shí)牛郑，把cell中的value都累加一下不就可以了么
讓我們看一下LongAdder更新的原則：
1.當(dāng)并發(fā)低時(shí)先采用CAS進(jìn)行更新础芍，如果更新成功即返回
2.當(dāng)并發(fā)高且CAS更新失敗時(shí)，則進(jìn)入分段更新
LongAdder的部分代碼實(shí)現(xiàn)：
/**
* Adds the given value.
*
* @param x the value to add
*/
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
// 當(dāng)并發(fā)低時(shí)先采用CAS進(jìn)行add，如果更新成功即返回
// 當(dāng)并發(fā)高且CAS更新失敗時(shí)，則進(jìn)入分段更新
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
// 找到cells數(shù)組中該值對(duì)應(yīng)的cell對(duì)象
(a = as[getProbe() & m]) == null ||
// 使用cell對(duì)象的cas方法進(jìn)行更新
!(uncontended = a.cas(v = a.value, v + x)))
longAccumulate(x, null, uncontended);
}
}
/**
* Handles cases of updates involving initialization, resizing,
* creating new Cells, and/or contention. See above for
* explanation. This method suffers the usual non-modularity
* problems of optimistic retry code, relying on rechecked sets of
* reads.
*
* @param x the value
* @param fn the update function, or null for add (this convention
* avoids the need for an extra field or function in LongAdder).
* @param wasUncontended false if CAS failed before call
*/
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
}
}
需要注意的是朦佩，雖然AtomicLong等原子類的更新是原子的弄砍，但是多個(gè)原子操作合并后的操作卻不是原子的衣式，也即：原子+原子!=原子乃正，下面將用一個(gè)例子來說明該問題：
private CountDownLatch latch;
private class Discovery{
private Map<String,SlaveNode> slaveNodeMap;
private AtomicInteger slaveIndex;
public Discovery(){
slaveNodeMap = new HashMap<String,SlaveNode>();
SlaveNode slaveNode1 = new SlaveNode("127.0.0.1",8081);
SlaveNode slaveNode2 = new SlaveNode("127.0.0.1",8082);
slaveNodeMap.put(slaveNode1.getId(),slaveNode1);
slaveNodeMap.put(slaveNode2.getId(),slaveNode2);
slaveIndex = new AtomicInteger(0);
}
public SlaveNode discover() {
if (slaveNodeMap.size() == 0) {
System.err.println("No available SlaveNode!");
return null;
}
SlaveNode[] nodes = new SlaveNode[]{};
nodes = slaveNodeMap.values().toArray(nodes);
// 通過CAS循環(huán)獲取下一個(gè)可用服務(wù)
// 當(dāng)當(dāng)前索引為數(shù)組的長(zhǎng)度是划栓，將索引值更新為0
slaveIndex.compareAndSet(nodes.length,0);
System.out.println("currentIndex=" + slaveIndex + ",currentThread=" + Thread.currentThread().getName());
// 根據(jù)數(shù)組的下標(biāo)獲取可用的服務(wù)委煤，之后將索引通過原子方式加1
return nodes[slaveIndex.getAndIncrement()];
}
}
@Test
public void testConcurrentDiscover(){
int loopTimes = 300;
latch = new CountDownLatch(loopTimes);
Discovery discovery = new Discovery();
class Runner implements Runnable{
@Override
public void run() {
Object object = discovery.discover();
System.out.println(String.format("object={%s},currentThread={%s}",(object!=null?object.toString():"null"),Thread.currentThread().getName()));
latch.countDown();
}
}
for(int i=0;i<loopTimes;i++){
new Thread(new Runner()).start();
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
該方法執(zhí)行的可能會(huì)成功，即多線程交替獲得SlaveNode，但也會(huì)出現(xiàn)錯(cuò)誤，slaveIndex的值會(huì)超過數(shù)組的長(zhǎng)度赏淌，問題就出在這段代碼：
slaveIndex.compareAndSet(nodes.length,0);
return nodes[slaveIndex.getAndIncrement()];
這兩個(gè)操作本身都是原子的，但是合并在一起就不是原子的了场靴，因此會(huì)出現(xiàn)錯(cuò)誤，解決的方法還是對(duì)整個(gè)執(zhí)行的過程加鎖哮兰。
