非阻塞同步算法與CAS(Compare and Swap)無鎖算法
鎖（lock）的代價
鎖是用來作併發最簡單的方式，固然其代價也是最高的。內核態的鎖的時候須要操做系統進行一次上下文切換，加鎖、釋放鎖會致使比較多的上下文切換和調度延時，等待鎖的線程會被掛起直至鎖釋放。在上下文切換的時候，cpu以前緩存的指令和數據都將失效，對性能有很大的損失。操做系統對多線程的鎖進行判斷就像兩姐妹在爲一個玩具在爭吵，而後操做系統就是能決定他們誰能拿到玩具的父母，這是很慢的。用戶態的鎖雖然避免了這些問題，可是其實它們只是在沒有真實的競爭時纔有效。html
Java在JDK1.5以前都是靠synchronized關鍵字保證同步的，這種經過使用一致的鎖定協議來協調對共享狀態的訪問，能夠確保不管哪一個線程持有守護變量的鎖，都採用獨佔的方式來訪問這些變量，若是出現多個線程同時訪問鎖，那第一些線線程將被掛起，當線程恢復執行時，必須等待其它線程執行完他們的時間片之後才能被調度執行，在掛起和恢復執行過程當中存在着很大的開銷。鎖還存在着其它一些缺點，當一個線程正在等待鎖時，它不能作任何事。若是一個線程在持有鎖的狀況下被延遲執行，那麼全部須要這個鎖的線程都沒法執行下去。若是被阻塞的線程優先級高，而持有鎖的線程優先級低，將會致使優先級反轉(Priority Inversion)。java
樂觀鎖與悲觀鎖
獨佔鎖是一種悲觀鎖，synchronized就是一種獨佔鎖，它假設最壞的狀況，而且只有在確保其它線程不會形成干擾的狀況下執行，會致使其它全部須要鎖的線程掛起，等待持有鎖的線程釋放鎖。而另外一個更加有效的鎖就是樂觀鎖。所謂樂觀鎖就是，每次不加鎖而是假設沒有衝突而去完成某項操做，若是由於衝突失敗就重試，直到成功爲止。node
volatile的問題
與鎖相比，volatile變量是一和更輕量級的同步機制，由於在使用這些變量時不會發生上下文切換和線程調度等操做，可是volatile變量也存在一些侷限：不能用於構建原子的複合操做，所以當一個變量依賴舊值時就不能使用volatile變量。（參考：談談volatiile）nginx
volatile只能保證變量對各個線程的可見性，但不能保證原子性。爲何？見個人另一篇文章：《爲何volatile不能保證原子性而Atomic能夠？》git
Java中的原子操做( atomic operations)
原子操做指的是在一步以內就完成並且不能被中斷。原子操做在多線程環境中是線程安全的，無需考慮同步的問題。在java中，下列操做是原子操做：github
- all assignments of primitive types except for long and double
- all assignments of references
- all operations of java.concurrent.Atomic* classes
- all assignments to volatile longs and doubles
問題來了，爲何long型賦值不是原子操做呢？例如：redis
long foo = 65465498L;
實時上java會分兩步寫入這個long變量，先寫32位，再寫後32位。這樣就線程不安全了。若是改爲下面的就線程安全了：算法
private volatile long foo;
由於volatile內部已經作了synchronized.數據庫
CAS無鎖算法
要實現無鎖（lock-free）的非阻塞算法有多種實現方法，其中CAS（比較與交換，Compare and swap）是一種有名的無鎖算法。CAS, CPU指令，在大多數處理器架構，包括IA3二、Space中採用的都是CAS指令，CAS的語義是「我認爲V的值應該爲A，若是是，那麼將V的值更新爲B，不然不修改並告訴V的值實際爲多少」，CAS是項樂觀鎖技術，當多個線程嘗試使用CAS同時更新同一個變量時，只有其中一個線程能更新變量的值，而其它線程都失敗，失敗的線程並不會被掛起，而是被告知此次競爭中失敗，並能夠再次嘗試。CAS有3個操做數，內存值V，舊的預期值A，要修改的新值B。當且僅當預期值A和內存值V相同時，將內存值V修改成B，不然什麼都不作。CAS無鎖算法的C實現以下：編程
int compare_and_swap (int* reg, int oldval, int newval)
{
ATOMIC();
int old_reg_val = *reg;
if (old_reg_val == oldval)
*reg = newval;
END_ATOMIC();
return old_reg_val;
}
CAS（樂觀鎖算法）的基本假設前提
CAS比較與交換的僞代碼能夠表示爲：
do{
備份舊數據；
基於舊數據構造新數據；
}while(!CAS( 內存地址，備份的舊數據，新數據 ))
（上圖的解釋：CPU去更新一個值，但若是想改的值再也不是原來的值，操做就失敗，由於很明顯，有其它操做先改變了這個值。）
就是指當二者進行比較時，若是相等，則證實共享數據沒有被修改，替換成新值，而後繼續往下運行；若是不相等，說明共享數據已經被修改，放棄已經所作的操做，而後從新執行剛纔的操做。容易看出 CAS 操做是基於共享數據不會被修改的假設，採用了相似於數據庫的 commit-retry 的模式。當同步衝突出現的機會不多時，這種假設能帶來較大的性能提高。
CAS的開銷（CPU Cache Miss problem）
前面說過了，CAS（比較並交換）是CPU指令級的操做，只有一步原子操做，因此很是快。並且CAS避免了請求操做系統來裁定鎖的問題，不用麻煩操做系統，直接在CPU內部就搞定了。但CAS就沒有開銷了嗎？不！有cache miss的狀況。這個問題比較複雜，首先須要瞭解CPU的硬件體系結構：
上圖能夠看到一個8核CPU計算機系統，每一個CPU有cache（CPU內部的高速緩存，寄存器），管芯內還帶有一個互聯模塊，使管芯內的兩個核能夠互相通訊。在圖中央的系統互聯模塊可讓四個管芯相互通訊，而且將管芯與主存鏈接起來。數據以「緩存線」爲單位在系統中傳輸，「緩存線」對應於內存中一個 2 的冪大小的字節塊，大小一般爲 32 到 256 字節之間。當 CPU 從內存中讀取一個變量到它的寄存器中時，必須首先將包含了該變量的緩存線讀取到 CPU 高速緩存。一樣地，CPU 將寄存器中的一個值存儲到內存時，不只必須將包含了該值的緩存線讀到 CPU 高速緩存，還必須確保沒有其餘 CPU 擁有該緩存線的拷貝。
好比，若是 CPU0 在對一個變量執行「比較並交換」（CAS）操做，而該變量所在的緩存線在 CPU7 的高速緩存中，就會發生如下通過簡化的事件序列：
- CPU0 檢查本地高速緩存，沒有找到緩存線。
- 請求被轉發到 CPU0 和 CPU1 的互聯模塊，檢查 CPU1 的本地高速緩存，沒有找到緩存線。
- 請求被轉發到系統互聯模塊，檢查其餘三個管芯，得知緩存線被 CPU6和 CPU7 所在的管芯持有。
- 請求被轉發到 CPU6 和 CPU7 的互聯模塊，檢查這兩個 CPU 的高速緩存，在 CPU7 的高速緩存中找到緩存線。
- CPU7 將緩存線發送給所屬的互聯模塊，而且刷新本身高速緩存中的緩存線。
- CPU6 和 CPU7 的互聯模塊將緩存線發送給系統互聯模塊。
- 系統互聯模塊將緩存線發送給 CPU0 和 CPU1 的互聯模塊。
- CPU0 和 CPU1 的互聯模塊將緩存線發送給 CPU0 的高速緩存。
- CPU0 如今能夠對高速緩存中的變量執行 CAS 操做了
以上是刷新不一樣CPU緩存的開銷。最好狀況下的 CAS 操做消耗大概 40 納秒，超過 60 個時鐘週期。這裏的「最好狀況」是指對某一個變量執行 CAS 操做的 CPU 正好是最後一個操做該變量的CPU，因此對應的緩存線已經在 CPU 的高速緩存中了，相似地，最好狀況下的鎖操做（一個「round trip 對」包括獲取鎖和隨後的釋放鎖）消耗超過 60 納秒，超過 100 個時鐘週期。這裏的「最好狀況」意味着用於表示鎖的數據結構已經在獲取和釋放鎖的 CPU 所屬的高速緩存中了。鎖操做比 CAS 操做更加耗時，是因深刻理解並行編程
爲鎖操做的數據結構中須要兩個原子操做。緩存未命中消耗大概 140 納秒，超過 200 個時鐘週期。須要在存儲新值時查詢變量的舊值的 CAS 操做，消耗大概 300 納秒，超過 500 個時鐘週期。想一想這個，在執行一次 CAS 操做的時間裏，CPU 能夠執行 500 條普通指令。這代表了細粒度鎖的侷限性。
如下是cache miss cas 和lock的性能對比：
JVM對CAS的支持：AtomicInt, AtomicLong.incrementAndGet()
在JDK1.5以前，若是不編寫明確的代碼就沒法執行CAS操做，在JDK1.5中引入了底層的支持，在int、long和對象的引用等類型上都公開了CAS的操做，而且JVM把它們編譯爲底層硬件提供的最有效的方法，在運行CAS的平臺上，運行時把它們編譯爲相應的機器指令，若是處理器/CPU不支持CAS指令，那麼JVM將使用自旋鎖。所以，值得注意的是，CAS解決方案與平臺/編譯器緊密相關（好比x86架構下其對應的彙編指令是lock cmpxchg，若是想要64Bit的交換，則應使用lock cmpxchg8b。在.NET中咱們可使用Interlocked.CompareExchange函數）。
在原子類變量中，如java.util.concurrent.atomic中的AtomicXXX，都使用了這些底層的JVM支持爲數字類型的引用類型提供一種高效的CAS操做，而在java.util.concurrent中的大多數類在實現時都直接或間接的使用了這些原子變量類。
Java 1.6中AtomicLong.incrementAndGet()的實現源碼爲：
1: /*
2: * Written by Doug Lea with assistance from members of JCP JSR-166
3: * Expert Group and released to the public domain, as explained at
4: * http://creativecommons.org/licenses/publicdomain
5: */
6:
7: package java.util.concurrent.atomic;
8: import sun.misc.Unsafe;
9:
10: /**
11: * A <tt>long</tt> value that may be updated atomically. See the
12: * {@link java.util.concurrent.atomic} package specification for
13: * description of the properties of atomic variables. An
14: * <tt>AtomicLong</tt> is used in applications such as atomically
15: * incremented sequence numbers, and cannot be used as a replacement
16: * for a {@link java.lang.Long}. However, this class does extend
17: * <tt>Number</tt> to allow uniform access by tools and utilities that
18: * deal with numerically-based classes.
19: *
20: * @since 1.5
21: * @author Doug Lea
22: */
23: public class AtomicLong extends Number implements java.io.Serializable {
24: private static final long serialVersionUID = 1927816293512124184L;
25:
26: // setup to use Unsafe.compareAndSwapLong for updates
27: private static final Unsafe unsafe = Unsafe.getUnsafe();
28: private static final long valueOffset;
29:
30: /**
31: * Records whether the underlying JVM supports lockless
32: * CompareAndSet for longs. While the unsafe.CompareAndSetLong
33: * method works in either case, some constructions should be
34: * handled at Java level to avoid locking user-visible locks.
35: */
36: static final boolean VM_SUPPORTS_LONG_CAS = VMSupportsCS8();
37:
38: /**
39: * Returns whether underlying JVM supports lockless CompareAndSet
40: * for longs. Called only once and cached in VM_SUPPORTS_LONG_CAS.
41: */
42: private static native boolean VMSupportsCS8();
43:
44: static {
45: try {
46: valueOffset = unsafe.objectFieldOffset
47: (AtomicLong.class.getDeclaredField("value"));
48: } catch (Exception ex) { throw new Error(ex); }
49: }
50:
51: private volatile long value;
52:
53: /**
54: * Creates a new AtomicLong with the given initial value.
55: *
56: * @param initialValue the initial value
57: */
58: public AtomicLong(long initialValue) {
59: value = initialValue;
60: }
61:
62: /**
63: * Creates a new AtomicLong with initial value <tt>0</tt>.
64: */
65: public AtomicLong() {
66: }
67:
68: /**
69: * Gets the current value.
70: *
71: * @return the current value
72: */
73: public final long get() {
74: return value;
75: }
76:
77: /**
78: * Sets to the given value.
79: *
80: * @param newValue the new value
81: */
82: public final void set(long newValue) {
83: value = newValue;
84: }
85:
86: /**
87: * Eventually sets to the given value.
88: *
89: * @param newValue the new value
90: * @since 1.6
91: */
92: public final void lazySet(long newValue) {
93: unsafe.putOrderedLong(this, valueOffset, newValue);
94: }
95:
96: /**
97: * Atomically sets to the given value and returns the old value.
98: *
99: * @param newValue the new value
100: * @return the previous value
101: */
102: public final long getAndSet(long newValue) {
103: while (true) {
104: long current = get();
105: if (compareAndSet(current, newValue))
106: return current;
107: }
108: }
109:
110: /**
111: * Atomically sets the value to the given updated value
112: * if the current value <tt>==</tt> the expected value.
113: *
114: * @param expect the expected value
115: * @param update the new value
116: * @return true if successful. False return indicates that
117: * the actual value was not equal to the expected value.
118: */
119: public final boolean compareAndSet(long expect, long update) {
120: return unsafe.compareAndSwapLong(this, valueOffset, expect, update);
121: }
122:
123: /**
124: * Atomically sets the value to the given updated value
125: * if the current value <tt>==</tt> the expected value.
126: * May fail spuriously and does not provide ordering guarantees,
127: * so is only rarely an appropriate alternative to <tt>compareAndSet</tt>.
128: *
129: * @param expect the expected value
130: * @param update the new value
131: * @return true if successful.
132: */
133: public final boolean weakCompareAndSet(long expect, long update) {
134: return unsafe.compareAndSwapLong(this, valueOffset, expect, update);
135: }
136:
137: /**
138: * Atomically increments by one the current value.
139: *
140: * @return the previous value
141: */
142: public final long getAndIncrement() {
143: while (true) {
144: long current = get();
145: long next = current + 1;
146: if (compareAndSet(current, next))
147: return current;
148: }
149: }
150:
151: /**
152: * Atomically decrements by one the current value.
153: *
154: * @return the previous value
155: */
156: public final long getAndDecrement() {
157: while (true) {
158: long current = get();
159: long next = current - 1;
160: if (compareAndSet(current, next))
161: return current;
162: }
163: }
164:
165: /**
166: * Atomically adds the given value to the current value.
167: *
168: * @param delta the value to add
169: * @return the previous value
170: */
171: public final long getAndAdd(long delta) {
172: while (true) {
173: long current = get();
174: long next = current + delta;
175: if (compareAndSet(current, next))
176: return current;
177: }
178: }
179:
180: /**
181: * Atomically increments by one the current value.
182: *
183: * @return the updated value
184: */
185: public final long incrementAndGet() {
186: for (;;) {
187: long current = get();
188: long next = current + 1;
189: if (compareAndSet(current, next))
190: return next;
191: }
192: }
193:
194: /**
195: * Atomically decrements by one the current value.
196: *
197: * @return the updated value
198: */
199: public final long decrementAndGet() {
200: for (;;) {
201: long current = get();
202: long next = current - 1;
203: if (compareAndSet(current, next))
204: return next;
205: }
206: }
207:
208: /**
209: * Atomically adds the given value to the current value.
210: *
211: * @param delta the value to add
212: * @return the updated value
213: */
214: public final long addAndGet(long delta) {
215: for (;;) {
216: long current = get();
217: long next = current + delta;
218: if (compareAndSet(current, next))
219: return next;
220: }
221: }
222:
223: /**
224: * Returns the String representation of the current value.
225: * @return the String representation of the current value.
226: */
227: public String toString() {
228: return Long.toString(get());
229: }
230:
231:
232: public int intValue() {
233: return (int)get();
234: }
235:
236: public long longValue() {
237: return (long)get();
238: }
239:
240: public float floatValue() {
241: return (float)get();
242: }
243:
244: public double doubleValue() {
245: return (double)get();
246: }
247:
248: }
因而可知，AtomicLong.incrementAndGet的實現用了樂觀鎖技術，調用了sun.misc.Unsafe類庫裏面的 CAS算法，用CPU指令來實現無鎖自增。因此，AtomicLong.incrementAndGet的自增比用synchronized的鎖效率倍增。
public final int getAndIncrement() {
for (;;) {
int current = get();
int next = current + 1;
if (compareAndSet(current, next))
return current;
}
}
public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
下面是測試代碼：能夠看到用AtomicLong.incrementAndGet的性能比用synchronized高出幾倍。
package console;
import java.util.concurrent.atomic.AtomicLong;
public class main {
/**
* @param args
*/
public static void main(String[] args) {
System.out.println("START -- ");
calc();
calcSynchro();
calcAtomic();
testThreadsSync();
testThreadsAtomic();
testThreadsSync2();
testThreadsAtomic2();
System.out.println("-- FINISHED ");
}
private static void calc() {
stopwatch sw = new stopwatch();
sw.start();
long val = 0;
while (val < 10000000L) {
val++;
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" calc() elapsed (ms): " + milSecds);
}
private static void calcSynchro() {
stopwatch sw = new stopwatch();
sw.start();
long val = 0;
while (val < 10000000L) {
synchronized (main.class) {
val++;
}
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" calcSynchro() elapsed (ms): " + milSecds);
}
private static void calcAtomic() {
stopwatch sw = new stopwatch();
sw.start();
AtomicLong val = new AtomicLong(0);
while (val.incrementAndGet() < 10000000L) {
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" calcAtomic() elapsed (ms): " + milSecds);
}
private static void testThreadsSync(){
stopwatch sw = new stopwatch();
sw.start();
Thread t1 = new Thread(new LoopSync());
t1.start();
Thread t2 = new Thread(new LoopSync());
t2.start();
while (t1.isAlive() || t2.isAlive()) {
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" testThreadsSync() 1 thread elapsed (ms): " + milSecds);
}
private static void testThreadsAtomic(){
stopwatch sw = new stopwatch();
sw.start();
Thread t1 = new Thread(new LoopAtomic());
t1.start();
Thread t2 = new Thread(new LoopAtomic());
t2.start();
while (t1.isAlive() || t2.isAlive()) {
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" testThreadsAtomic() 1 thread elapsed (ms): " + milSecds);
}
private static void testThreadsSync2(){
stopwatch sw = new stopwatch();
sw.start();
Thread t1 = new Thread(new LoopSync());
t1.start();
Thread t2 = new Thread(new LoopSync());
t2.start();
while (t1.isAlive() || t2.isAlive()) {
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" testThreadsSync() 2 threads elapsed (ms): " + milSecds);
}
private static void testThreadsAtomic2(){
stopwatch sw = new stopwatch();
sw.start();
Thread t1 = new Thread(new LoopAtomic());
t1.start();
Thread t2 = new Thread(new LoopAtomic());
t2.start();
while (t1.isAlive() || t2.isAlive()) {
}
sw.stop();
long milSecds = sw.getElapsedTime();
System.out.println(" testThreadsAtomic() 2 threads elapsed (ms): " + milSecds);
}
private static class LoopAtomic implements Runnable {
public void run() {
AtomicLong val = new AtomicLong(0);
while (val.incrementAndGet() < 10000000L) {
}
}
}
private static class LoopSync implements Runnable {
public void run() {
long val = 0;
while (val < 10000000L) {
synchronized (main.class) {
val++;
}
}
}
}
}
public class stopwatch {
private long startTime = 0;
private long stopTime = 0;
private boolean running = false;
public void start() {
this.startTime = System.currentTimeMillis();
this.running = true;
}
public void stop() {
this.stopTime = System.currentTimeMillis();
this.running = false;
}
public long getElapsedTime() {
long elapsed;
if (running) {
elapsed = (System.currentTimeMillis() - startTime);
} else {
elapsed = (stopTime - startTime);
}
return elapsed;
}
public long getElapsedTimeSecs() {
long elapsed;
if (running) {
elapsed = ((System.currentTimeMillis() - startTime) / 1000);
} else {
elapsed = ((stopTime - startTime) / 1000);
}
return elapsed;
}
// sample usage
// public static void main(String[] args) {
// StopWatch s = new StopWatch();
// s.start();
// //code you want to time goes here
// s.stop();
// System.out.println("elapsed time in milliseconds: " +
// s.getElapsedTime());
// }
}
CAS的例子：非阻塞堆棧
下面是比非阻塞自增稍微複雜一點的CAS的例子：非阻塞堆棧/ConcurrentStack 。ConcurrentStack 中的 push() 和 pop() 操做在結構上與NonblockingCounter 上類似，只是作的工做有些冒險，但願在 「提交」 工做的時候，底層假設沒有失效。push() 方法觀察當前最頂的節點，構建一個新節點放在堆棧上，而後，若是最頂端的節點在初始觀察以後沒有變化，那麼就安裝新節點。若是 CAS 失敗，意味着另外一個線程已經修改了堆棧，那麼過程就會從新開始。
public class ConcurrentStack<E> {
AtomicReference<Node<E>> head = new AtomicReference<Node<E>>();
public void push(E item) {
Node<E> newHead = new Node<E>(item);
Node<E> oldHead;
do {
oldHead = head.get();
newHead.next = oldHead;
} while (!head.compareAndSet(oldHead, newHead));
}
public E pop() {
Node<E> oldHead;
Node<E> newHead;
do {
oldHead = head.get();
if (oldHead == null)
return null;
newHead = oldHead.next;
} while (!head.compareAndSet(oldHead,newHead));
return oldHead.item;
}
static class Node<E> {
final E item;
Node<E> next;
public Node(E item) { this.item = item; }
}
}
在輕度到中度的爭用狀況下，非阻塞算法的性能會超越阻塞算法，由於 CAS 的多數時間都在第一次嘗試時就成功，而發生爭用時的開銷也不涉及線程掛起和上下文切換，只多了幾個循環迭代。沒有爭用的 CAS 要比沒有爭用的鎖便宜得多（這句話確定是真的，由於沒有爭用的鎖涉及 CAS 加上額外的處理），而爭用的 CAS 比爭用的鎖獲取涉及更短的延遲。
在高度爭用的狀況下（即有多個線程不斷爭用一個內存位置的時候），基於鎖的算法開始提供比非阻塞算法更好的吞吐率，由於當線程阻塞時，它就會中止爭用，耐心地等候輪到本身，從而避免了進一步爭用。可是，這麼高的爭用程度並不常見，由於多數時候，線程會把線程本地的計算與爭用共享數據的操做分開，從而給其餘線程使用共享數據的機會。
CAS的例子3：非阻塞鏈表
以上的示例（自增計數器和堆棧）都是很是簡單的非阻塞算法，一旦掌握了在循環中使用 CAS，就能夠容易地模仿它們。對於更復雜的數據結構，非阻塞算法要比這些簡單示例複雜得多，由於修改鏈表、樹或哈希表可能涉及對多個指針的更新。CAS 支持對單一指針的原子性條件更新，可是不支持兩個以上的指針。因此，要構建一個非阻塞的鏈表、樹或哈希表，須要找到一種方式，能夠用 CAS 更新多個指針，同時不會讓數據結構處於不一致的狀態。
在鏈表的尾部插入元素，一般涉及對兩個指針的更新：「尾」 指針老是指向列表中的最後一個元素，「下一個」 指針從過去的最後一個元素指向新插入的元素。由於須要更新兩個指針，因此須要兩個 CAS。在獨立的 CAS 中更新兩個指針帶來了兩個須要考慮的潛在問題：若是第一個 CAS 成功，而第二個 CAS 失敗，會發生什麼？若是其餘線程在第一個和第二個 CAS 之間企圖訪問鏈表，會發生什麼？
對於非複雜數據結構，構建非阻塞算法的 「技巧」 是確保數據結構總處於一致的狀態（甚至包括在線程開始修改數據結構和它完成修改之間），還要確保其餘線程不只可以判斷出第一個線程已經完成了更新仍是處在更新的中途，還可以判斷出若是第一個線程走向 AWOL，完成更新還須要什麼操做。若是線程發現了處在更新中途的數據結構，它就能夠 「幫助」 正在執行更新的線程完成更新，而後再進行本身的操做。當第一個線程回來試圖完成本身的更新時，會發現再也不須要了，返回便可，由於 CAS 會檢測到幫助線程的干預（在這種狀況下，是建設性的干預）。
這種 「幫助鄰居」 的要求，對於讓數據結構免受單個線程失敗的影響，是必需的。若是線程發現數據結構正處在被其餘線程更新的中途，而後就等候其餘線程完成更新，那麼若是其餘線程在操做中途失敗，這個線程就可能永遠等候下去。即便不出現故障，這種方式也會提供糟糕的性能，由於新到達的線程必須放棄處理器，致使上下文切換，或者等到本身的時間片過時（而這更糟）。
public class LinkedQueue <E> {
private static class Node <E> {
final E item;
final AtomicReference<Node<E>> next;
Node(E item, Node<E> next) {
this.item = item;
this.next = new AtomicReference<Node<E>>(next);
}
}
private AtomicReference<Node<E>> head
= new AtomicReference<Node<E>>(new Node<E>(null, null));
private AtomicReference<Node<E>> tail = head;
public boolean put(E item) {
Node<E> newNode = new Node<E>(item, null);
while (true) {
Node<E> curTail = tail.get();
Node<E> residue = curTail.next.get();
if (curTail == tail.get()) {
if (residue == null) /* A */ {
if (curTail.next.compareAndSet(null, newNode)) /* C */ {
tail.compareAndSet(curTail, newNode) /* D */ ;
return true;
}
} else {
tail.compareAndSet(curTail, residue) /* B */;
}
}
}
}
}
具體算法相見IBM Developerworks
Java的ConcurrentHashMap的實現原理
Java5中的ConcurrentHashMap，線程安全，設計巧妙，用桶粒度的鎖，避免了put和get中對整個map的鎖定，尤爲在get中，只對一個HashEntry作鎖定操做，性能提高是顯而易見的。
具體實現中使用了鎖分離機制，在這個帖子中有很是詳細的討論。這裏有關於Java內存模型結合ConcurrentHashMap的分析。如下是JDK6的ConcurrentHashMap的源碼：
/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/*
* This file is available under and governed by the GNU General Public
* License version 2 only, as published by the Free Software Foundation.
* However, the following notice accompanied the original version of this
* file:
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/licenses/publicdomain
*/
package java.util.concurrent;
import java.util.concurrent.locks.*;
import java.util.*;
import java.io.Serializable;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamField;
/**
* A hash table supporting full concurrency of retrievals and
* adjustable expected concurrency for updates. This class obeys the
* same functional specification as {@link java.util.Hashtable}, and
* includes versions of methods corresponding to each method of
* <tt>Hashtable</tt>. However, even though all operations are
* thread-safe, retrieval operations do <em>not</em> entail locking,
* and there is <em>not</em> any support for locking the entire table
* in a way that prevents all access. This class is fully
* interoperable with <tt>Hashtable</tt> in programs that rely on its
* thread safety but not on its synchronization details.
*
* <p> Retrieval operations (including <tt>get</tt>) generally do not
* block, so may overlap with update operations (including
* <tt>put</tt> and <tt>remove</tt>). Retrievals reflect the results
* of the most recently <em>completed</em> update operations holding
* upon their onset. For aggregate operations such as <tt>putAll</tt>
* and <tt>clear</tt>, concurrent retrievals may reflect insertion or
* removal of only some entries. Similarly, Iterators and
* Enumerations return elements reflecting the state of the hash table
* at some point at or since the creation of the iterator/enumeration.
* They do <em>not</em> throw {@link ConcurrentModificationException}.
* However, iterators are designed to be used by only one thread at a time.
*
* <p> The allowed concurrency among update operations is guided by
* the optional <tt>concurrencyLevel</tt> constructor argument
* (default <tt>16</tt>), which is used as a hint for internal sizing. The
* table is internally partitioned to try to permit the indicated
* number of concurrent updates without contention. Because placement
* in hash tables is essentially random, the actual concurrency will
* vary. Ideally, you should choose a value to accommodate as many
* threads as will ever concurrently modify the table. Using a
* significantly higher value than you need can waste space and time,
* and a significantly lower value can lead to thread contention. But
* overestimates and underestimates within an order of magnitude do
* not usually have much noticeable impact. A value of one is
* appropriate when it is known that only one thread will modify and
* all others will only read. Also, resizing this or any other kind of
* hash table is a relatively slow operation, so, when possible, it is
* a good idea to provide estimates of expected table sizes in
* constructors.
*
* <p>This class and its views and iterators implement all of the
* <em>optional</em> methods of the {@link Map} and {@link Iterator}
* interfaces.
*
* <p> Like {@link Hashtable} but unlike {@link HashMap}, this class
* does <em>not</em> allow <tt>null</tt> to be used as a key or value.
*
* <p>This class is a member of the
* <a href="{@docRoot}/../technotes/guides/collections/index.html">
* Java Collections Framework</a>.
*
* @since 1.5
* @author Doug Lea
* @param <K> the type of keys maintained by this map
* @param <V> the type of mapped values
*/
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {
private static final long serialVersionUID = 7249069246763182397L;
/*
* The basic strategy is to subdivide the table among Segments,
* each of which itself is a concurrently readable hash table. To
* reduce footprint, all but one segments are constructed only
* when first needed (see ensureSegment). To maintain visibility
* in the presence of lazy construction, accesses to segments as
* well as elements of segment's table must use volatile access,
* which is done via Unsafe within methods segmentAt etc
* below. These provide the functionality of AtomicReferenceArrays
* but reduce the levels of indirection. Additionally,
* volatile-writes of table elements and entry "next" fields
* within locked operations use the cheaper "lazySet" forms of
* writes (via putOrderedObject) because these writes are always
* followed by lock releases that maintain sequential consistency
* of table updates.
*
* Historical note: The previous version of this class relied
* heavily on "final" fields, which avoided some volatile reads at
* the expense of a large initial footprint. Some remnants of
* that design (including forced construction of segment 0) exist
* to ensure serialization compatibility.
*/
/* ---------------- Constants -------------- */
/**
* The default initial capacity for this table,
* used when not otherwise specified in a constructor.
*/
static final int DEFAULT_INITIAL_CAPACITY = 16;
/**
* The default load factor for this table, used when not
* otherwise specified in a constructor.
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/**
* The default concurrency level for this table, used when not
* otherwise specified in a constructor.
*/
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
/**
* The maximum capacity, used if a higher value is implicitly
* specified by either of the constructors with arguments. MUST
* be a power of two <= 1<<30 to ensure that entries are indexable
* using ints.
*/
static final int MAXIMUM_CAPACITY = 1 << 30;
/**
* The minimum capacity for per-segment tables. Must be a power
* of two, at least two to avoid immediate resizing on next use
* after lazy construction.
*/
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
/**
* The maximum number of segments to allow; used to bound
* constructor arguments. Must be power of two less than 1 << 24.
*/
static final int MAX_SEGMENTS = 1 << 16; // slightly conservative
/**
* Number of unsynchronized retries in size and containsValue
* methods before resorting to locking. This is used to avoid
* unbounded retries if tables undergo continuous modification
* which would make it impossible to obtain an accurate result.
*/
static final int RETRIES_BEFORE_LOCK = 2;
/* ---------------- Fields -------------- */
/**
* Mask value for indexing into segments. The upper bits of a
* key's hash code are used to choose the segment.
*/
final int segmentMask;
/**
* Shift value for indexing within segments.
*/
final int segmentShift;
/**
* The segments, each of which is a specialized hash table.
*/
final Segment<K,V>[] segments;
transient Set<K> keySet;
transient Set<Map.Entry<K,V>> entrySet;
transient Collection<V> values;
/**
* ConcurrentHashMap list entry. Note that this is never exported
* out as a user-visible Map.Entry.
*/
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
/**
* Sets next field with volatile write semantics. (See above
* about use of putOrderedObject.)
*/
final void setNext(HashEntry<K,V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
// Unsafe mechanics
static final sun.misc.Unsafe UNSAFE;
static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = HashEntry.class;
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
/**
* Gets the ith element of given table (if nonnull) with volatile
* read semantics.
*/
@SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V> entryAt(HashEntry<K,V>[] tab, int i) {
return (tab == null) ? null :
(HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)i << TSHIFT) + TBASE);
}
/**
* Sets the ith element of given table, with volatile write
* semantics. (See above about use of putOrderedObject.)
*/
static final <K,V> void setEntryAt(HashEntry<K,V>[] tab, int i,
HashEntry<K,V> e) {
UNSAFE.putOrderedObject(tab, ((long)i << TSHIFT) + TBASE, e);
}
/**
* Applies a supplemental hash function to a given hashCode, which
* defends against poor quality hash functions. This is critical
* because ConcurrentHashMap uses power-of-two length hash tables,
* that otherwise encounter collisions for hashCodes that do not
* differ in lower or upper bits.
*/
private static int hash(int h) {
// Spread bits to regularize both segment and index locations,
// using variant of single-word Wang/Jenkins hash.
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
/**
* Segments are specialized versions of hash tables. This
* subclasses from ReentrantLock opportunistically, just to
* simplify some locking and avoid separate construction.
*/
static final class Segment<K,V> extends ReentrantLock implements Serializable {
/*
* Segments maintain a table of entry lists that are always
* kept in a consistent state, so can be read (via volatile
* reads of segments and tables) without locking. This
* requires replicating nodes when necessary during table
* resizing, so the old lists can be traversed by readers
* still using old version of table.
*
* This class defines only mutative methods requiring locking.
* Except as noted, the methods of this class perform the
* per-segment versions of ConcurrentHashMap methods. (Other
* methods are integrated directly into ConcurrentHashMap
* methods.) These mutative methods use a form of controlled
* spinning on contention via methods scanAndLock and
* scanAndLockForPut. These intersperse tryLocks with
* traversals to locate nodes. The main benefit is to absorb
* cache misses (which are very common for hash tables) while
* obtaining locks so that traversal is faster once
* acquired. We do not actually use the found nodes since they
* must be re-acquired under lock anyway to ensure sequential
* consistency of updates (and in any case may be undetectably
* stale), but they will normally be much faster to re-locate.
* Also, scanAndLockForPut speculatively creates a fresh node
* to use in put if no node is found.
*/
private static final long serialVersionUID = 2249069246763182397L;
/**
* The maximum number of times to tryLock in a prescan before
* possibly blocking on acquire in preparation for a locked
* segment operation. On multiprocessors, using a bounded
* number of retries maintains cache acquired while locating
* nodes.
*/
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
/**
* The per-segment table. Elements are accessed via
* entryAt/setEntryAt providing volatile semantics.
*/
transient volatile HashEntry<K,V>[] table;
/**
* The number of elements. Accessed only either within locks
* or among other volatile reads that maintain visibility.
*/
transient int count;
/**
* The total number of mutative operations in this segment.
* Even though this may overflows 32 bits, it provides
* sufficient accuracy for stability checks in CHM isEmpty()
* and size() methods. Accessed only either within locks or
* among other volatile reads that maintain visibility.
*/
transient int modCount;
/**
* The table is rehashed when its size exceeds this threshold.
* (The value of this field is always <tt>(int)(capacity *
* loadFactor)</tt>.)
*/
transient int threshold;
/**
* The load factor for the hash table. Even though this value
* is same for all segments, it is replicated to avoid needing
* links to outer object.
* @serial
*/
final float loadFactor;
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
if (c > threshold && first != null &&
tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
/**
* Doubles size of table and repacks entries, also adding the
* given node to new table
*/
@SuppressWarnings("unchecked")
private void rehash(HashEntry<K,V> node) {
/*
* Reclassify nodes in each list to new table. Because we
* are using power-of-two expansion, the elements from
* each bin must either stay at same index, or move with a
* power of two offset. We eliminate unnecessary node
* creation by catching cases where old nodes can be
* reused because their next fields won't change.
* Statistically, at the default threshold, only about
* one-sixth of them need cloning when a table
* doubles. The nodes they replace will be garbage
* collectable as soon as they are no longer referenced by
* any reader thread that may be in the midst of
* concurrently traversing table. Entry accesses use plain
* array indexing because they are followed by volatile
* table write.
*/
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
/**
* Scans for a node containing given key while trying to
* acquire lock, creating and returning one if not found. Upon
* return, guarantees that lock is held. UNlike in most
* methods, calls to method equals are not screened: Since
* traversal speed doesn't matter, we might as well help warm
* up the associated code and accesses as well.
*
* @return a new node if key not found, else null
*/
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // negative while locating node
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
/**
* Scans for a node containing the given key while trying to
* acquire lock for a remove or replace operation. Upon
* return, guarantees that lock is held. Note that we must
* lock even if the key is not found, to ensure sequential
* consistency of updates.
*/
private void scanAndLock(Object key, int hash) {
// similar to but simpler than scanAndLockForPut
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
int retries = -1;
while (!tryLock()) {
HashEntry<K,V> f;
if (retries < 0) {
if (e == null || key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f;
retries = -1;
}
}
}
/**
* Remove; match on key only if value null, else match both.
*/
final V remove(Object key, int hash, Object value) {
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> e = entryAt(tab, index);
HashEntry<K,V> pred = null;
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)
setEntryAt(tab, index, next);
else
pred.setNext(next);
++modCount;
--count;
oldValue = v;
}
break;
}
pred = e;
e = next;
}
} finally {
unlock();
}
return oldValue;
}
final boolean replace(K key, int hash, V oldValue, V newValue) {
if (!tryLock())
scanAndLock(key, hash);
boolean replaced = false;
try {
HashEntry<K,V> e;
for (e = entryForHash(this, hash); e != null; e = e.next) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
if (oldValue.equals(e.value)) {
e.value = newValue;
++modCount;
replaced = true;
}
break;
}
}
} finally {
unlock();
}
return replaced;
}
final V replace(K key, int hash, V value) {
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
HashEntry<K,V> e;
for (e = entryForHash(this, hash); e != null; e = e.next) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
e.value = value;
++modCount;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
final void clear() {
lock();
try {
HashEntry<K,V>[] tab = table;
for (int i = 0; i < tab.length ; i++)
setEntryAt(tab, i, null);
++modCount;
count = 0;
} finally {
unlock();
}
}
}
// Accessing segments
/**
* Gets the jth element of given segment array (if nonnull) with
* volatile element access semantics via Unsafe.
*/
@SuppressWarnings("unchecked")
static final <K,V> Segment<K,V> segmentAt(Segment<K,V>[] ss, int j) {
long u = (j << SSHIFT) + SBASE;
return ss == null ? null :
(Segment<K,V>) UNSAFE.getObjectVolatile(ss, u);
}
/**
* Returns the segment for the given index, creating it and
* recording in segment table (via CAS) if not already present.
*
* @param k the index
* @return the segment
*/
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
Segment<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
// Hash-based segment and entry accesses
/**
* Get the segment for the given hash
*/
@SuppressWarnings("unchecked")
private Segment<K,V> segmentForHash(int h) {
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u);
}
/**
* Gets the table entry for the given segment and hash
*/
@SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V> entryForHash(Segment<K,V> seg, int h) {
HashEntry<K,V>[] tab;
return (seg == null || (tab = seg.table) == null) ? null :
(HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
}
/* ---------------- Public operations -------------- */
/**
* Creates a new, empty map with the specified initial
* capacity, load factor and concurrency level.
*
* @param initialCapacity the initial capacity. The implementation
* performs internal sizing to accommodate this many elements.
* @param loadFactor the load factor threshold, used to control resizing.
* Resizing may be performed when the average number of elements per
* bin exceeds this threshold.
* @param concurrencyLevel the estimated number of concurrently
* updating threads. The implementation performs internal sizing
* to try to accommodate this many threads.
* @throws IllegalArgumentException if the initial capacity is
* negative or the load factor or concurrencyLevel are
* nonpositive.
*/
@SuppressWarnings("unchecked")
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching arguments
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// create segments and segments[0]
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
/**
* Creates a new, empty map with the specified initial capacity
* and load factor and with the default concurrencyLevel (16).
*
* @param initialCapacity The implementation performs internal
* sizing to accommodate this many elements.
* @param loadFactor the load factor threshold, used to control resizing.
* Resizing may be performed when the average number of elements per
* bin exceeds this threshold.
* @throws IllegalArgumentException if the initial capacity of
* elements is negative or the load factor is nonpositive
*
* @since 1.6
*/
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL);
}
/**
* Creates a new, empty map with the specified initial capacity,
* and with default load factor (0.75) and concurrencyLevel (16).
*
* @param initialCapacity the initial capacity. The implementation
* performs internal sizing to accommodate this many elements.
* @throws IllegalArgumentException if the initial capacity of
* elements is negative.
*/
public ConcurrentHashMap(int initialCapacity) {
this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
/**
* Creates a new, empty map with a default initial capacity (16),
* load factor (0.75) and concurrencyLevel (16).
*/
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
/**
* Creates a new map with the same mappings as the given map.
* The map is created with a capacity of 1.5 times the number
* of mappings in the given map or 16 (whichever is greater),
* and a default load factor (0.75) and concurrencyLevel (16).
*
* @param m the map
*/
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1,
DEFAULT_INITIAL_CAPACITY),
DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
putAll(m);
}
/**
* Returns <tt>true</tt> if this map contains no key-value mappings.
*
* @return <tt>true</tt> if this map contains no key-value mappings
*/
public boolean isEmpty() {
/*
* Sum per-segment modCounts to avoid mis-reporting when
* elements are concurrently added and removed in one segment
* while checking another, in which case the table was never
* actually empty at any point. (The sum ensures accuracy up
* through at least 1<<31 per-segment modifications before
* recheck.) Methods size() and containsValue() use similar
* constructions for stability checks.
*/
long sum = 0L;
final Segment<K,V>[] segments = this.segments;
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
if (seg.count != 0)
return false;
sum += seg.modCount;
}
}
if (sum != 0L) { // recheck unless no modifications
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
if (seg.count != 0)
return false;
sum -= seg.modCount;
}
}
if (sum != 0L)
return false;
}
return true;
}
/**
* Returns the number of key-value mappings in this map. If the
* map contains more than <tt>Integer.MAX_VALUE</tt> elements, returns
* <tt>Integer.MAX_VALUE</tt>.
*
* @return the number of key-value mappings in this map
*/
public int size() {
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
final Segment<K,V>[] segments = this.segments;
int size;
boolean overflow; // true if size overflows 32 bits
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // first iteration isn't retry
try {
for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
if (sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
/**
* Returns the value to which the specified key is mapped,
* or {@code null} if this map contains no mapping for the key.
*
* <p>More formally, if this map contains a mapping from a key
* {@code k} to a value {@code v} such that {@code key.equals(k)},
* then this method returns {@code v}; otherwise it returns
* {@code null}. (There can be at most one such mapping.)
*
* @throws NullPointerException if the specified key is null
*/
public V get(Object key) {
int hash = hash(key.hashCode());
for (HashEntry<K,V> e = entryForHash(segmentForHash(hash), hash);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == hash && key.equals(k)))
return e.value;
}
return null;
}
/**
* Tests if the specified object is a key in this table.
*
* @param key possible key
* @return <tt>true</tt> if and only if the specified object
* is a key in this table, as determined by the
* <tt>equals</tt> method; <tt>false</tt> otherwise.
* @throws NullPointerException if the specified key is null
*/
public boolean containsKey(Object key) {
int hash = hash(key.hashCode());
for (HashEntry<K,V> e = entryForHash(segmentForHash(hash), hash);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == hash && key.equals(k)))
return true;
}
return false;
}
/**
* Returns <tt>true</tt> if this map maps one or more keys to the
* specified value. Note: This method requires a full internal
* traversal of the hash table, and so is much slower than
* method <tt>containsKey</tt>.
*
* @param value value whose presence in this map is to be tested
* @return <tt>true</tt> if this map maps one or more keys to the
* specified value
* @throws NullPointerException if the specified value is null
*/
public boolean containsValue(Object value) {
// Same idea as size()
if (value == null)
throw new NullPointerException();
final Segment<K,V>[] segments = this.segments;
boolean found = false;
long last = 0;
int retries = -1;
try {
outer: for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
long hashSum = 0L;
int sum = 0;
for (int j = 0; j < segments.length; ++j) {
HashEntry<K,V>[] tab;
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null && (tab = seg.table) != null) {
for (int i = 0 ; i < tab.length; i++) {
HashEntry<K,V> e;
for (e = entryAt(tab, i); e != null; e = e.next) {
V v = e.value;
if (v != null && value.equals(v)) {
found = true;
break outer;
}
}
}
sum += seg.modCount;
}
}
if (retries > 0 && sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return found;
}
/**
* Legacy method testing if some key maps into the specified value
* in this table. This method is identical in functionality to
* {@link #containsValue}, and exists solely to ensure
* full compatibility with class {@link java.util.Hashtable},
* which supported this method prior to introduction of the
* Java Collections framework.
* @param value a value to search for
* @return <tt>true</tt> if and only if some key maps to the
* <tt>value</tt> argument in this table as
* determined by the <tt>equals</tt> method;
* <tt>false</tt> otherwise
* @throws NullPointerException if the specified value is null
*/
public boolean contains(Object value) {
return containsValue(value);
}
/**
* Maps the specified key to the specified value in this table.
* Neither the key nor the value can be null.
*
* <p> The value can be retrieved by calling the <tt>get</tt> method
* with a key that is equal to the original key.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
* @return the previous value associated with <tt>key</tt>, or
* <tt>null</tt> if there was no mapping for <tt>key</tt>
* @throws NullPointerException if the specified key or value is null
*/
public V put(K key, V value) {
if (value == null)
throw new NullPointerException();
int hash = hash(key.hashCode());
int j = (hash >>> segmentShift) & segmentMask;
Segment<K,V> s = segmentAt(segments, j);
if (s == null)
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
/**
* {@inheritDoc}
*
* @return the previous value associated with the specified key,
* or <tt>null</tt> if there was no mapping for the key
* @throws NullPointerException if the specified key or value is null
*/
public V putIfAbsent(K key, V value) {
if (value == null)
throw new NullPointerException();
int hash = hash(key.hashCode());
int j = (hash >>> segmentShift) & segmentMask;
Segment<K,V> s = segmentAt(segments, j);
if (s == null)
s = ensureSegment(j);
return s.put(key, hash, value, true);
}
/**
* Copies all of the mappings from the specified map to this one.
* These mappings replace any mappings that this map had for any of the
* keys currently in the specified map.
*
* @param m mappings to be stored in this map
*/
public void putAll(Map<? extends K, ? extends V> m) {
for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
put(e.getKey(), e.getValue());
}
/**
* Removes the key (and its corresponding value) from this map.
* This method does nothing if the key is not in the map.
*
* @param key the key that needs to be removed
* @return the previous value associated with <tt>key</tt>, or
* <tt>null</tt> if there was no mapping for <tt>key</tt>
* @throws NullPointerException if the specified key is null
*/
public V remove(Object key) {
int hash = hash(key.hashCode());
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.remove(key, hash, null);
}
/**
* {@inheritDoc}
*
* @throws NullPointerException if the specified key is null
*/
public boolean remove(Object key, Object value) {
int hash = hash(key.hashCode());
Segment<K,V> s;
return value != null && (s = segmentForHash(hash)) != null &&
s.remove(key, hash, value) != null;
}
/**
* {@inheritDoc}
*
* @throws NullPointerException if any of the arguments are null
*/
public boolean replace(K key, V oldValue, V newValue) {
int hash = hash(key.hashCode());
if (oldValue == null || newValue == null)
throw new NullPointerException();
Segment<K,V> s = segmentForHash(hash);
return s != null && s.replace(key, hash, oldValue, newValue);
}
/**
* {@inheritDoc}
*
* @return the previous value associated with the specified key,
* or <tt>null</tt> if there was no mapping for the key
* @throws NullPointerException if the specified key or value is null
*/
public V replace(K key, V value) {
int hash = hash(key.hashCode());
if (value == null)
throw new NullPointerException();
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.replace(key, hash, value);
}
/**
* Removes all of the mappings from this map.
*/
public void clear() {
final Segment<K,V>[] segments = this.segments;
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> s = segmentAt(segments, j);
if (s != null)
s.clear();
}
}
/**
* Returns a {@link Set} view of the keys contained in this map.
* The set is backed by the map, so changes to the map are
* reflected in the set, and vice-versa. The set supports element
* removal, which removes the corresponding mapping from this map,
* via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>,
* <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt>
* operations. It does not support the <tt>add</tt> or
* <tt>addAll</tt> operations.
*
* <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
* that will never throw {@link ConcurrentModificationException},
* and guarantees to traverse elements as they existed upon
* construction of the iterator, and may (but is not guaranteed to)
* reflect any modifications subsequent to construction.
*/
public Set<K> keySet() {
Set<K> ks = keySet;
return (ks != null) ? ks : (keySet = new KeySet());
}
/**
* Returns a {@link Collection} view of the values contained in this map.
* The collection is backed by the map, so changes to the map are
* reflected in the collection, and vice-versa. The collection
* supports element removal, which removes the corresponding
* mapping from this map, via the <tt>Iterator.remove</tt>,
* <tt>Collection.remove</tt>, <tt>removeAll</tt>,
* <tt>retainAll</tt>, and <tt>clear</tt> operations. It does not
* support the <tt>add</tt> or <tt>addAll</tt> operations.
*
* <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
* that will never throw {@link ConcurrentModificationException},
* and guarantees to traverse elements as they existed upon
* construction of the iterator, and may (but is not guaranteed to)
* reflect any modifications subsequent to construction.
*/
public Collection<V> values() {
Collection<V> vs = values;
return (vs != null) ? vs : (values = new Values());
}
/**
* Returns a {@link Set} view of the mappings contained in this map.
* The set is backed by the map, so changes to the map are
* reflected in the set, and vice-versa. The set supports element
* removal, which removes the corresponding mapping from the map,
* via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>,
* <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt>
* operations. It does not support the <tt>add</tt> or
* <tt>addAll</tt> operations.
*
* <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
* that will never throw {@link ConcurrentModificationException},
* and guarantees to traverse elements as they existed upon
* construction of the iterator, and may (but is not guaranteed to)
* reflect any modifications subsequent to construction.
*/
public Set<Map.Entry<K,V>> entrySet() {
Set<Map.Entry<K,V>> es = entrySet;
return (es != null) ? es : (entrySet = new EntrySet());
}
/**
* Returns an enumeration of the keys in this table.
*
* @return an enumeration of the keys in this table
* @see #keySet()
*/
public Enumeration<K> keys() {
return new KeyIterator();
}
/**
* Returns an enumeration of the values in this table.
*
* @return an enumeration of the values in this table
* @see #values()
*/
public Enumeration<V> elements() {
return new ValueIterator();
}
/* ---------------- Iterator Support -------------- */
abstract class HashIterator {
int nextSegmentIndex;
int nextTableIndex;
HashEntry<K,V>[] currentTable;
HashEntry<K, V> nextEntry;
HashEntry<K, V> lastReturned;
HashIterator() {
nextSegmentIndex = segments.length - 1;
nextTableIndex = -1;
advance();
}
/**
* Set nextEntry to first node of next non-empty table
* (in backwards order, to simplify checks).
*/
final void advance() {
for (;;) {
if (nextTableIndex >= 0) {
if ((nextEntry = entryAt(currentTable,
nextTableIndex--)) != null)
break;
}
else if (nextSegmentIndex >= 0) {
Segment<K,V> seg = segmentAt(segments, nextSegmentIndex--);
if (seg != null && (currentTable = seg.table) != null)
nextTableIndex = currentTable.length - 1;
}
else
break;
}
}
final HashEntry<K,V> nextEntry() {
HashEntry<K,V> e = nextEntry;
if (e == null)
throw new NoSuchElementException();
lastReturned = e; // cannot assign until after null check
if ((nextEntry = e.next) == null)
advance();
return e;
}
public final boolean hasNext() { return nextEntry != null; }
public final boolean hasMoreElements() { return nextEntry != null; }
public final void remove() {
if (lastReturned == null)
throw new IllegalStateException();
ConcurrentHashMap.this.remove(lastReturned.key);
lastReturned = null;
}
}
final class KeyIterator
extends HashIterator
implements Iterator<K>, Enumeration<K>
{
public final K next() { return super.nextEntry().key; }
public final K nextElement() { return super.nextEntry().key; }
}
final class ValueIterator
extends HashIterator
implements Iterator<V>, Enumeration<V>
{
public final V next() { return super.nextEntry().value; }
public final V nextElement() { return super.nextEntry().value; }
}
/**
* Custom Entry class used by EntryIterator.next(), that relays
* setValue changes to the underlying map.
*/
final class WriteThroughEntry
extends AbstractMap.SimpleEntry<K,V>
{
WriteThroughEntry(K k, V v) {
super(k,v);
}
/**
* Set our entry's value and write through to the map. The
* value to return is somewhat arbitrary here. Since a
* WriteThroughEntry does not necessarily track asynchronous
* changes, the most recent "previous" value could be
* different from what we return (or could even have been
* removed in which case the put will re-establish). We do not
* and cannot guarantee more.
*/
public V setValue(V value) {
if (value == null) throw new NullPointerException();
V v = super.setValue(value);
ConcurrentHashMap.this.put(getKey(), value);
return v;
}
}
final class EntryIterator
extends HashIterator
implements Iterator<Entry<K,V>>
{
public Map.Entry<K,V> next() {
HashEntry<K,V> e = super.nextEntry();
return new WriteThroughEntry(e.key, e.value);
}
}
final class KeySet extends AbstractSet<K> {
public Iterator<K> iterator() {
return new KeyIterator();
}
public int size() {
return ConcurrentHashMap.this.size();
}
public boolean isEmpty() {
return ConcurrentHashMap.this.isEmpty();
}
public boolean contains(Object o) {
return ConcurrentHashMap.this.containsKey(o);
}
public boolean remove(Object o) {
return ConcurrentHashMap.this.remove(o) != null;
}
public void clear() {
ConcurrentHashMap.this.clear();
}
}
final class Values extends AbstractCollection<V> {
public Iterator<V> iterator() {
return new ValueIterator();
}
public int size() {
return ConcurrentHashMap.this.size();
}
public boolean isEmpty() {
return ConcurrentHashMap.this.isEmpty();
}
public boolean contains(Object o) {
return ConcurrentHashMap.this.containsValue(o);
}
public void clear() {
ConcurrentHashMap.this.clear();
}
}
final class EntrySet extends AbstractSet<Map.Entry<K,V>> {
public Iterator<Map.Entry<K,V>> iterator() {
return new EntryIterator();
}
public boolean contains(Object o) {
if (!(o instanceof Map.Entry))
return false;
Map.Entry<?,?> e = (Map.Entry<?,?>)o;
V v = ConcurrentHashMap.this.get(e.getKey());
return v != null && v.equals(e.getValue());
}
public boolean remove(Object o) {
if (!(o instanceof Map.Entry))
return false;
Map.Entry<?,?> e = (Map.Entry<?,?>)o;
return ConcurrentHashMap.this.remove(e.getKey(), e.getValue());
}
public int size() {
return ConcurrentHashMap.this.size();
}
public boolean isEmpty() {
return ConcurrentHashMap.this.isEmpty();
}
public void clear() {
ConcurrentHashMap.this.clear();
}
}
/* ---------------- Serialization Support -------------- */
/**
* Save the state of the <tt>ConcurrentHashMap</tt> instance to a
* stream (i.e., serialize it).
* @param s the stream
* @serialData
* the key (Object) and value (Object)
* for each key-value mapping, followed by a null pair.
* The key-value mappings are emitted in no particular order.
*/
private void writeObject(java.io.ObjectOutputStream s) throws IOException {
// force all segments for serialization compatibility
for (int k = 0; k < segments.length; ++k)
ensureSegment(k);
s.defaultWriteObject();
final Segment<K,V>[] segments = this.segments;
for (int k = 0; k < segments.length; ++k) {
Segment<K,V> seg = segmentAt(segments, k);
seg.lock();
try {
HashEntry<K,V>[] tab = seg.table;
for (int i = 0; i < tab.length; ++i) {
HashEntry<K,V> e;
for (e = entryAt(tab, i); e != null; e = e.next) {
s.writeObject(e.key);
s.writeObject(e.value);
}
}
} finally {
seg.unlock();
}
}
s.writeObject(null);
s.writeObject(null);
}
/**
* Reconstitute the <tt>ConcurrentHashMap</tt> instance from a
* stream (i.e., deserialize it).
* @param s the stream
*/
@SuppressWarnings("unchecked")
private void readObject(java.io.ObjectInputStream s)
throws IOException, ClassNotFoundException {
// Don't call defaultReadObject()
ObjectInputStream.GetField oisFields = s.readFields();
final Segment<K,V>[] oisSegments = (Segment<K,V>[])oisFields.get("segments", null);
final int ssize = oisSegments.length;
if (ssize < 1 || ssize > MAX_SEGMENTS
|| (ssize & (ssize-1)) != 0 ) // ssize not power of two
throw new java.io.InvalidObjectException("Bad number of segments:"
+ ssize);
int sshift = 0, ssizeTmp = ssize;
while (ssizeTmp > 1) {
++sshift;
ssizeTmp >>>= 1;
}
UNSAFE.putIntVolatile(this, SEGSHIFT_OFFSET, 32 - sshift);
UNSAFE.putIntVolatile(this, SEGMASK_OFFSET, ssize - 1);
UNSAFE.putObjectVolatile(this, SEGMENTS_OFFSET, oisSegments);
// Re-initialize segments to be minimally sized, and let grow.
int cap = MIN_SEGMENT_TABLE_CAPACITY;
final Segment<K,V>[] segments = this.segments;
for (int k = 0; k < segments.length; ++k) {
Segment<K,V> seg = segments[k];
if (seg != null) {
seg.threshold = (int)(cap * seg.loadFactor);
seg.table = (HashEntry<K,V>[]) new HashEntry[cap];
}
}
// Read the keys and values, and put the mappings in the table
for (;;) {
K key = (K) s.readObject();
V value = (V) s.readObject();
if (key == null)
break;
put(key, value);
}
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long SBASE;
private static final int SSHIFT;
private static final long TBASE;
private static final int TSHIFT;
private static final long SEGSHIFT_OFFSET;
private static final long SEGMASK_OFFSET;
private static final long SEGMENTS_OFFSET;
static {
int ss, ts;
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class tc = HashEntry[].class;
Class sc = Segment[].class;
TBASE = UNSAFE.arrayBaseOffset(tc);
SBASE = UNSAFE.arrayBaseOffset(sc);
ts = UNSAFE.arrayIndexScale(tc);
ss = UNSAFE.arrayIndexScale(sc);
SEGSHIFT_OFFSET = UNSAFE.objectFieldOffset(
ConcurrentHashMap.class.getDeclaredField("segmentShift"));
SEGMASK_OFFSET = UNSAFE.objectFieldOffset(
ConcurrentHashMap.class.getDeclaredField("segmentMask"));
SEGMENTS_OFFSET = UNSAFE.objectFieldOffset(
ConcurrentHashMap.class.getDeclaredField("segments"));
} catch (Exception e) {
throw new Error(e);
}
if ((ss & (ss-1)) != 0 || (ts & (ts-1)) != 0)
throw new Error("data type scale not a power of two");
SSHIFT = 31 - Integer.numberOfLeadingZeros(ss);
TSHIFT = 31 - Integer.numberOfLeadingZeros(ts);
}
}
Java的ConcurrentLinkedQueue實現方法
ConcurrentLinkedQueue也是一樣使用了CAS指令，但其性能並不高由於太多CAS操做。其源碼以下：
/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/*
* This file is available under and governed by the GNU General Public
* License version 2 only, as published by the Free Software Foundation.
* However, the following notice accompanied the original version of this
* file:
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/licenses/publicdomain
*/
package java.util.concurrent;
import java.util.*;
import java.util.concurrent.atomic.*;
/**
* An unbounded thread-safe {@linkplain Queue queue} based on linked nodes.
* This queue orders elements FIFO (first-in-first-out).
* The <em>head</em> of the queue is that element that has been on the
* queue the longest time.
* The <em>tail</em> of the queue is that element that has been on the
* queue the shortest time. New elements
* are inserted at the tail of the queue, and the queue retrieval
* operations obtain elements at the head of the queue.
* A <tt>ConcurrentLinkedQueue</tt> is an appropriate choice when
* many threads will share access to a common collection.
* This queue does not permit <tt>null</tt> elements.
*
* <p>This implementation employs an efficient "wait-free"
* algorithm based on one described in <a
* href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple,
* Fast, and Practical Non-Blocking and Blocking Concurrent Queue
* Algorithms</a> by Maged M. Michael and Michael L. Scott.
*
* <p>Beware that, unlike in most collections, the <tt>size</tt> method
* is <em>NOT</em> a constant-time operation. Because of the
* asynchronous nature of these queues, determining the current number
* of elements requires a traversal of the elements.
*
* <p>This class and its iterator implement all of the
* <em>optional</em> methods of the {@link Collection} and {@link
* Iterator} interfaces.
*
* <p>Memory consistency effects: As with other concurrent
* collections, actions in a thread prior to placing an object into a
* {@code ConcurrentLinkedQueue}
* <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
* actions subsequent to the access or removal of that element from
* the {@code ConcurrentLinkedQueue} in another thread.
*
* <p>This class is a member of the
* <a href="{@docRoot}/../technotes/guides/collections/index.html">
* Java Collections Framework</a>.
*
* @since 1.5
* @author Doug Lea
* @param <E> the type of elements held in this collection
*
*/
public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
implements Queue<E>, java.io.Serializable {
private static final long serialVersionUID = 196745693267521676L;
/*
* This is a straight adaptation of Michael & Scott algorithm.
* For explanation, read the paper. The only (minor) algorithmic
* difference is that this version supports lazy deletion of
* internal nodes (method remove(Object)) -- remove CAS'es item
* fields to null. The normal queue operations unlink but then
* pass over nodes with null item fields. Similarly, iteration
* methods ignore those with nulls.
*
* Also note that like most non-blocking algorithms in this
* package, this implementation relies on the fact that in garbage
* collected systems, there is no possibility of ABA problems due
* to recycled nodes, so there is no need to use "counted
* pointers" or related techniques seen in versions used in
* non-GC'ed settings.
*/
private static class Node<E> {
private volatile E item;
private volatile Node<E> next;
private static final
AtomicReferenceFieldUpdater<Node, Node>
nextUpdater =
AtomicReferenceFieldUpdater.newUpdater
(Node.class, Node.class, "next");
private static final
AtomicReferenceFieldUpdater<Node, Object>
itemUpdater =
AtomicReferenceFieldUpdater.newUpdater
(Node.class, Object.class, "item");
Node(E x) { item = x; }
Node(E x, Node<E> n) { item = x; next = n; }
E getItem() {
return item;
}
boolean casItem(E cmp, E val) {
return itemUpdater.compareAndSet(this, cmp, val);
}
void setItem(E val) {
itemUpdater.set(this, val);
}
Node<E> getNext() {
return next;
}
boolean casNext(Node<E> cmp, Node<E> val) {
return nextUpdater.compareAndSet(this, cmp, val);
}
void setNext(Node<E> val) {
nextUpdater.set(this, val);
}
}
private static final
AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
tailUpdater =
AtomicReferenceFieldUpdater.newUpdater
(ConcurrentLinkedQueue.class, Node.class, "tail");
private static final
AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
headUpdater =
AtomicReferenceFieldUpdater.newUpdater
(ConcurrentLinkedQueue.class, Node.class, "head");
private boolean casTail(Node<E> cmp, Node<E> val) {
return tailUpdater.compareAndSet(this, cmp, val);
}
private boolean casHead(Node<E> cmp, Node<E> val) {
return headUpdater.compareAndSet(this, cmp, val);
}
/**
* Pointer to header node, initialized to a dummy node. The first
* actual node is at head.getNext().
*/
private transient volatile Node<E> head = new Node<E>(null, null);
/** Pointer to last node on list **/
private transient volatile Node<E> tail = head;
/**
* Creates a <tt>ConcurrentLinkedQueue</tt> that is initially empty.
*/
public ConcurrentLinkedQueue() {}
/**
* Creates a <tt>ConcurrentLinkedQueue</tt>
* initially containing the elements of the given collection,
* added in traversal order of the collection's iterator.
* @param c the collection of elements to initially contain
* @throws NullPointerException if the specified collection or any
* of its elements are null
*/
public ConcurrentLinkedQueue(Collection<? extends E> c) {
for (Iterator<? extends E> it = c.iterator(); it.hasNext();)
add(it.next());
}
// Have to override just to update the javadoc
/**
* Inserts the specified element at the tail of this queue.
*
* @return <tt>true</tt> (as specified by {@link Collection#add})
* @throws NullPointerException if the specified element is null
*/
public boolean add(E e) {
return offer(e);
}
/**
* Inserts the specified element at the tail of this queue.
*
* @return <tt>true</tt> (as specified by {@link Queue#offer})
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
Node<E> n = new Node<E>(e, null);
for (;;) {
Node<E> t = tail;
Node<E> s = t.getNext();
if (t == tail) {
if (s == null) {
if (t.casNext(s, n)) {
casTail(t, n);
return true;
}
} else {
casTail(t, s);
}
}
}
}
public E poll() {
for (;;) {
Node<E> h = head;
Node<E> t = tail;
Node<E> first = h.getNext();
if (h == head) {
if (h == t) {
if (first == null)
return null;
else
casTail(t, first);
} else if (casHead(h, first)) {
E item = first.getItem();
if (item != null) {
first.setItem(null);
return item;
}
// else skip over deleted item, continue loop,
}
}
}
}
public E peek() { // same as poll except don't remove item
for (;;) {
Node<E> h = head;
Node<E> t = tail;
Node<E> first = h.getNext();
if (h == head) {
if (h == t) {
if (first == null)
return null;
else
casTail(t, first);
} else {
E item = first.getItem();
if (item != null)
return item;
else // remove deleted node and continue
casHead(h, first);
}
}
}
}
/**
* Returns the first actual (non-header) node on list. This is yet
* another variant of poll/peek; here returning out the first
* node, not element (so we cannot collapse with peek() without
* introducing race.)
*/
Node<E> first() {
for (;;) {
Node<E> h = head;
Node<E> t = tail;
Node<E> first = h.getNext();
if (h == head) {
if (h == t) {
if (first == null)
return null;
else
casTail(t, first);
} else {
if (first.getItem() != null)
return first;
else // remove deleted node and continue
casHead(h, first);
}
}
}
}
/**
* Returns <tt>true</tt> if this queue contains no elements.
*
* @return <tt>true</tt> if this queue contains no elements
*/
public boolean isEmpty() {
return first() == null;
}
/**
* Returns the number of elements in this queue. If this queue
* contains more than <tt>Integer.MAX_VALUE</tt> elements, returns
* <tt>Integer.MAX_VALUE</tt>.
*
* <p>Beware that, unlike in most collections, this method is
* <em>NOT</em> a constant-time operation. Because of the
* asynchronous nature of these queues, determining the current
* number of elements requires an O(n) traversal.
*
* @return the number of elements in this queue
*/
public int size() {
int count = 0;
for (Node<E> p = first(); p != null; p = p.getNext()) {
if (p.getItem() != null) {
// Collections.size() spec says to max out
if (++count == Integer.MAX_VALUE)
break;
}
}
return count;
}
/**
* Returns <tt>true</tt> if this queue contains the specified element.
* More formally, returns <tt>true</tt> if and only if this queue contains
* at least one element <tt>e</tt> such that <tt>o.equals(e)</tt>.
*
* @param o object to be checked for containment in this queue
* @return <tt>true</tt> if this queue contains the specified element
*/
public boolean contains(Object o) {
if (o == null) return false;
for (Node<E> p = first(); p != null; p = p.getNext()) {
E item = p.getItem();
if (item != null &&
o.equals(item))
return true;
}
return false;
}
/**
* Removes a single instance of the specified element from this queue,
* if it is present. More formally, removes an element <tt>e</tt> such
* that <tt>o.equals(e)</tt>, if this queue contains one or more such
* elements.
* Returns <tt>true</tt> if this queue contained the specified element
* (or equivalently, if this queue changed as a result of the call).
*
* @param o element to be removed from this queue, if present
* @return <tt>true</tt> if this queue changed as a result of the call
*/
public boolean remove(Object o) {
if (o == null) return false;
for (Node<E> p = first(); p != null; p = p.getNext()) {
E item = p.getItem();
if (item != null &&
o.equals(item) &&
p.casItem(item, null))
return true;
}
return false;
}
/**
* Returns an array containing all of the elements in this queue, in
* proper sequence.
*
* <p>The returned array will be "safe" in that no references to it are
* maintained by this queue. (In other words, this method must allocate
* a new array). The caller is thus free to modify the returned array.
*
* <p>This method acts as bridge between array-based and collection-based
* APIs.
*
* @return an array containing all of the elements in this queue
*/
public Object[] toArray() {
// Use ArrayList to deal with resizing.
ArrayList<E> al = new ArrayList<E>();
for (Node<E> p = first(); p != null; p = p.getNext()) {
E item = p.getItem();
if (item != null)
al.add(item);
}
return al.toArray();
}
/**
* Returns an array containing all of the elements in this queue, in
* proper sequence; the runtime type of the returned array is that of
* the specified array. If the queue fits in the specified array, it
* is returned therein. Otherwise, a new array is allocated with the
* runtime type of the specified array and the size of this queue.
*
* <p>If this queue fits in the specified array with room to spare
* (i.e., the array has more elements than this queue), the element in
* the array immediately following the end of the queue is set to
* <tt>null</tt>.
*
* <p>Like the {@link #toArray()} method, this method acts as bridge between
* array-based and collection-based APIs. Further, this method allows
* precise control over the runtime type of the output array, and may,
* under certain circumstances, be used to save allocation costs.
*
* <p>Suppose <tt>x</tt> is a queue known to contain only strings.
* The following code can be used to dump the queue into a newly
* allocated array of <tt>String</tt>:
*
* <pre>
* String[] y = x.toArray(new String[0]);</pre>
*
* Note that <tt>toArray(new Object[0])</tt> is identical in function to
* <tt>toArray()</tt>.
*
* @param a the array into which the elements of the queue are to
* be stored, if it is big enough; otherwise, a new array of the
* same runtime type is allocated for this purpose
* @return an array containing all of the elements in this queue
* @throws ArrayStoreException if the runtime type of the specified array
* is not a supertype of the runtime type of every element in
* this queue
* @throws NullPointerException if the specified array is null
*/
public <T> T[] toArray(T[] a) {
// try to use sent-in array
int k = 0;
Node<E> p;
for (p = first(); p != null && k < a.length; p = p.getNext()) {
E item = p.getItem();
if (item != null)
a[k++] = (T)item;
}
if (p == null) {
if (k < a.length)
a[k] = null;
return a;
}
// If won't fit, use ArrayList version
ArrayList<E> al = new ArrayList<E>();
for (Node<E> q = first(); q != null; q = q.getNext()) {
E item = q.getItem();
if (item != null)
al.add(item);
}
return al.toArray(a);
}
/**
* Returns an iterator over the elements in this queue in proper sequence.
* The returned iterator is a "weakly consistent" iterator that
* will never throw {@link ConcurrentModificationException},
* and guarantees to traverse elements as they existed upon
* construction of the iterator, and may (but is not guaranteed to)
* reflect any modifications subsequent to construction.
*
* @return an iterator over the elements in this queue in proper sequence
*/
public Iterator<E> iterator() {
return new Itr();
}
private class Itr implements Iterator<E> {
/**
* Next node to return item for.
*/
private Node<E> nextNode;
/**
* nextItem holds on to item fields because once we claim
* that an element exists in hasNext(), we must return it in
* the following next() call even if it was in the process of
* being removed when hasNext() was called.
*/
private E nextItem;
/**
* Node of the last returned item, to support remove.
*/
private Node<E> lastRet;
Itr() {
advance();
}
/**
* Moves to next valid node and returns item to return for
* next(), or null if no such.
*/
private E advance() {
lastRet = nextNode;
E x = nextItem;
Node<E> p = (nextNode == null)? first() : nextNode.getNext();
for (;;) {
if (p == null) {
nextNode = null;
nextItem = null;
return x;
}
E item = p.getItem();
if (item != null) {
nextNode = p;
nextItem = item;
return x;
} else // skip over nulls
p = p.getNext();
}
}
public boolean hasNext() {
return nextNode != null;
}
public E next() {
if (nextNode == null) throw new NoSuchElementException();
return advance();
}
public void remove() {
Node<E> l = lastRet;
if (l == null) throw new IllegalStateException();
// rely on a future traversal to relink.
l.setItem(null);
lastRet = null;
}
}
/**
* Save the state to a stream (that is, serialize it).
*
* @serialData All of the elements (each an <tt>E</tt>) in
* the proper order, followed by a null
* @param s the stream
*/
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
// Write out any hidden stuff
s.defaultWriteObject();
// Write out all elements in the proper order.
for (Node<E> p = first(); p != null; p = p.getNext()) {
Object item = p.getItem();
if (item != null)
s.writeObject(item);
}
// Use trailing null as sentinel
s.writeObject(null);
}
/**
* Reconstitute the Queue instance from a stream (that is,
* deserialize it).
* @param s the stream
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
// Read in capacity, and any hidden stuff
s.defaultReadObject();
head = new Node<E>(null, null);
tail = head;
// Read in all elements and place in queue
for (;;) {
E item = (E)s.readObject();
if (item == null)
break;
else
offer(item);
}
}
}
高併發環境下優化鎖或無鎖（lock-free）的設計思路
服務端編程的3大性能殺手：一、大量線程致使的線程切換開銷。二、鎖。三、非必要的內存拷貝。在高併發下,對於純內存操做來講,單線程是要比多線程快的, 能夠比較一下多線程程序在壓力測試下cpu的sy和ni百分比。高併發環境下要實現高吞吐量和線程安全，兩個思路：一個是用優化的鎖實現，一個是lock-free的無鎖結構。但非阻塞算法要比基於鎖的算法複雜得多。開發非阻塞算法是至關專業的訓練，並且要證實算法的正確也極爲困難，不只和具體的目標機器平臺和編譯器相關，並且須要複雜的技巧和嚴格的測試。雖然Lock-Free編程很是困難，可是它一般能夠帶來比基於鎖編程更高的吞吐量。因此Lock-Free編程是大有前途的技術。它在線程停止、優先級倒置以及信號安全等方面都有着良好的表現。
- 優化鎖實現的例子：Java中的ConcurrentHashMap，設計巧妙，用桶粒度的鎖和鎖分離機制，避免了put和get中對整個map的鎖定，尤爲在get中，只對一個HashEntry作鎖定操做，性能提高是顯而易見的（詳細分析見《探索 ConcurrentHashMap 高併發性的實現機制》）。
- Lock-free無鎖的例子：CAS（CPU的Compare-And-Swap指令）的利用和LMAX的disruptor無鎖消息隊列數據結構等。有興趣瞭解LMAX的disruptor無鎖消息隊列數據結構的能夠移步slideshare。
另外，在設計思路上除了儘可能減小資源爭用之外，還能夠借鑑nginx/node.js等單線程大循環的機制，用單線程或CPU數相同的線程開闢大的隊列，併發的時候任務壓入隊列，線程輪詢而後一個個順序執行。因爲每一個都採用異步I/O，沒有阻塞線程。這個大隊列可使用RabbitMQueue，或是JDK的同步隊列（性能稍差），或是使用Disruptor無鎖隊列(Java)。任務處理能夠所有放在內存（多級緩存、讀寫分離、ConcurrentHashMap、甚至分佈式緩存Redis）中進行增刪改查。最後用Quarz維護定時把緩存數據同步到DB中。固然，這只是中小型系統的思路，若是是大型分佈式系統會很是複雜，須要分而治理，用SOA的思路，參考這篇文章的圖。（注：Redis是單線程的純內存數據庫，單線程無需鎖，而Memcache是多線程的帶CAS算法，二者都使用epoll，no-blocking io）
深刻JVM的OS的無鎖非阻塞算法
若是深刻 JVM 和操做系統，會發現非阻塞算法無處不在。垃圾收集器使用非阻塞算法加快併發和平行的垃圾蒐集；調度器使用非阻塞算法有效地調度線程和進程，實現內在鎖。在 Mustang（Java 6.0）中，基於鎖的 SynchronousQueue 算法被新的非阻塞版本代替。不多有開發人員會直接使用 SynchronousQueue，可是經過 Executors.newCachedThreadPool() 工廠構建的線程池用它做爲工做隊列。比較緩存線程池性能的對比測試顯示，新的非阻塞同步隊列實現提供了幾乎是當前實現 3 倍的速度。在 Mustang 的後續版本（代碼名稱爲 Dolphin）中，已經規劃了進一步的改進。
參考文獻
相關文章
- 1. 非阻塞同步算法與CAS(Compare and Swap)無鎖算法
- 2. 【Java併發編程】九、非阻塞同步算法與CAS(Compare and Swap)無鎖算法
- 3. 非阻塞算法
- 4. 同步阻塞與異步非阻塞
- 5. 同步異步與阻塞非阻塞
- 6. 同步、異步、阻塞與非阻塞
- 7. 關於阻塞/非阻塞、同步/非同步、死鎖
- 8. 同步、異步與阻塞、非阻塞
- 9. 阻塞、非阻塞、同步與異步
- 10. 同步/異步與阻塞/非阻塞
- 更多相關文章...
- • SQLite AND/OR 運算符 - SQLite教程
- • SQL AND & OR 運算符 - SQL 教程
- • 算法總結-廣度優先算法
- • 算法總結-深度優先算法
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 升級Gradle後報錯Gradle‘s dependency cache may be corrupt (this sometimes occurs
- 2. Smarter, Not Harder
- 3. mac-2019-react-native 本地環境搭建(xcode-11.1和android studio3.5.2中Genymotion2.12.1 和VirtualBox-5.2.34 )
- 4. 查看文件中關鍵字前後幾行的內容
- 5. XXE萌新進階全攻略
- 6. Installation failed due to: ‘Connection refused: connect‘安卓studio端口占用
- 7. zabbix5.0通過agent監控winserve12
- 8. IT行業UI前景、潛力如何？
- 9. Mac Swig 3.0.12 安裝
- 10. Windows上FreeRDP-WebConnect是一個開源HTML5代理，它提供對使用RDP的任何Windows服務器和工作站的Web訪問
歡迎關注本站公眾號,獲取更多信息
相關文章