五、共享模型之無鎖

5.1 問題提出

有如下需求，保證 account.withdraw 取款方法的線程安全

interface Account {
    // 獲取余額
    Integer getBalance();

    // 取款
    void withdraw(Integer amount);

    /**
     * 方法內(nèi)會啟動 1000 個線程，每個線程做 -10 元 的操作
     * 如果初始余額為 10000 那么正確的結(jié)果應當是 0
     */
    static void demo(Account account) {
        List<Thread> ts = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            ts.add(new Thread(() -> {
                account.withdraw(10);
            }));
        }
        long start = System.nanoTime();
        ts.forEach(Thread::start);
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        long end = System.nanoTime();
        System.out.println(account.getBalance()
                + " cost: " + (end-start)/1000_000 + " ms");
    }
}

原有實現(xiàn)并不是線程安全的

class AccountUnsafe implements Account {

    private Integer balance;

    public AccountUnsafe(Integer balance) {
        this.balance = balance;
    }

    @Override
    public Integer getBalance() {
        return this.balance;
    }

    @Override
    public void withdraw(Integer amount) {
    	this.balance -= amount;
    }
}

測試代碼

public static void main(String[] args) {
        Account account = new AccountUnsafe(10000);
        Account.demo(account);
}

執(zhí)行測試代碼，某次執(zhí)行結(jié)果

280 cost: 82 ms

5.1.1 為么不安全

withdraw 方法是臨界區(qū)，會存在線程安全問題

查看下字節(jié)碼

ALOAD 0	 													   // <- this
ALOAD 0
GETFIELD cn/itcast/AccountUnsafe.balance : Ljava/lang/Integer; // <- this.balance
INVOKEVIRTUAL java/lang/Integer.intValue ()I 				   // 拆箱
ALOAD 1 													   // <- amount
INVOKEVIRTUAL java/lang/Integer.intValue ()I 				   // 拆箱
ISUB 														   // 減法
INVOKESTATIC java/lang/Integer.valueOf (I)Ljava/lang/Integer;  // 結(jié)果裝箱
PUTFIELD cn/itcast/AccountUnsafe.balance : Ljava/lang/Integer; // -> this.balance

多線程在執(zhí)行過程中可能會出現(xiàn)指令的交錯，從而結(jié)果錯誤！

5.1.2 解決思路1 - 鎖

首先想到的是給 Account 對象加鎖

class AccountUnsafe implements Account{

    private Integer balance;

    public AccountUnsafe(Integer balance) {
        this.balance = balance;
    }

    @Override
    public Integer getBalance() {
        return this.balance;
    }

    @Override
    public void withdraw(Integer amount) {
        synchronized (this){
            this.balance -= amount;
        }
    }
}

運行結(jié)果

0 cost: 122 ms

5.1.3 解決思路2 - 無鎖

class AccountCas implements Account{
    private AtomicInteger balance;

    public AccountCas(int balance) {
        this.balance = new AtomicInteger(balance);
    }

    @Override
    public Integer getBalance() {
        return balance.get();
    }

    @Override
    public void withdraw(Integer amount) {
        while (true){
            //獲取金額的最新值
            int prev = balance.get();
            //要修改的金額
            int next = prev - amount;
            //真正修改
            if(balance.compareAndSet(prev,next)){
                break;
            }
        }
    }
}

運行結(jié)果

0 cost: 63 ms

5.2 CAS 與 volatile

前面看到的 AtomicInteger 的解決方法，內(nèi)部并沒有用鎖來保護共享變量的線程安全。那么它是如何實現(xiàn)的呢？

public void withdraw(Integer amount) {
    while(true) {
        // 需要不斷嘗試，直到成功為止
        while (true) {
            // 比如拿到了舊值 1000
            int prev = balance.get();
            // 在這個基礎上 1000-10 = 990
            int next = prev - amount;
            /*
                compareAndSet 正是做這個檢查，在 set 前，先比較 prev 與當前值
                - 不一致了，next 作廢，返回 false 表示失敗
                比如，別的線程已經(jīng)做了減法，當前值已經(jīng)被減成了 990
                那么本線程的這次 990 就作廢了，進入 while 下次循環(huán)重試
                - 一致，以 next 設置為新值，返回 true 表示成功
                */
            if (balance.compareAndSet(prev, next)) {
                break;
            }
        }
    }
}

其中的關(guān)鍵是 compareAndSet，它的簡稱就是 CAS （也有 Compare And Swap 的說法），它必須是原子操作。

//判斷期望值和當前主存中的值是否一致，如果一致則將其更新成update，否則繼續(xù)重試
public final boolean compareAndSet(int expect,int update)

分析：

• 線程1拿到 balance后，進行-10運算 。在這個過程中，balance已經(jīng)被線程2修改成90了。當線程1進行cas操作時，發(fā)現(xiàn)oldBalance和newBalance不一致，就放棄進行設置next，嘗試進行下一輪

• 當進行到第三輪，cas比較時，發(fā)現(xiàn)oldBalance和newBalance一致，說明并沒有被其他線程修改，就將balance更新成 next

注意

• 其實 CAS 的底層是 lock cmpxchg 指令（X86 架構(gòu)），在單核 CPU 和多核 CPU 下都能夠保證【比較-交換】的原子性。
• 在多核狀態(tài)下，某個核執(zhí)行到帶 lock 的指令時，CPU 會讓總線鎖住，當這個核把此指令執(zhí)行完畢，再開啟總線。這個過程中不會被線程的調(diào)度機制所打斷，保證了多個線程對內(nèi)存操作的準確性，是原子的。

5.2.1 CAS和volatile的關(guān)系

獲取共享變量時，為了保證該變量的可見性，需要使用 volatile 修飾。

它可以用來修飾成員變量和靜態(tài)成員變量，可以避免線程從自己的工作緩存中查找變量的值，必須到主存中獲取它的值，線程操作 volatile 變量都是直接操作主存。即一個線程對 volatile 變量的修改，對另一個線程可見。

CAS 必須借助 volatile 才能讀取到共享變量的最新值來實現(xiàn)【比較并交換】的效果

**注意：**volatile 僅僅保證了共享變量的可見性，讓其它線程能夠看到最新值，但不能解決指令交錯問題（不能保證原子性）

5.2.2 為什么無鎖效率高

• 無鎖情況下，即使重試失敗，線程始終在高速運行，沒有停歇，而 synchronized 會讓線程在沒有獲得鎖的時候，發(fā)生上下文切換，進入阻塞。
• 打個比喻：線程就好像高速跑道上的賽車，高速運行時，速度超快，一旦發(fā)生上下文切換，就好比賽車要減速、熄火，等被喚醒又得重新打火、啟動、加速… 恢復到高速運行，代價比較大
• 但無鎖情況下，因為線程要保持運行，需要額外 CPU 的支持，CPU 在這里就好比高速跑道，沒有額外的跑道，線程想高速運行也無從談起，雖然不會進入阻塞，但由于沒有分到時間片，仍然會進入可運行狀態(tài)，還是會導致上下文切換。

所以CAS在多核CPU下才能發(fā)揮作用，而且線程數(shù)最好不要超過CPU數(shù)，否則也會發(fā)生上下文切換，影響效率。

5.2.3 CAS 的特點

結(jié)合 CAS 和 volatile 可以實現(xiàn)無鎖并發(fā)，適用于線程數(shù)少、多核 CPU 的場景下。

• CAS 是基于樂觀鎖的思想：最樂觀的估計，不怕別的線程來修改共享變量，就算改了也沒關(guān)系，我吃虧點再重試唄。（基于樂觀鎖，但本質(zhì)上沒有上鎖）
• synchronized 是基于悲觀鎖的思想：最悲觀的估計，得防著其它線程來修改共享變量，我上了鎖你們都別想改，我改完了解開鎖，你們才有機會。
• CAS 體現(xiàn)的是無鎖并發(fā)（不加鎖）、無阻塞并發(fā)(上下文切換少)，請仔細體會這兩句話的意思
  • 因為沒有使用 synchronized，所以線程不會陷入阻塞，這是效率提升的因素之一
  • 但如果競爭激烈，可以想到重試必然頻繁發(fā)生，反而效率會受影響

5.3 原子整數(shù)

J.U.C 并發(fā)包提供了：

• AtomicBoolean
• AtomicInteger
• AtomicLong

以 AtomicInteger 為例

public class Test43 {
    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(0);

        System.out.println(i.getAndIncrement());// i++   結(jié)果 i  = 1,返回 0
        System.out.println(i.incrementAndGet());// ++i   結(jié)果 i  = 2,返回 2
        System.out.println(i.getAndDecrement());// i--   結(jié)果 i  = 1,返回 2
        System.out.println(i.decrementAndGet());// --i   結(jié)果 i  = 0,返回 0

        System.out.println(i.getAndAdd(5));// 獲取并加值 結(jié)果 i = 5, 返回 0
        System.out.println(i.addAndGet(5));// 加值并獲取 結(jié)果 i = 10,返回 10

        System.out.println(i.updateAndGet(value -> value * 10));// 更新(+-*/)并獲取 結(jié)果 i=100,返回100
        System.out.println(i.getAndUpdate(value -> value / 2));//  獲取并更新       結(jié)果 i=50,返回1002
        System.out.println(i.get());//50
    }
}

updateAndGet()的原理

• 使用CAS實現(xiàn)updateAndGet（）方法

• 查看一下JDK的updateAndGet的方法源碼，和咱們實現(xiàn)的幾乎一樣！Nice！

5.4 原子引用

為什么需要原子引用類型？— 用于保證引用類型數(shù)據(jù)的原子性

• AtomicReference
• AtomicMarkableReference
• AtomicStampedReference

有如下方法

interface DecimalAccount {
    // 獲取余額
    BigDecimal getBalance();

    // 取款
    void withdraw(BigDecimal amount);

    /**
     * 方法內(nèi)會啟動 1000 個線程，每個線程做 -10 元 的操作
     * 如果初始余額為 10000 那么正確的結(jié)果應當是 0
     */
    static void demo(DecimalAccount account) {
        List <Thread> ts = new ArrayList <>();
        for (int i = 0; i < 1000; i++) {
            ts.add(new Thread(() -> {
                account.withdraw(BigDecimal.TEN);
            }));
        }
        ts.forEach(Thread::start);
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(account.getBalance());
    }
}

試著提供不同的 DecimalAccount 實現(xiàn)，實現(xiàn)安全的取款操作

5.4.1 不安全實現(xiàn)

public class Test27 {
    public static void main(String[] args) {
        DecimalAccount.demo(new DecimalAccountUnsafe(new BigDecimal("10000")));
    }
}

class DecimalAccountUnsafe implements DecimalAccount {
    BigDecimal balance;
    public DecimalAccountUnsafe(BigDecimal balance) {
        this.balance = balance;
    }
    @Override
    public BigDecimal getBalance() {
        return balance;
    }
    @Override
    public void withdraw(BigDecimal amount) {
        BigDecimal balance = this.getBalance();
        this.balance = balance.subtract(amount);
    }
}

5.4.2 安全實現(xiàn)-使用鎖

class DecimalAccountUnsafe implements DecimalAccount {
    BigDecimal balance;
    public DecimalAccountUnsafe(BigDecimal balance) {
        this.balance = balance;
    }
    @Override
    public BigDecimal getBalance() {
        return balance;
    }
    @Override
    public void withdraw(BigDecimal amount) {
        synchronized (this){
            BigDecimal balance = this.getBalance();
            this.balance = balance.subtract(amount);
        }

    }
}

5.4.3 安全實現(xiàn)-使用CAS

public class Test27 {
    public static void main(String[] args) {
        DecimalAccount.demo(new DecimalAccountCas(new BigDecimal("10000")));
    }
}

class DecimalAccountCas implements DecimalAccount{

    private AtomicReference<BigDecimal> balance;

    public DecimalAccountCas(BigDecimal balance) {
        this.balance = new AtomicReference <>(balance);
    }

    @Override
    public BigDecimal getBalance() {
        return balance.get();
    }

    @Override
    public void withdraw(BigDecimal amount) {
        while (true){
            BigDecimal prev = balance.get();
            BigDecimal next = prev.subtract(amount);
            if(balance.compareAndSet(prev,next)){
                break;
            }
        }
    }
}

5.4.4 ABA 問題及解決

1. ABA 問題

**問題描述：**一個線程t1在修改共享變量w前，另外一個線程t2也對w進行了修改，但是w修改前后值相等。最終會導致t1也會修改成功！ 我們期望的是，只有其他線程修改w了，t1本輪就修改失?。?/p>

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 15:15
 */
@Slf4j(topic = "c.Test28")
public class Test28 {

    static AtomicReference<String> ref = new AtomicReference <>("A");

    public static void main(String[] args) {
        log.debug("main start...");
        //獲取值
        //這個共享變量被其他線程修改過且修改前后值未變能否可以檢測出來呢? ---目前的寫發(fā)不可以
        String prev = ref.get();
        //如果中間有其他線程干擾，發(fā)生了ABA現(xiàn)象，則依舊可以修改成功
        other();
        Sleeper.sleep(1);
        //嘗試改為C
        log.debug("change A->C {}",ref.compareAndSet(prev,"C"));
    }

    private static void other(){
        new Thread(()->{
            log.debug("change A->B {}",ref.compareAndSet(ref.get(),"B"));
        },"t1").start();
        Sleeper.sleep(0.5);
        new Thread(()->{
            log.debug("change B->A {}",ref.compareAndSet(ref.get(),"A"));
        },"t2").start();
    }

}

輸出為：

16:03:10.444 c.Test28 [main] - main start...
16:03:10.481 c.Test28 [t1] - change A->B true
16:03:10.995 c.Test28 [t2] - change B->A true
16:03:12.009 c.Test28 [main] - change A->C true

**結(jié)論：**主線程僅能判斷出共享變量的值與最初值 A 是否相同，不能感知到這種從 A 改為 B 又 改回 A 的情況，如果主線程希望：

只要有其它線程【動過了】共享變量，那么自己的 cas 就算失敗，這時，僅比較值是不夠的，需要再加一個版本號 — 可以通過我們接下來的AtomicStampedReference 實現(xiàn)

2. AtomicStampedReference

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 15:15
 */
@Slf4j(topic = "c.Test29")
public class Test29 {

    static AtomicStampedReference <String> ref = new AtomicStampedReference <>("A",0);

    public static void main(String[] args) {
        log.debug("main start...");
        //獲取值
        //這個共享變量被其他線程修改過且修改前后值未變能否可以檢測出來呢? ---目前的寫發(fā)可以
        String prev = ref.getReference();
        //獲取版本號
        int stamp = ref.getStamp();
        log.debug("版本 {}",stamp);
        //如果中間有其他線程干擾，發(fā)生了ABA現(xiàn)象，由于含有stamp，則無法修改成功
        other();
        Sleeper.sleep(1);
        //嘗試改為C
        log.debug("change A->C {}",ref.compareAndSet(prev,"C",stamp,stamp+1));
    }
	//注意：必須先獲取版本號，再獲取值，才能解決ABA。
	//如果先獲取值在獲取版本號之前就ABA了再獲取版本號，就是ABA之后的。
    private static void other(){
        new Thread(()->{
            log.debug("change A->B {}",ref.compareAndSet(ref.getReference(),"B",ref.getStamp(),ref.getStamp()+1));
            log.debug("版本 {}",ref.getStamp());
        },"t1").start();
        Sleeper.sleep(0.5);
        new Thread(()->{
            log.debug("change B->A {}",ref.compareAndSet(ref.getReference(),"A",ref.getStamp(),ref.getStamp()+1));
            log.debug("版本 {}",ref.getStamp());
        },"t2").start();
    }

}

輸出為：

16:08:56.826 c.Test29 [main] - main start...
16:08:56.828 c.Test29 [main] - 版本 0
16:08:56.864 c.Test29 [t1] - change A->B true
16:08:56.864 c.Test29 [t1] - 版本 1
16:08:57.373 c.Test29 [t2] - change B->A true
16:08:57.373 c.Test29 [t2] - 版本 2
16:08:58.386 c.Test29 [main] - change A->C false

結(jié)論：AtomicStampedReference 可以給原子引用加上版本號，追蹤原子引用整個的變化過程，如： A -> B -> A ->C ，通過AtomicStampedReference，我們可以知道，引用變量中途被更改了幾次。

**改進：**但是有時候，并不關(guān)心引用變量更改了幾次，只是單純的關(guān)心是否更改過，所以就有了AtomicMarkableReference

3. AtomicMarkableReference

AtomicMarkableReference 可以不關(guān)心引用變量的修改次數(shù)，只關(guān)心是否被修改過 ~ ~~

如下圖：主人只關(guān)心 垃圾袋是否為空，并不關(guān)心垃圾袋被保潔倒了幾次

代碼實現(xiàn)

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 15:39
 */
@Slf4j(topic = "c.Test30")
public class Test30 {
    public static void main(String[] args) {
        GarbageBag bag = new GarbageBag("裝滿了垃圾");
        // 參數(shù)2 mark可以看做一個標記，表示垃圾裝滿了
        AtomicMarkableReference <GarbageBag> ref = new AtomicMarkableReference <>(bag, true);

        log.debug("start...");
        GarbageBag prev = ref.getReference();
        log.debug(prev.toString());

        new Thread(()->{
            log.debug("start...");
            bag.setDesc("空垃圾袋");
            ref.compareAndSet(bag,bag,true,false);
            log.debug(bag.toString());
        },"保潔阿姨").start();

        Sleeper.sleep(1);
        log.debug("想換一只新垃圾袋？");
        boolean success = ref.compareAndSet(prev, new GarbageBag("空垃圾袋"), true, false);
        log.debug("換了么？"+success);
        log.debug(ref.getReference().toString());
    }
}

class GarbageBag {
    String desc;

    public GarbageBag(String desc) {
        this.desc = desc;
    }

    public void setDesc(String desc) {
        this.desc = desc;
    }

    @Override
    public String toString() {
        return super.toString() + " " + desc;
    }
}

輸出為：

16:17:12.629 c.Test30 [main] - start...
16:17:12.631 c.Test30 [main] - com.rg.thread.GarbageBag@33e5ccce 裝滿了垃圾
16:17:12.667 c.Test30 [保潔阿姨] - start...
16:17:12.667 c.Test30 [保潔阿姨] - com.rg.thread.GarbageBag@33e5ccce 空垃圾袋
16:17:13.671 c.Test30 [main] - 想換一只新垃圾袋？
16:17:13.671 c.Test30 [main] - 換了么？false
16:17:13.671 c.Test30 [main] - com.rg.thread.GarbageBag@33e5ccce 空垃圾袋

5.5 原子數(shù)組

通過原子的方式，更新數(shù)組里的某個元素（修改元素的內(nèi)容，而非元素的引用地址）

• AtomicIntegerArray
• AtomicLongArray
• AtomicReferenceArray

注意：compareAndSet()的比較是通過引用地址比較的，之前是對String舉例，String的不可變性導致了我們每次對String的更改也導致了引用的更改

代碼演示：

public class Test31 {

    public static void main(String[] args) {
        // 普通數(shù)組
        demo(
                ()->new int[10],
                array->array.length,
                (array,index)->array[index]++,
                array->System.out.println(Arrays.toString(array))
        );
        //原子數(shù)組
        demo(
                ()->new AtomicIntegerArray(10),
                array->array.length(),
                (array,index)->array.getAndIncrement(index),
                array-> System.out.println(array)
        );
    }

    /**
     參數(shù)1，提供數(shù)組、可以是線程不安全數(shù)組或線程安全數(shù)組
     參數(shù)2，獲取數(shù)組長度的方法
     參數(shù)3，自增方法，回傳 array, index
     參數(shù)4，打印數(shù)組的方法
     */
    // supplier 提供者 無中生有  ()->結(jié)果
    // function 函數(shù)   一個參數(shù)一個結(jié)果   (參數(shù))->結(jié)果  ,  BiFunction (參數(shù)1,參數(shù)2)->結(jié)果
    // consumer 消費者 一個參數(shù)沒結(jié)果  (參數(shù))->void,      BiConsumer (參數(shù)1,參數(shù)2)->
    private static <T> void demo(
            Supplier <T> arraySupplier,
            Function <T, Integer> lengthFun,
            BiConsumer <T, Integer> putConsumer,
            Consumer <T> printConsumer ) {
        List <Thread> ts = new ArrayList <>();
        T array = arraySupplier.get();
        int length = lengthFun.apply(array);
        for (int i = 0; i < length; i++) {// 10個線程 10000 自增操作，最終每個線程的值應該都是10000
            // 每個線程對數(shù)組作 10000 次操作
            ts.add(new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    putConsumer.accept(array, j%length);
                }
            }));
        }

        ts.forEach(t -> t.start()); // 啟動所有線程
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });     // 等所有線程結(jié)束
        printConsumer.accept(array);
    }
}

輸出為：

[9984, 9983, 9979, 9985, 9985, 9985, 9985, 9982, 9982, 9986] //普通數(shù)組
[10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000]//原子數(shù)組

5.6 字段更新器

用于保證多個線程訪問對象時，保證成員變量（屬性）的安全性

• AtomicReferenceFieldUpdater // 域 字段是引用類型

• AtomicIntegerFieldUpdater //字段時Integer類型

• AtomicLongFieldUpdater //字段時Long類型

利用字段更新器，可以針對對象的某個域（Field）進行原子操作，只能配合 volatile 修飾的字段使用，否則會出現(xiàn)異常

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 17:21
 */
@Slf4j(topic = "c.Test32")
public class Test32 {
    public static void main(String[] args) {
        Student student = new Student();
        AtomicReferenceFieldUpdater <Student, String> updater = AtomicReferenceFieldUpdater.newUpdater(Student.class, String.class, "name");
        System.out.println(updater.compareAndSet(student,null,"張三"));
        System.out.println(student);
    }
}

class Student {
    //字段一定要用volatile修飾哦，否則會報異常
    volatile String name;

    @Override
    public String toString() {
        return "Student{" +
                "name='" + name + '\'' +
                '}';
    }
}

5.7 原子累加器

5.7.1 累加器性能比較

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 17:27
 */
@Slf4j(topic = "c.Test33")
public class Test33 {
    public static void main(String[] args) {
        demo(
                ()->new AtomicLong(0),
                adder->adder.getAndIncrement()
        ); 
        demo(
                ()->new LongAdder(),
                adder->adder.increment()
        );
    }
    /*
   () -> 結(jié)果    提供累加器對象
   (參數(shù)) ->     執(zhí)行累加操作
    */
    private static <T> void demo(Supplier <T> adderSupplier, Consumer <T> action) {
        T adder = adderSupplier.get();
        List <Thread> ts = new ArrayList <>();
        // 4 個線程，每人累加 50 萬
        for (int i = 0; i < 4; i++) {
            ts.add(new Thread(() -> {
                for (int j = 0; j < 500000; j++) {
                    action.accept(adder);
                }
            }));
        }
        long start = System.nanoTime();
        ts.forEach(t -> t.start());
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        long end = System.nanoTime();
        System.out.println(adder + " cost:" + (end - start) / 1000_000);
    }
}

輸出為：

2000000 cost:32
2000000 cost:14

**分析：**性能提升的原因很簡單，就是在有競爭時，設置多個累加單元，Therad-0 累加 Cell[0]，而 Thread-1 累加Cell[1]… 最后將結(jié)果匯總。這樣它們在累加時操作的不同的 Cell 變量，因此減少了 CAS 重試失敗，從而提高性能。

* 源碼之 LongAdder

LongAdder 是并發(fā)大師 @author Doug Lea （大哥李）的作品，設計的非常精巧
LongAdder 類有幾個關(guān)鍵域

// 累加單元數(shù)組, 懶惰初始化
transient volatile Cell[] cells;
// 基礎值, 如果沒有競爭, 則用 cas 累加這個域
transient volatile long base;
// 在 cells 創(chuàng)建或擴容時, 置為 1, 表示加鎖
transient volatile int cellsBusy;

注意：volatile保證共享變量的可見性，transient保證序列化下這些屬性信息不會被序列化

5.7.2 CAS鎖

//使用CAS實現(xiàn)Lock
//不推薦使用：因為會造成CPU的空轉(zhuǎn)...
@Slf4j(topic = "c.Test42")
public class LockCas {
    // 0 沒加鎖
    // 1 加鎖
    private AtomicInteger state = new AtomicInteger(0);

    public void lock() {
        while (true) {
            if (state.compareAndSet(0, 1)) {
                break;
            }
        }
    }

    public void unlock() {
        log.debug("unlock...");
        state.set(0);
    }

    public static void main(String[] args) {
        LockCas lock = new LockCas();
        new Thread(() -> {
            log.debug("begin...");
            lock.lock();
            try {
                log.debug("lock...");
                sleep(1);
            } finally {
                lock.unlock();
            }
        }).start();

        new Thread(() -> {
            log.debug("begin...");
            lock.lock();
            try {
                log.debug("lock...");
            } finally {
                lock.unlock();
            }
        }).start();
    }
}

輸出為：

16:35:41.029 c.Test42 [Thread-1] - begin...
16:35:41.028 c.Test42 [Thread-0] - begin...
16:35:41.032 c.Test42 [Thread-1] - lock...
16:35:41.032 c.Test42 [Thread-1] - unlock...
16:35:41.032 c.Test42 [Thread-0] - lock...
16:35:42.040 c.Test42 [Thread-0] - unlock...

* 原理之偽共享

其中 Cell 即為累加單元

// 防止緩存行偽共享
@sun.misc.Contended
static final class Cell {
    volatile long value;
    Cell(long x) { value = x; }
    // 最重要的方法, 用來 cas 方式進行累加, prev 表示舊值, next 表示新值
    final boolean cas(long prev, long next) {
        return UNSAFE.compareAndSwapLong(this, valueOffset, prev, next);
    }
	// 省略不重要代碼
}

關(guān)于緩存行偽共享，得從緩存說起~ ~~

1. 緩存與內(nèi)存的速度比較

因為 CPU 與 內(nèi)存的速度差異很大，需要靠預讀數(shù)據(jù)至緩存來提升效率。

而緩存以緩存行為單位，每個緩存行對應著一塊內(nèi)存，一般是 64 byte（8 個 long）

緩存的加入會造成數(shù)據(jù)副本的產(chǎn)生，即同一份數(shù)據(jù)會緩存在不同核心的緩存行中

CPU 要保證數(shù)據(jù)的一致性，如果某個 CPU 核心更改了數(shù)據(jù)，其它 CPU 核心對應的整個緩存行必須失效

• 比如核心Core1和Core2都從內(nèi)存中讀入數(shù)據(jù)a = 2 到緩存中。之后Core1把數(shù)據(jù)改成了2000，此時為了保持數(shù)據(jù)的一致性，Core2中的數(shù)據(jù)也需要修改！即把Core2中的緩存行失效，重新去內(nèi)存中取最新的數(shù)據(jù)

因為 Cell 是數(shù)組形式，在內(nèi)存中是連續(xù)存儲的，一個 Cell 為 24 字節(jié)（16 字節(jié)的對象頭和 8 字節(jié)的 value），因此緩存行可以存下 2 個的 Cell 對象（1行64個字節(jié)）。這樣問題來了：

• Core-0 要修改 Cell[0]
• Core-1 要修改 Cell[1]

無論誰修改成功，都會導致對方 Core 的緩存行失效，比如 Core-0 中 Cell[0]=6000, Cell[1]=8000 要修改為Cell[0]=6001, Cell[1]=8000 ，這時會讓 Core-1 的緩存行失效

@sun.misc.Contended 用來解決這個問題，它的原理是在使用此注解的對象或字段的前后各增加 128 字節(jié)大小的padding，從而讓 CPU 將對象預讀至緩存時占用不同的緩存行，這樣，不會造成對方緩存行的失效 。

我們把一個緩存行加入了多個cell對象稱為偽共享，注解的作用就是防止偽共享

累加主要調(diào)用下面的方法

2. add()方法

public void add(long x) {
    // as 為累加單元數(shù)組
    // b 為基礎值
    // x 為累加值
    Cell[] as;
    long b, v;
    int m;
    Cell a;
    // 進入 if 的兩個條件
    // 1. as 有值, 表示已經(jīng)發(fā)生過競爭, 進入 if
    // 2. cas 給 base 累加時失敗了, 表示 base 發(fā)生了競爭, 進入 if
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        // uncontended 表示 cell 沒有競爭
        boolean uncontended = true;
        if (
            // as 還沒有創(chuàng)建
            as == null || (m = as.length - 1) < 0 ||
            // 當前線程對應的 cell 還沒有
            (a = as[getProbe() & m]) == null ||
            // cas 給當前線程的 cell 累加失敗 uncontended=false ( a 為當前線程的 cell )
            !(uncontended = a.cas(v = a.value, v + x))
        ) {
            // 進入 cell 數(shù)組創(chuàng)建、cell 創(chuàng)建的流程
            longAccumulate(x, null, uncontended);
        }
    }
}

add流程圖

3. longAccumulate()

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
    int h;
    // 當前線程還沒有對應的 cell, 需要隨機生成一個 h 值用來將當前線程綁定到 cell
    if ((h = getProbe()) == 0) {
        // 初始化 probe
        ThreadLocalRandom.current();
        // h 對應新的 probe 值, 用來對應 cell
        h = getProbe();
        wasUncontended = true;
    }
    // collide 為 true 表示需要擴容
    boolean collide = false;
    for (; ; ) {
        Cell[] as;
        Cell a;
        int n;
        long v;
        // 已經(jīng)有了 cells
        if ((as = cells) != null && (n = as.length) > 0) {
            // cells存在，但是里面還沒有 cell
            if ((a = as[(n - 1) & h]) == null) {
               //如果cellsBusy沒有加鎖，為 cellsBusy 加鎖, 創(chuàng)建 cell
                if (cellsBusy == 0) {       // Try to attach new Cell
                     // 創(chuàng)建cell,且cell 的初始累加值為 x
                    Cell r = new Cell(x);   // Optimistically create
                    if (cellsBusy == 0 && casCellsBusy()) {//進行加鎖，并修改cellsBusy=1
                        boolean created = false;
                        try {               // Recheck under lock
                            Cell[] rs; int m, j;
                            if ((rs = cells) != null &&//判斷槽位是否為空
                                (m = rs.length) > 0 &&
                                rs[j = (m - 1) & h] == null) {//不為空則進行下一次循環(huán)嘗試
                                rs[j] = r;//為空，則將元素放到槽位上去
                                created = true;
                            }
                        } finally {
                            cellsBusy = 0;
                        }
                        if (created)//如果上面cell創(chuàng)建成功，則退出
                            break;
                        continue;           // Slot is now non-empty
                    }
                // 成功則 break, 否則繼續(xù) continue 循環(huán)
            }
            // 有競爭, 改變線程對應的 cell 來重試 cas
            else if (!wasUncontended)
                wasUncontended = true;
            // cas 嘗試累加, fn 配合 LongAccumulator 不為 null, 配合 LongAdder 為 null
            else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
                break;
            // 如果 cells 長度已經(jīng)超過了最大長度（CPU的核數(shù)）, 或者已經(jīng)擴容, 改變線程對應的 cell 來重試 cas
            else if (n >= NCPU || cells != as)
                collide = false;
            // 確保 collide 為 false 進入此分支, 就不會進入最后的 else if 進行擴容了
            else if (!collide)
                collide = true;
            //修改標志位進行加鎖
            else if (cellsBusy == 0 && casCellsBusy()) {
                // 加鎖成功, 擴容
                try {
                    if (cells == as) {      // 擴容cells
                        Cell[] rs = new Cell[n << 1];
                        for (int i = 0; i < n; ++i)
                            rs[i] = as[i];
                        cells = rs;
                    }
                } finally {
                    cellsBusy = 0;
                }
                collide = false;
                continue;        
            }
            // 改變線程對應的 cell，再重新循環(huán)
            h = advanceProbe(h);
        }
        // 還沒有 cells, 嘗試給 cellsBusy 加鎖
        //cellsBusy == 0表示還沒有其他線程為其加鎖
        //cells == as表示還有沒其他線程創(chuàng)建該cells數(shù)組（未新建）
        //casCellsBusy() 采用CAS方法將cellsBusy 從0 改為 1。這樣做是為了防止其他線程也對其進行加鎖..
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
            boolean init = false;
            // 加鎖成功, 創(chuàng)建 cell數(shù)組, 最開始長度為 2。 并創(chuàng)建一個累加單元 cell
            try {                           // Initialize table
                if (cells == as) {
                    Cell[] rs = new Cell[2];
                    //注意：雖然我們創(chuàng)建的Cell數(shù)組大小是2，但是我們只向其加入了一個元素，
                    //不到萬不得已不創(chuàng)建另外一個cell，懶惰化的思想~~~
                    rs[h & 1] = new Cell(x);
                    cells = rs;//賦值給cells
                    init = true;
                }
            } finally {//標志位設置為0，也就是解鎖
                cellsBusy = 0;
            }
            // 成功則 break，退出循環(huán);
            if (init)
                break;
        }
        // 上兩種情況失?。ㄈ缂渔i失敗）, 嘗試給 base 累加，如果失敗則重新循環(huán)
        else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
            break;
    }
}

longAccumulate 流程圖

每個線程剛進入 longAccumulate 時，會嘗試對應一個 cell 對象（找到一個坑位）

4. 獲取最終結(jié)果通過 sum 方法

//對cells數(shù)組的元素進行累加
public long sum() {
    Cell[] as = cells; Cell a;
    long sum = base;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

5.8 Unsafe

Unsafe 對象提供了非常底層的，操作內(nèi)存、線程的方法，Unsafe 對象不能直接調(diào)用，只能通過反射獲得

unsafe 與線程安全性無關(guān)。因為比較接近底層，用來操作內(nèi)存、線程，不建議我們直接使用

5.8.1 Unsafe CAS 操作

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 17:47
 */
@Slf4j(topic = "c.TestUnsafe")
public class TestUnsafe {
    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
        //使用反射獲取到Unsafe對象
        Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
        theUnsafe.setAccessible(true);
        Unsafe unsafe = (Unsafe) theUnsafe.get(null);

        System.out.println(unsafe);

        //1. 獲取域的偏移地址
        long idOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("id"));
        long nameOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("name"));

        Teacher teacher = new Teacher();
        //2. 執(zhí)行cas操作
        unsafe.compareAndSwapInt(teacher,idOffset,0,1);
        unsafe.compareAndSwapObject(teacher,nameOffset,null,"張三");

        //3. 驗證
        System.out.println(teacher);
    }
}

@Data
class Teacher{
    volatile int id;
    volatile String name;
}

輸出為：

sun.misc.Unsafe@490d6c15
Teacher(id=1, name=張三)

6.8.2 unsafe對象模擬實現(xiàn)原子整數(shù)

使用自定義的 MyAtomicInteger 實現(xiàn)之前線程安全的原子整數(shù)類 AtomicInteger文章來源地址http://www.zghlxwxcb.cn/news/detail-411723.html

/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/7/22 18:13
 */
public class Test35 {
    public static void main(String[] args) {
        Account.demo(new MyAtomicInteger(10000));
    }
}


class MyAtomicInteger implements Account {
    private volatile int value;
    private static final long valueOffset;
    private static final Unsafe UNSAFE;

    public MyAtomicInteger(int value) {
        this.value = value;
    }

    static {
        UNSAFE = UnsafeAccessor.getUnsafe();
        try {
            valueOffset = UNSAFE.objectFieldOffset(MyAtomicInteger.class.getDeclaredField("value"));
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
            throw new RuntimeException();
        }
    }

    private int getValue(){
        return value;
    }

    public void decrement(int amount){
        while (true){
            int prev = this.value;
            int next = prev-amount;
            if(UNSAFE.compareAndSwapInt(this,valueOffset,prev,next)){
                break;
            }
        }
    }

    @Override
    public Integer getBalance() {
        return getValue();
    }

    @Override
    public void withdraw(Integer amount) {
        decrement(amount);
    }
}

interface Account {
    // 獲取余額
    Integer getBalance();

    // 取款
    void withdraw(Integer amount);

    /**
     * 方法內(nèi)會啟動 1000 個線程，每個線程做 -10 元 的操作
     * 如果初始余額為 10000 那么正確的結(jié)果應當是 0
     */
    static void demo(Account account) {
        List <Thread> ts = new ArrayList <>();
        for (int i = 0; i < 1000; i++) {
            ts.add(new Thread(() -> {
                account.withdraw(10);
            }));
        }
        long start = System.nanoTime();
        ts.forEach(Thread::start);
        ts.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        long end = System.nanoTime();
        System.out.println(account.getBalance()
                + " cost: " + (end-start)/1000_000 + " ms");
    }
}

本章小結(jié)

• CAS 與 volatile
• API
  • 原子整數(shù)
  • 原子引用
  • 原子數(shù)組
  • 字段更新器
  • 原子累加器
• Unsafe

• 原理方面*
  • LongAdder 源碼
  • 偽共享

到了這里，關(guān)于《JUC并發(fā)編程 - 高級篇》05 -共享模型之無鎖 （CAS | 原子整數(shù) | 原子引用 | 原子數(shù)組 | 字段更新器 | 原子累加器 | Unsafe類 ）的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！