在[高并發(fā)Java 一] 前言中已經(jīng)提到了無(wú)鎖的概念，由于在jdk源碼中有大量的無(wú)鎖應(yīng)用，所以在這里介紹下無(wú)鎖。
1 無(wú)鎖類的原理詳解
1.1 CAS
CAS算法的過(guò)程是這樣：它包含3個(gè)參數(shù)CAS(V,E,N)。V表示要更新的變量，E表示預(yù)期值，N表示新值。僅當(dāng)V
值等于E值時(shí)，才會(huì)將V的值設(shè)為N，如果V值和E值不同，則說(shuō)明已經(jīng)有其他線程做了更新，則當(dāng)前線程什么
都不做。最后，CAS返回當(dāng)前V的真實(shí)值。CAS操作是抱著樂(lè)觀的態(tài)度進(jìn)行的，它總是認(rèn)為自己可以成功完成
操作。當(dāng)多個(gè)線程同時(shí)使用CAS操作一個(gè)變量時(shí)，只有一個(gè)會(huì)勝出，并成功更新，其余均會(huì)失敗。失敗的線程
不會(huì)被掛起，僅是被告知失敗，并且允許再次嘗試，當(dāng)然也允許失敗的線程放棄操作。基于這樣的原理，CAS
操作即時(shí)沒(méi)有鎖，也可以發(fā)現(xiàn)其他線程對(duì)當(dāng)前線程的干擾，并進(jìn)行恰當(dāng)?shù)奶幚怼?/p>
我們會(huì)發(fā)現(xiàn)，CAS的步驟太多，有沒(méi)有可能在判斷V和E相同后，正要賦值時(shí)，切換了線程，更改了值。造成了數(shù)據(jù)不一致呢？
事實(shí)上，這個(gè)擔(dān)心是多余的。CAS整一個(gè)操作過(guò)程是一個(gè)原子操作，它是由一條CPU指令完成的。
1.2 CPU指令
CAS的CPU指令是cmpxchg
指令代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
/*accumulator = AL, AX, or EAX, depending on whethera byte, word, or doubleword comparison is being performed*/if(accumulator == Destination) {ZF = 1;Destination = Source;}else {ZF = 0;accumulator = Destination;} |
目標(biāo)值和寄存器里的值相等的話，就設(shè)置一個(gè)跳轉(zhuǎn)標(biāo)志，并且把原始數(shù)據(jù)設(shè)到目標(biāo)里面去。如果不等的話，就不設(shè)置跳轉(zhuǎn)標(biāo)志了。
Java當(dāng)中提供了很多無(wú)鎖類，下面來(lái)介紹下無(wú)鎖類。
2 無(wú)所類的使用
我們已經(jīng)知道，無(wú)鎖比阻塞效率要高得多。我們來(lái)看看Java是如何實(shí)現(xiàn)這些無(wú)鎖類的。
2.1. AtomicInteger
AtomicInteger和Integer一樣，都繼承與Number類
public class AtomicInteger extends Number implements java.io.Serializable
AtomicInteger里面有很多CAS操作，典型的有：
public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
這里來(lái)解釋一下unsafe.compareAndSwapInt方法，他的意思是，對(duì)于this這個(gè)類上的偏移量為valueOffset的變量值如果與期望值expect相同，那么把這個(gè)變量的值設(shè)為update。
其實(shí)偏移量為valueOffset的變量就是value
|
1
2
3
4
5
6
|
static { try { valueOffset = unsafe.objectFieldOffset (AtomicInteger.class.getDeclaredField("value")); } catch (Exception ex) { throw new Error(ex); }} |
我們此前說(shuō)過(guò)，CAS是有可能會(huì)失敗的，但是失敗的代價(jià)是很小的，所以一般的實(shí)現(xiàn)都是在一個(gè)無(wú)限循環(huán)體內(nèi)，直到成功為止。
|
1
2
3
4
5
6
7
8
|
public final int getAndIncrement() { for (;;) { int current = get(); int next = current + 1; if (compareAndSet(current, next)) return current; } } |
2.2 Unsafe
從類名就可知，Unsafe操作是非安全的操作，比如：
根據(jù)偏移量設(shè)置值（在剛剛介紹的AtomicInteger中已經(jīng)看到了這個(gè)功能）
park()（把這個(gè)線程停下來(lái)，在以后的Blog中會(huì)提到）
底層的CAS操作
非公開(kāi)API，在不同版本的JDK中，可能有較大差異
2.3. AtomicReference
前面已經(jīng)提到了AtomicInteger，當(dāng)然還有AtomicBoolean，AtomicLong等等，都大同小異。
這里要介紹的是AtomicReference。
AtomicReference是一種模板類
public class AtomicReference<V> implements java.io.Serializable
它可以用來(lái)封裝任意類型的數(shù)據(jù)。
比如String
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
package test;import java.util.concurrent.atomic.AtomicReference;public class Test{ public final static AtomicReference<String> atomicString = new AtomicReference<String>("hosee"); public static void main(String[] args) { for (int i = 0; i < 10; i++) { final int num = i; new Thread() { public void run() { try { Thread.sleep(Math.abs((int)Math.random()*100)); } catch (Exception e) { e.printStackTrace(); } if (atomicString.compareAndSet("hosee", "ztk")) { System.out.println(Thread.currentThread().getId() + "Change value"); }else { System.out.println(Thread.currentThread().getId() + "Failed"); } }; }.start(); } }} |
結(jié)果：
10Failed
13Failed
9Change value
11Failed
12Failed
15Failed
17Failed
14Failed
16Failed
18Failed
可以看到只有一個(gè)線程能夠修改值，并且后面的線程都不能再修改。
2.4.AtomicStampedReference
我們會(huì)發(fā)現(xiàn)CAS操作還是有一個(gè)問(wèn)題的
比如之前的AtomicInteger的incrementAndGet方法
|
1
2
3
4
5
6
7
8
|
public final int incrementAndGet() { for (;;) { int current = get(); int next = current + 1; if (compareAndSet(current, next)) return next; } } |
假設(shè)當(dāng)前value=1當(dāng)某線程int current = get()執(zhí)行后，切換到另一個(gè)線程，這個(gè)線程將1變成了2，然后又一個(gè)線程將2又變成了1。此時(shí)再切換到最開(kāi)始的那個(gè)線程，由于value仍等于1，所以還是能執(zhí)行CAS操作，當(dāng)然加法是沒(méi)有問(wèn)題的，如果有些情況，對(duì)數(shù)據(jù)的狀態(tài)敏感時(shí)，這樣的過(guò)程就不被允許了。
此時(shí)就需要AtomicStampedReference類。
其內(nèi)部實(shí)現(xiàn)一個(gè)Pair類來(lái)封裝值和時(shí)間戳。
|
1
2
3
4
5
6
7
8
9
10
11
|
private static class Pair<T> { final T reference; final int stamp; private Pair(T reference, int stamp) { this.reference = reference; this.stamp = stamp; } static <T> Pair<T> of(T reference, int stamp) { return new Pair<T>(reference, stamp); } } |
這個(gè)類的主要思想是加入時(shí)間戳來(lái)標(biāo)識(shí)每一次改變。
//比較設(shè)置 參數(shù)依次為：期望值 寫入新值 期望時(shí)間戳 新時(shí)間戳
|
1
2
3
4
5
6
7
8
9
10
11
12
|
public boolean compareAndSet(V expectedReference, V newReference, int expectedStamp, int newStamp) { Pair<V> current = pair; return expectedReference == current.reference && expectedStamp == current.stamp && ((newReference == current.reference && newStamp == current.stamp) || casPair(current, Pair.of(newReference, newStamp))); } |
當(dāng)期望值等于當(dāng)前值，并且期望時(shí)間戳等于現(xiàn)在的時(shí)間戳?xí)r，才寫入新值，并且更新新的時(shí)間戳。
這里舉個(gè)用AtomicStampedReference的場(chǎng)景，可能不太適合，但是想不到好的場(chǎng)景了。
場(chǎng)景背景是，某公司給余額少的用戶免費(fèi)充值，但是每個(gè)用戶只能充值一次。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
package test;import java.util.concurrent.atomic.AtomicStampedReference;public class Test{ static AtomicStampedReference<Integer> money = new AtomicStampedReference<Integer>( 19, 0); public static void main(String[] args) { for (int i = 0; i < 3; i++) { final int timestamp = money.getStamp(); new Thread() { public void run() { while (true) { while (true) { Integer m = money.getReference(); if (m < 20) { if (money.compareAndSet(m, m + 20, timestamp, timestamp + 1)) { System.out.println("充值成功，余額:" + money.getReference()); break; } } else { break; } } } }; }.start(); } new Thread() { public void run() { for (int i = 0; i < 100; i++) { while (true) { int timestamp = money.getStamp(); Integer m = money.getReference(); if (m > 10) { if (money.compareAndSet(m, m - 10, timestamp, timestamp + 1)) { System.out.println("消費(fèi)10元，余額:" + money.getReference()); break; } }else { break; } } try { Thread.sleep(100); } catch (Exception e) { // TODO: handle exception } } }; }.start(); }} |
解釋下代碼，有3個(gè)線程在給用戶充值，當(dāng)用戶余額少于20時(shí)，就給用戶充值20元。有100個(gè)線程在消費(fèi)，每次消費(fèi)10元。用戶初始有9元，當(dāng)使用AtomicStampedReference來(lái)實(shí)現(xiàn)時(shí)，只會(huì)給用戶充值一次，因?yàn)槊看尾僮魇沟脮r(shí)間戳+1。運(yùn)行結(jié)果：
充值成功，余額:39
消費(fèi)10元，余額:29
消費(fèi)10元，余額:19
消費(fèi)10元，余額:9
如果使用AtomicReference<Integer>或者 Atomic Integer來(lái)實(shí)現(xiàn)就會(huì)造成多次充值。
充值成功，余額:39
消費(fèi)10元，余額:29
消費(fèi)10元，余額:19
充值成功，余額:39
消費(fèi)10元，余額:29
消費(fèi)10元，余額:19
充值成功，余額:39
消費(fèi)10元，余額:29
2.5. AtomicIntegerArray
與AtomicInteger相比，數(shù)組的實(shí)現(xiàn)不過(guò)是多了一個(gè)下標(biāo)。
public final boolean compareAndSet(int i, int expect, int update) {
return compareAndSetRaw(checkedByteOffset(i), expect, update);
}
它的內(nèi)部只是封裝了一個(gè)普通的array
private final int[] array;
里面有意思的是運(yùn)用了二進(jìn)制數(shù)的前導(dǎo)零來(lái)算數(shù)組中的偏移量。
shift = 31 - Integer.numberOfLeadingZeros(scale);
前導(dǎo)零的意思就是比如8位表示12,00001100，那么前導(dǎo)零就是1前面的0的個(gè)數(shù)，就是4。
具體偏移量如何計(jì)算，這里就不再做介紹了。
2.6. AtomicIntegerFieldUpdater
AtomicIntegerFieldUpdater類的主要作用是讓普通變量也享受原子操作。
就比如原本有一個(gè)變量是int型，并且很多地方都應(yīng)用了這個(gè)變量，但是在某個(gè)場(chǎng)景下，想讓int型變成AtomicInteger，但是如果直接改類型，就要改其他地方的應(yīng)用。AtomicIntegerFieldUpdater就是為了解決這樣的問(wèn)題產(chǎn)生的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
package test;import java.util.concurrent.atomic.AtomicInteger;import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;public class Test{ public static class V{ int id; volatile int score; public int getScore() { return score; } public void setScore(int score) { this.score = score; } } public final static AtomicIntegerFieldUpdater<V> vv = AtomicIntegerFieldUpdater.newUpdater(V.class, "score"); public static AtomicInteger allscore = new AtomicInteger(0); public static void main(String[] args) throws InterruptedException { final V stu = new V(); Thread[] t = new Thread[10000]; for (int i = 0; i < 10000; i++) { t[i] = new Thread() { @Override public void run() { if(Math.random()>0.4) { vv.incrementAndGet(stu); allscore.incrementAndGet(); } } }; t[i].start(); } for (int i = 0; i < 10000; i++) { t[i].join(); } System.out.println("score="+stu.getScore()); System.out.println("allscore="+allscore); }} |
上述代碼將score使用 AtomicIntegerFieldUpdater變成 AtomicInteger。保證了線程安全。
這里使用allscore來(lái)驗(yàn)證，如果score和allscore數(shù)值相同，則說(shuō)明是線程安全的。
小說(shuō)明：
- Updater只能修改它可見(jiàn)范圍內(nèi)的變量。因?yàn)閁pdater使用反射得到這個(gè)變量。如果變量不可見(jiàn)，就會(huì)出錯(cuò)。比如如果某變量申明為private，就是不可行的。
- 為了確保變量被正確的讀取，它必須是volatile類型的。如果我們?cè)写a中未申明這個(gè)類型，那么簡(jiǎn)單得申明一下就行，這不會(huì)引起什么問(wèn)題。
- 由于CAS操作會(huì)通過(guò)對(duì)象實(shí)例中的偏移量直接進(jìn)行賦值，因此，它不支持static字段（Unsafe.objectFieldOffset()不支持靜態(tài)變量）。
