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

1 無鎖類的原理詳解

1.1 CAS

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

我們會(huì)發(fā)現(xiàn)，CAS的步驟太多，有沒有可能在判斷V和E相同后，正要賦值時(shí)，切換了線程，更改了值。造成了數(shù)據(jù)不一致呢？

事實(shí)上，這個(gè)擔(dān)心是多余的。CAS整一個(gè)操作過程是一個(gè)原子操作，它是由一條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;
 }

目標(biāo)值和寄存器里的值相等的話，就設(shè)置一個(gè)跳轉(zhuǎn)標(biāo)志，并且把原始數(shù)據(jù)設(shè)到目標(biāo)里面去。如果不等的話，就不設(shè)置跳轉(zhuǎn)標(biāo)志了。

Java當(dāng)中提供了很多無鎖類，下面來介紹下無鎖類。

2 無所類的使用

我們已經(jīng)知道，無鎖比阻塞效率要高得多。我們來看看Java是如何實(shí)現(xiàn)這些無鎖類的。

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方法，他的意思是，對(duì)于this這個(gè)類上的偏移量為valueOffset的變量值如果與期望值expect相同，那么把這個(gè)變量的值設(shè)為update。
其實(shí)偏移量為valueOffset的變量就是value

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

我們此前說過，CAS是有可能會(huì)失敗的，但是失敗的代價(jià)是很小的，所以一般的實(shí)現(xiàn)都是在一個(gè)無限循環(huán)體內(nèi)，直到成功為止。

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è)線程停下來，在以后的Blog中會(huì)提到）
底層的CAS操作
非公開API，在不同版本的JDK中，可能有較大差異

2.3. AtomicReference

前面已經(jīng)提到了AtomicInteger，當(dāng)然還有AtomicBoolean，AtomicLong等等，都大同小異。

這里要介紹的是AtomicReference。

AtomicReference是一種模板類

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

它可以用來封裝任意類型的數(shù)據(jù)。

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

結(jié)果：

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

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

2.4.AtomicStampedReference

我們會(huì)發(fā)現(xiàn)CAS操作還是有一個(gè)問題的

比如之前的AtomicInteger的incrementAndGet方法

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í)再切換到最開始的那個(gè)線程，由于value仍等于1，所以還是能執(zhí)行CAS操作，當(dāng)然加法是沒有問題的，如果有些情況，對(duì)數(shù)據(jù)的狀態(tài)敏感時(shí)，這樣的過程就不被允許了。
此時(shí)就需要AtomicStampedReference類。

其內(nèi)部實(shí)現(xiàn)一個(gè)Pair類來封裝值和時(shí)間戳。

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í)間戳來標(biāo)識(shí)每一次改變。

//比較設(shè)置 參數(shù)依次為：期望值 寫入新值 期望時(shí)間戳 新時(shí)間戳

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è)用戶只能充值一次。

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來實(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來實(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)不過是多了一個(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)零來算數(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就是為了解決這樣的問題產(chǎn)生的。

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

小說明：

1. Updater只能修改它可見范圍內(nèi)的變量。因?yàn)閁pdater使用反射得到這個(gè)變量。如果變量不可見，就會(huì)出錯(cuò)。比如如果某變量申明為private，就是不可行的。
2. 為了確保變量被正確的讀取，它必須是volatile類型的。如果我們?cè)写a中未申明這個(gè)類型，那么簡(jiǎn)單得申明一下就行，這不會(huì)引起什么問題。
3. 由于CAS操作會(huì)通過對(duì)象實(shí)例中的偏移量直接進(jìn)行賦值，因此，它不支持static字段（Unsafe.objectFieldOffset()不支持靜態(tài)變量）。
   

最新評(píng)論