Java 高并發四：無鎖詳細介紹

2019-11-26 13:51:03

字體：大中小

來源：轉載

供稿：網友

在[高并發Java 一] 前言中已經提到了無鎖的概念，由于在jdk源碼中有大量的無鎖應用，所以在這里介紹下無鎖。

1 無鎖類的原理詳解

1.1 CAS

CAS算法的過程是這樣：它包含3個參數CAS(V,E,N)。V表示要更新的變量，E表示預期值，N表示新值。僅當V
值等于E值時，才會將V的值設為N，如果V值和E值不同，則說明已經有其他線程做了更新，則當前線程什么
都不做。最后，CAS返回當前V的真實值。CAS操作是抱著樂觀的態度進行的，它總是認為自己可以成功完成
操作。當多個線程同時使用CAS操作一個變量時，只有一個會勝出，并成功更新，其余均會失敗。失敗的線程
不會被掛起，僅是被告知失敗，并且允許再次嘗試，當然也允許失敗的線程放棄操作?；谶@樣的原理，CAS
操作即時沒有鎖，也可以發現其他線程對當前線程的干擾，并進行恰當的處理。

我們會發現，CAS的步驟太多，有沒有可能在判斷V和E相同后，正要賦值時，切換了線程，更改了值。造成了數據不一致呢？

事實上，這個擔心是多余的。CAS整一個操作過程是一個原子操作，它是由一條CPU指令完成的。

1.2 CPU指令

CAS的CPU指令是cmpxchg

指令代碼如下：

 /* accumulator = AL, AX, or EAX, depending on whether a byte, word, or doubleword comparison is being performed */ if(accumulator == Destination) { ZF = 1; Destination = Source; } else { ZF = 0; accumulator = Destination; }

目標值和寄存器里的值相等的話，就設置一個跳轉標志，并且把原始數據設到目標里面去。如果不等的話，就不設置跳轉標志了。

Java當中提供了很多無鎖類，下面來介紹下無鎖類。

2 無所類的使用

我們已經知道，無鎖比阻塞效率要高得多。我們來看看Java是如何實現這些無鎖類的。

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);
}

這里來解釋一下unsafe.compareAndSwapInt方法，他的意思是，對于this這個類上的偏移量為valueOffset的變量值如果與期望值expect相同，那么把這個變量的值設為update。
其實偏移量為valueOffset的變量就是value

static { try { valueOffset = unsafe.objectFieldOffset  (AtomicInteger.class.getDeclaredField("value")); } catch (Exception ex) { throw new Error(ex); }}

我們此前說過，CAS是有可能會失敗的，但是失敗的代價是很小的，所以一般的實現都是在一個無限循環體內，直到成功為止。

public final int getAndIncrement() { for (;;) {  int current = get();  int next = current + 1;  if (compareAndSet(current, next))  return current; } }

2.2 Unsafe

從類名就可知，Unsafe操作是非安全的操作，比如：

根據偏移量設置值（在剛剛介紹的AtomicInteger中已經看到了這個功能）
park()（把這個線程停下來，在以后的Blog中會提到）
底層的CAS操作
非公開API，在不同版本的JDK中，可能有較大差異

2.3. AtomicReference

前面已經提到了AtomicInteger，當然還有AtomicBoolean，AtomicLong等等，都大同小異。

這里要介紹的是AtomicReference。

AtomicReference是一種模板類

public class AtomicReference<V> implements java.io.Serializable

它可以用來封裝任意類型的數據。

比如String

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(); } }}

結果：

10Failed
13Failed
9Change value
11Failed
12Failed
15Failed
17Failed
14Failed
16Failed
18Failed

可以看到只有一個線程能夠修改值，并且后面的線程都不能再修改。

2.4.AtomicStampedReference

我們會發現CAS操作還是有一個問題的

比如之前的AtomicInteger的incrementAndGet方法

public final int incrementAndGet() { for (;;) {  int current = get();  int next = current + 1;  if (compareAndSet(current, next))  return next; } }

假設當前value=1當某線程int current = get()執行后，切換到另一個線程，這個線程將1變成了2，然后又一個線程將2又變成了1。此時再切換到最開始的那個線程，由于value仍等于1，所以還是能執行CAS操作，當然加法是沒有問題的，如果有些情況，對數據的狀態敏感時，這樣的過程就不被允許了。
此時就需要AtomicStampedReference類。

其內部實現一個Pair類來封裝值和時間戳。

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); } }

這個類的主要思想是加入時間戳來標識每一次改變。

//比較設置參數依次為：期望值寫入新值期望時間戳新時間戳

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))); }

當期望值等于當前值，并且期望時間戳等于現在的時間戳時，才寫入新值，并且更新新的時間戳。
這里舉個用AtomicStampedReference的場景，可能不太適合，但是想不到好的場景了。
場景背景是，某公司給余額少的用戶免費充值，但是每個用戶只能充值一次。

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("消費10元，余額:"  + money.getReference()); break; } }else { break; } } try { Thread.sleep(100); } catch (Exception e) { // TODO: handle exception } } }; }.start(); }}

解釋下代碼，有3個線程在給用戶充值，當用戶余額少于20時，就給用戶充值20元。有100個線程在消費，每次消費10元。用戶初始有9元，當使用AtomicStampedReference來實現時，只會給用戶充值一次，因為每次操作使得時間戳+1。運行結果：

充值成功，余額:39
消費10元，余額:29
消費10元，余額:19
消費10元，余額:9

如果使用AtomicReference<Integer>或者 Atomic Integer來實現就會造成多次充值。

充值成功，余額:39
消費10元，余額:29
消費10元，余額:19
充值成功，余額:39
消費10元，余額:29
消費10元，余額:19
充值成功，余額:39
消費10元，余額:29

2.5. AtomicIntegerArray

與AtomicInteger相比，數組的實現不過是多了一個下標。

public final boolean compareAndSet(int i, int expect, int update) {
return compareAndSetRaw(checkedByteOffset(i), expect, update);
}

它的內部只是封裝了一個普通的array

private final int[] array;

里面有意思的是運用了二進制數的前導零來算數組中的偏移量。

shift = 31 - Integer.numberOfLeadingZeros(scale);

前導零的意思就是比如8位表示12,00001100，那么前導零就是1前面的0的個數，就是4。

具體偏移量如何計算，這里就不再做介紹了。

2.6. AtomicIntegerFieldUpdater

AtomicIntegerFieldUpdater類的主要作用是讓普通變量也享受原子操作。

就比如原本有一個變量是int型，并且很多地方都應用了這個變量，但是在某個場景下，想讓int型變成AtomicInteger，但是如果直接改類型，就要改其他地方的應用。AtomicIntegerFieldUpdater就是為了解決這樣的問題產生的。

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來驗證，如果score和allscore數值相同，則說明是線程安全的。

小說明：