非阻塞同步機(jī)制

1、JAVA并發(fā)編程實(shí)踐中提供了 3中非阻塞算法的示例。第一個(gè)示例，非阻塞計(jì)數(shù)器。CAS比較并交換即Compare-And-Swap。假設(shè)CAS有3個(gè)操作數(shù)-內(nèi)存位置V、舊的預(yù)測值A(chǔ)和新值B,那么它的典型模式為：首先從 V中讀取值A(chǔ),由A生成新值B,然后使用CAS 原子化地把V的值改成B,并且期間不能有其他線程改變V的值，因?yàn)镃AS能夠發(fā)現(xiàn)來自其他線程的干擾。view plaincopy to clipboardprint?01.代碼1使用CAS實(shí)現(xiàn)的非阻塞計(jì)數(shù)器02.ThreadSafe03.public class CasCounter 04.private SimulatedCAS valu

2、e;05.publicintgetValue() 06.return value.get(); TOC o 1-5 h z 07.08. v;do v = value.get(); 1while(v !=pareAndSwap(v, v + 1);2return v +1;代碼2模才C CAS操作ThreadSafepublic class SimulatedCAS GuardedBy(this) private int value;public synchronized int get() return value; public synch

3、ronized int compareAndSwap(int expectedValue,int newValue) int oldValue = value;if (oldValue = expectedValue)value = newValue;return oldValue;public synchronized boolean compareAndSet(int expectedValue,int newValue) return (expectedValue= compareAndSwap(expectedValue, newValue);代碼1使用CAS實(shí)現(xiàn)的非阻塞計(jì)數(shù)器Thre

4、adSafe public class CasCounter private SimulatedCAS value; public int getValue() return value.get(); public int increment() int v; do v = value.get(); 1 while (v != pareAndSwap(v, v + 1);2return v + 1; 代碼2模CAS操作 ThreadSafe public class SimulatedCAS GuardedBy(this) private int value;public synchroniz

5、ed int get() return value; public synchronized int compareAndSwap(int expectedValue, int newValue) int oldValue = value;if (oldValue = expectedValue) value = newValue;return oldValue; public synchronized boolean compareAndSet(int expectedValue, int newValue) return (expectedValue= compareAndSwap(exp

6、ectedValue, newValue); 假設(shè)有兩個(gè)線程，同時(shí)執(zhí)行到 1 ,獲得了 value的舊值；然后同時(shí)執(zhí)行到2 ,根據(jù)原則當(dāng)多個(gè)線程試圖使用CAS同時(shí)更新相同的變量時(shí)，其中一個(gè)會(huì)勝出，并更新變量的值，而其他線程都會(huì)失敗，重新嘗試?！笨芍?，其中一個(gè)線程完成了加1操作，而另一個(gè)線程失敗，重新do循環(huán)。讓我們細(xì)細(xì)體會(huì)一下這個(gè)原則是怎么得到的：一個(gè)線程完成了加1操作后，另一個(gè)線程使用 CAS時(shí)，舊的預(yù)期值沒有變但內(nèi)存位置V的值已經(jīng)更新了，所以此時(shí)V的值不等于舊的預(yù)期值而導(dǎo)致失??！ 第二個(gè)示例，非阻塞棧view plaincopy to clipboardprint?01.代碼3使用Trei

7、ber算法的非阻塞棧02.ThreadSafe03.public class ConcurrentStack 04.05.06.07.08.09.代碼3AtomicReferenceNode top = new AtomicReferenceNode(); public void push(E item) Node newHead = new Node(item);Node oldHead;do oldHead = top.get();newHead.next = oldHead; 1 while (!pareAndSet(oldHead, newHead); 2public E pop()

8、Node oldHead;Node newHead;do oldHead = top.get();if (oldHead = null)return null;newHead = oldHead.next; while (!pareAndSet(oldHead, newHead);return oldHead.item;private static class Node public final E item;public Node next;public Node(E item) this.item = item;使用Treiber算法的非阻塞棧ThreadSafepublic class

9、ConcurrentStack AtomicReferenceNode top = new AtomicReferenceNode(); public void push(E item) Node newHead = new Node(item);Node oldHead;do oldHead = top.get();newHead.next = oldHead;1 while (!pareAndSet(oldHead, newHead); 2 public E pop() Node oldHead;Node newHead;do oldHead = top.get(); if (oldHea

10、d = null) return null; newHead = oldHead.next; while (!pareAndSet(oldHead, newHead); return oldHead.item;private static class Node public final E item;public Node next; public Node(E item) this.item = item; 注： AtomicReferencecompareAndSet ( V expect, V update)若當(dāng)前值與期望值 expect相等時(shí)，原子化地將update值賦給當(dāng)前值。假設(shè)有

11、兩個(gè)線程，同時(shí)執(zhí)行到1，獲得了棧頂元素，并創(chuàng)建了一個(gè)新節(jié)點(diǎn)指向當(dāng)前棧頂； 然后同時(shí)執(zhí)行到2 ,根據(jù)原則”當(dāng)多個(gè)線程試圖使用CAS同時(shí)更新相同的變量時(shí)，其中一個(gè)會(huì)勝出，并更新變量的值，而其他線程都會(huì)失敗，重新嘗試?！笨芍?，其中一個(gè)線程完成了插入操作，而另一個(gè)線程失敗，重新 do循環(huán)。讓我們細(xì)細(xì)體會(huì)一下這個(gè)原則是怎么得 到的：一個(gè)線程完成了插入操作后，另一個(gè)線程使用CAS時(shí)，舊的預(yù)期值沒有變但當(dāng)前棧頂?shù)闹狄呀?jīng)更新了，所以此時(shí)棧頂?shù)闹挡坏扔谂f的預(yù)期值而導(dǎo)致失??！ 第三個(gè)示例，非阻塞鏈表view plaincopy to clipboardprint?01.代碼4 Michael-Scott非阻塞隊(duì)列

12、算法中的插入02.ThreadSafe03.public class LinkedQueue 04. private static class Node 05.final E item;06.final AtomicReferenceNode next;07.public Node(E item, Node next) 08.this.item = item;09.this.next = new AtomicReferenceNode(next);private final Node dummy = new Node(null, null);private final AtomicRefere

13、nceNode head= new AtomicReferenceNode(dummy);private final AtomicReferenceNode tail= new AtomicReferenceNode(dummy);public boolean put(E item) Node newNode = new Node(item, null);while (true) Node curTail = tail.get();Node tailNext = curTail.next.get();if (curTail = tail.get() ?if (tailNext != null)

14、 A/ Queue in intermediate state, advance tailpareAndSet(curTail, tailNext); B else /In quiescent state, try inserting new nodeif (curTpareAndSet(null, newNode) C/ Insertion succeeded, try advancing tail TOC o 1-5 h z pareAndSet(curTail, newNode);Dreturn true;代碼4 Michael-Scott非阻塞隊(duì)列算法中的插入ThreadSafepub

15、lic class LinkedQueue private static class Node final E item;final AtomicReferenceNode next;public Node(E item, Node next) this.item = item;this.next = new AtomicReferenceNode(next);private final Node dummy = new Node(null, null);private final AtomicReferenceNode head=new AtomicReferenceNode(dummy);

16、private final AtomicReferenceNode tail=new AtomicReferenceNode(dummy);public boolean put(E item) Node newNode = new Node(item, null);while (true) Node curTail = tail.get();Node tailNext = curTail.next.get();if (curTail = tail.get() ?if (tailNext != null) A/ Queue in intermediate state, advance tail

17、pareAndSet(curTail, tailNext); B else / In quiescent state, try inserting new nodeif (curTpareAndSet(null, newNode) C/ Insertion succeeded, try advancing tail pareAndSet(curTail, newNode); D return true;首先看？處，為什么要判斷 curTail = tail.get()呢？必須要有這個(gè)判斷來保證數(shù)據(jù)結(jié)構(gòu)總能處于一致狀態(tài)。如果沒有這個(gè)判斷的話可能出現(xiàn)下面狀況。細(xì)細(xì)思索一下，如果沒有 curTail

18、 = tail.get（）這個(gè)的話，一個(gè)線程將元素3加入隊(duì)列，而另一個(gè)線程卻把next指針的指向了 3,元素3就這樣被丟棄了，這當(dāng)然是不行的！也就是我們 下文所說的第一個(gè)訣竅。插入新的元素涉及到兩個(gè)指針的更新（兩個(gè)指針分別為隊(duì)尾指針和隊(duì)尾元素的next指針），需要兩個(gè)操作過程。第一，更新當(dāng)前隊(duì)尾元素的 next指針，將新元素插入到列表隊(duì)尾；第 二，釋放隊(duì)尾指針，指向新的最末元素。在這兩個(gè)操作之間，隊(duì)列處于中間狀態(tài)，看圖示。圖a插入前穩(wěn)定狀態(tài)圖b插入期間，隊(duì)列處于中間狀態(tài)圖c插入完成后，隊(duì)列再一次回到穩(wěn)定狀態(tài)有幾個(gè)訣竅來完成我們的鏈表。第一個(gè)訣竅是即使在多步更新中，也要確保數(shù)據(jù)結(jié)構(gòu)總能處 于一致狀態(tài)。也就是說，如果線程B到達(dá)時(shí)發(fā)現(xiàn)線程 A正在更新中，B能夠知曉操作已經(jīng)部 分完成并且知道不能立即開始自己的更新。那么B就開始等待（通過反復(fù)檢查隊(duì)列狀態(tài)）直到A完成更