• Java并发编程学习笔记5——共享模型之无锁

    目录

    1、问题的提出

    2、CAS与volatile

    2.1、CAS

    2.2、volatile

    2.3、为什么无锁效率高？

    2.4、CAS的特点

    3、原子整数

    4、原子引用

    4.1、ABA问题及解决

    4.1.1、ABA问题

    4.1.2、AtomicStampedReference

    4.1.3、AtomicMarkableReference

    5、原子数组

    6、字段更新器

    7、原子累加器

    7.1、LongAdder原理分析

    7.1.1、CAS锁

    7.1.2、原理之伪共享

    8、Unsafe

    8.1、概述

    8.2、Unsafe CAS操作

    1、问题的提出

    有如下需求，保证account.withdraw取款方法的线程安全。

    1. public class TestAccount {
    2.     public static void main(String[] args) {
    3.         Account account1 = new LockAccount(10000);
    4.         Account.demo(account1);
    5.  
    6.         Account account2 = new CasAccount(10000);
    7.         Account.demo(account2);
    8.     }
    9. }
    10.  
    11. // 加锁实现
    12. class LockAccount implements Account {
    13.     private Integer balance;
    14.  
    15.     public LockAccount(Integer balance) {
    16.         this.balance = balance;
    17.     }
    18.  
    19.     @Override
    20.     public Integer getBalance() {
    21.         return balance;
    22.     }
    23.  
    24.     @Override
    25.     public void withdraw(Integer amount) {
    26.         synchronized (this) {
    27.             balance -= amount;
    28.         }
    29.     }
    30. }
    31.  
    32. // 无锁实现
    33. class CasAccount implements Account {
    34.     private AtomicInteger balance;
    35.  
    36.     public CasAccount(int balance) {
    37.         this.balance = new AtomicInteger(balance);
    38.     }
    39.  
    40.     @Override
    41.     public Integer getBalance() {
    42.         return balance.get();
    43.     }
    44.  
    45.     @Override
    46.     public void withdraw(Integer amount) {
    47.         while(true) {
    48.             // 获取余额的最新值
    49.             int prev = balance.get();
    50.             // 要修改的余额
    51.             int next = prev - amount;
    52.             // 真正修改
    53.             boolean flag;
    54.             if (flag = balance.compareAndSet(prev, next)) {
    55.                 break;
    56.             }
    57.         }
    58.     }
    59. }
    60.  
    61. interface Account {
    62.     // 获取余额
    63.     Integer getBalance();
    64.  
    65.     // 取款
    66.     void withdraw(Integer amount);
    67.  
    68.     /**
    69.      * 方法内部会启动1000个线程，每个线程做 -10元 的操作
    70.      * 如果初始余额为10000那么正确的结果应当是0
    71.      */
    72.     static void demo(Account account) {
    73.         List threads = new ArrayList<>();
    74.         for (int i = 0; i < 1000; i++) {
    75.             threads.add(new Thread(()->{
    76.                 account.withdraw(10);
    77.             }));
    78.         }
    79.         long start = System.nanoTime();
    80.         threads.forEach(Thread::start);
    81.         threads.forEach(t -> {
    82.             try {
    83.                 t.join();
    84.             } catch (InterruptedException e) {
    85.                 e.printStackTrace();
    86.             }
    87.         });
    88.         long end = System.nanoTime();
    89.         System.out.println(account.getBalance() + " cost: " + (end - start)/1000_000 + "ms");
    90.     }
    91. }
    92.  
    93. 结果：
    94. 0 cost: 125ms
    95. 0 cost: 80ms

    2、CAS与volatile

    2.1、CAS

    前面看到的AtomicInteger的解决方法，内部并没有用锁来保护共享变量的线程安全，那么它是如何实现的呢？

    1.     public void withdraw(Integer amount) {
    2.         while(true) {
    3.             // 获取余额的最新值
    4.             int prev = balance.get();
    5.             // 要修改的余额
    6.             int next = prev - amount;
    7.             // 真正修改
    8.             boolean flag;
    9.             if (flag = balance.compareAndSet(prev, next)) {
    10.                 break;
    11.             }
    12.         }
    13.     }

    其中的关键是compareAndSet()，它的简称就是CAS（也有Compare And Swap的说法），它必须是原子操作。

    注意：其实CAS的底层是lock cmpxchg指令（X86架构），在单核CPU和多核CPU下都能够保证【比较-交换】的原子性。

    在多核状态下，某个核执行到带lock的指令时，CPU会让总线锁住，当这个核把此指令执行完毕，再开启总线，这个过程中不会被线程的调度机制所打断，保证了多个线程对内存操作的准确性，是原子的。

    2.2、volatile

    获取共享变量时，为了保证该变量的可见性，需要使用volatile修饰。它可以用来修饰成员变量和静态成员变量，它可以避免线程从自己的工作缓存中查找变量的值，必须到主存中获取它的值，线程操作volatile变量都是直接操作主存。即一个线程对volatile变量的修改，对另一个线程可见。

    注意：volatile仅仅保证了共享变量的可见性，让其它线程能够看到最新值，但不能解决指令交错问题（不能保证原子性）。

    CAS必须借助volatile才能读取到共享变量的最新值来实现【比较并交换】的效果。

    2.3、为什么无锁效率高？

    无锁情况下，即使重试失败，线程始终在高速运行，没有停歇，而synchronized会让线程在没有获得锁的时候，发生上下文切换，进入阻塞。

    打个比喻。线程就好像高速跑道上的赛车，高速运行时，速度超快，一旦发生上下文切换，就好比赛车要减速、熄火，等被唤醒又得重新打火、启动、加速...恢复到高速运行，代价比较大。但无锁情况下，因为线程要保持运行，需要额外CPU的支持，CPU在这里就好比高速跑道，没有额外的跑道，线程想高速运行也无从谈起，虽然不会进入阻塞，但是由于没有分到时间片，仍然会进入可运行状态，还是会导致上下文切换。

    2.4、CAS的特点

    结合CAS和volatile可以实现无锁并发，适用于线程数少，多核CPU的场景下。

    • CAS是基于乐观锁的思想：最乐观的估计，不怕别的线程来修改共享变量，就算改了也没关系，我吃亏点再重试呗；
    • synchronized是基于悲观锁的思想：最悲观的估计，得防着其他线程来修改共享变量，我上了锁你们都别想改，我改完了解开锁，你们才有机会。
    • CAS体现的是无锁并发、无阻塞并发：① 因为没有使用synchronized，所以线程不会陷入阻塞，这是效率提升的因素之一；② 但如果竞争激烈，可以想到重试必然频繁发生，反而效率会受影响。

    3、原子整数

    J.U.C并发包提供了：① AtomicBoolean；② AtomicInteger；③ AtomicLong。

    以AtomicInteger为例：

    1. public class Test2 {
    2.     public static void main(String[] args) {
    3.         AtomicInteger i = new AtomicInteger(0);
    4.  
    5.         // 获取并自增(i=0,结果i=1,返回0),类似于i++
    6.         System.out.println(i.getAndIncrement());
    7.  
    8.         // 自增并获取(i=1,结果i=2,返回2),类似于++i
    9.         System.out.println(i.incrementAndGet());
    10.  
    11.         // 自减并获取(i=2,结果i=1,返回1),类似于 --i
    12.         System.out.println(i.decrementAndGet());
    13.  
    14.         // 获取并自减(i=1, 结果i=0, 返回1),类似于 i--
    15.         System.out.println(i.getAndDecrement());
    16.  
    17.         // 获取并加值(i=0,结果i=5,返回0)
    18.         System.out.println(i.getAndAdd(5));
    19.  
    20.         // 加值并获取(i=5, 结果i=0, 返回10)
    21.         System.out.println(i.addAndGet(5));
    22.     }
    23. }
    24.  
    25. 结果：
    26. 0
    27. 2
    28. 1
    29. 1
    30. 0
    31. 10

    原来代码的改进：

    1.     public void withdraw(Integer amount) {
    2. //        while(true) {
    3. //            // 获取余额的最新值
    4. //            int prev = balance.get();
    5. //            // 要修改的余额
    6. //            int next = prev - amount;
    7. //            // 真正修改
    8. //            boolean flag;
    9. //            if (flag = balance.compareAndSet(prev, next)) {
    10. //                break;
    11. //            }
    12. //        }
    13.         // 上述注释的代码等价于以下：
    14.         balance.getAndAdd(-amount);
    15.     }
    1. /**
    2.  * 模拟AtomicInteger中的updateAndGet()方法（使用回调函数）
    3.  */
    4. public class Test3 {
    5.     public static void main(String[] args) {
    6.         AtomicInteger i = new AtomicInteger(5);
    7.  
    8.         // lambda表达式：读取到的值 -> 设置的值
    9.         i.updateAndGet(value -> value * 10); // 50
    10.         i.getAndUpdate(value -> value * 10); // 50
    11.  
    12.         updateAndGet(i, p -> p / 2); // 250
    13.  
    14.         System.out.println(i.get());
    15.     }
    16.  
    17.     public static void updateAndGet(AtomicInteger i, IntUnaryOperator operator) {
    18.         while (true) {
    19.             // 获取当前值
    20.             int prev = i.get();
    21.             // 进行计算
    22.             int next = operator.applyAsInt(prev);
    23.             // 重新赋值结果给共享变量
    24.             if (i.compareAndSet(prev, next)) {
    25.                 break;
    26.             }
    27.         }
    28.     }
    29. }
    30.  
    31. 结果：
    32. 250

    4、原子引用

    为什么需要原子引用类型？

    • AtomicReference；
    • AtomicMarkableReference；
    • AtomicStampedReference。

    有如下方法：

    1. public class Test4 {
    2.     public static void main(String[] args) {
    3.         DecimalAccount.demo(new CasDecimalAccount(new BigDecimal(10000)));
    4.     }
    5. }
    6.  
    7. class CasDecimalAccount implements DecimalAccount {
    8.     // <>内为要保护的对象
    9.     private AtomicReference balance;
    10.  
    11.     public CasDecimalAccount(BigDecimal balance) {
    12.         this.balance = new AtomicReference<>(balance);
    13.     }
    14.  
    15.     @Override
    16.     public BigDecimal getBalance() {
    17.         return balance.get();
    18.     }
    19.  
    20.     @Override
    21.     public void withdraw(BigDecimal amount) {
    22.         while(true) {
    23.             // 获取值
    24.             BigDecimal prev = balance.get();
    25.             // 运算
    26.             BigDecimal next = prev.subtract(amount);
    27.             // 更新结果
    28.             if (balance.compareAndSet(prev, next)) {
    29.                 break;
    30.             }
    31.         }
    32.     }
    33. }
    34.  
    35. interface DecimalAccount {
    36.     // 获取余额
    37.     BigDecimal getBalance();
    38.  
    39.     // 取款
    40.     void withdraw(BigDecimal amount);
    41.  
    42.     /**
    43.      * 方法内部会启动1000个线程，每个线程做 -10元 的操作
    44.      * 如果初始余额为10000那么正确的结果应当是0
    45.      */
    46.     static void demo(DecimalAccount account) {
    47.         List threads = new ArrayList<>();
    48.         for (int i = 0; i < 1000; i++) {
    49.             threads.add(new Thread(()->{
    50.                 account.withdraw(BigDecimal.TEN);
    51.             }));
    52.         }
    53.         long start = System.nanoTime();
    54.         threads.forEach(Thread::start);
    55.         threads.forEach(t -> {
    56.             try {
    57.                 t.join();
    58.             } catch (InterruptedException e) {
    59.                 e.printStackTrace();
    60.             }
    61.         });
    62.         long end = System.nanoTime();
    63.         System.out.println(account.getBalance() + " cost: " + (end - start)/1000_000 + "ms");
    64.     }
    65. }
    66.  
    67. 结果：
    68. 0 cost: 143ms

    4.1、ABA问题及解决

    4.1.1、ABA问题

    1. /**
    2.  * A -> B -> A 问题
    3.  */
    4. public class Test5 {
    5.     private static final Logger LOGGER = LoggerFactory.getLogger(Test5.class);
    6.     static AtomicReference ref = new AtomicReference<>("A");
    7.  
    8.     public static void main(String[] args) throws InterruptedException {
    9.         LOGGER.debug("main start ...");
    10.         // 获取值A
    11.         // 这个共享变量被其他线程修改过
    12.         String prev = ref.get();
    13.         other();
    14.         Thread.sleep(1000);
    15.         // 尝试改为C（因为A改为B又改回C，主线程无法感知到共享变量被修改）
    16.         LOGGER.debug("change A to C {}", ref.compareAndSet(prev, "C"));
    17.     }
    18.     
    19.     private static void other() {
    20.         new Thread(()->{
    21.             LOGGER.debug("change A to B {}", ref.compareAndSet(ref.get(), "B"));
    22.         }, "t1").start();
    23.         new Thread(()-> {
    24.             LOGGER.debug("change B to A {}", ref.compareAndSet(ref.get(), "A"));
    25.         },"t2").start();
    26.     }
    27. }
    28.  
    29. 结果：
    30. 10:14:49.832 [main] DEBUG com.multiThreads.test16.Test5 - main start ...
    31. 10:14:49.876 [t2] DEBUG com.multiThreads.test16.Test5 - change B to A true
    32. 10:14:49.875 [t1] DEBUG com.multiThreads.test16.Test5 - change A to B true
    33. 10:14:50.875 [main] DEBUG com.multiThreads.test16.Test5 - change A to C true

    主线程仅能判断出共享变量的值与最初值A是否相同，不能感知到这种从A改为B又改回A的情况，如果主线程希望：只要有其他线程【动过了】共享变量，那么自己的CAS就算失败，这时，仅比较值是不够的，需要再加一个版本号。

    4.1.2、AtomicStampedReference

    1. public class Test6 {
    2.     private static final Logger LOGGER = LoggerFactory.getLogger(Test5.class);
    3.     static AtomicStampedReference ref = new AtomicStampedReference<>("A", 0);
    4.  
    5.     public static void main(String[] args) throws InterruptedException {
    6.         LOGGER.debug("main start ...");
    7.         // 获取值A
    8.         // 这个共享变量被其他线程修改过
    9.         String prev = ref.getReference();
    10.         // 获取版本号
    11.         int stamp = ref.getStamp();
    12.         LOGGER.debug("main before {}", stamp);
    13.         other();
    14.         Thread.sleep(1000);
    15.         // 尝试改为C
    16.         LOGGER.debug("main before {}", stamp);
    17.         // 额外参数为原版本号和新版本号，会比较版本号的值来解决ABA问题
    18.         LOGGER.debug("change A to C {}", ref.compareAndSet(prev, "C", stamp, stamp + 1));
    19.     }
    20.  
    21.     private static void other() {
    22.         new Thread(() -> {
    23.             int stamp1 = ref.getStamp();
    24.             LOGGER.debug("t1 before {}", stamp1);
    25.             LOGGER.debug("change A to B {}", ref.compareAndSet(ref.getReference(), "B", stamp1, stamp1 + 1));
    26.         }, "t1").start();
    27.         new Thread(() -> {
    28.             int stamp2 = ref.getStamp();
    29.             LOGGER.debug("t2 before {}", stamp2);
    30.             LOGGER.debug("change B to A {}", ref.compareAndSet(ref.getReference(), "A", stamp2, stamp2 + 1));
    31.         }, "t2").start();
    32.     }
    33. }
    34.  
    35. 结果：
    36. 10:27:15.667 [main] DEBUG com.multiThreads.test16.Test5 - main start ...
    37. 10:27:15.670 [main] DEBUG com.multiThreads.test16.Test5 - main before 0
    38. 10:27:15.729 [t1] DEBUG com.multiThreads.test16.Test5 - t1 before 0
    39. 10:27:15.729 [t1] DEBUG com.multiThreads.test16.Test5 - change A to B true
    40. 10:27:15.730 [t2] DEBUG com.multiThreads.test16.Test5 - t2 before 1
    41. 10:27:15.730 [t2] DEBUG com.multiThreads.test16.Test5 - change B to A true
    42. 10:27:16.730 [main] DEBUG com.multiThreads.test16.Test5 - main before 0
    43. 10:27:16.730 [main] DEBUG com.multiThreads.test16.Test5 - change A to C false

    AtomicStampedReference可以给原子引用加上版本号，追踪原子引用整个的变化过程，如：A -> B -> A -> C，通过AtomicStampedReference，我们可以知道，引用变量中途被更改了几次。

    但是有时候，并不关心引用变量更改了几次，只是单纯地关心是否更改过，所以就有了AtomicMarkableReference。

    4.1.3、AtomicMarkableReference

    1. public class Test7 {
    2.     private static final Logger LOGGER = LoggerFactory.getLogger(Test7.class);
    3.  
    4.     public static void main(String[] args) throws InterruptedException {
    5.         GarbageBag bag = new GarbageBag("装满了垃圾");
    6.         // 参数2 mark 可以看作一个标记，表示垃圾袋满了
    7.         AtomicMarkableReference ref = new AtomicMarkableReference<>(bag, true);
    8.  
    9.         LOGGER.debug("start");
    10.         GarbageBag prev = ref.getReference();
    11.         LOGGER.debug(prev.toString());
    12.  
    13.         new Thread(()->{
    14.             LOGGER.debug("start...");
    15.             bag.setBag("空垃圾袋");
    16.             // 不换垃圾袋，仅拿出垃圾
    17.             ref.compareAndSet(bag, bag, true, false);
    18.             LOGGER.debug(bag.toString());
    19.         }, "保洁阿姨").start();
    20.  
    21.         Thread.sleep(1000);
    22.         LOGGER.debug("想换一只新垃圾袋？");
    23.         boolean success = ref.compareAndSet(prev, new GarbageBag("空垃圾袋"), true, false);
    24.         LOGGER.debug("换了么？" + success);
    25.         LOGGER.debug(ref.getReference().toString());
    26.     }
    27. }
    28.  
    29. @Data
    30. class GarbageBag {
    31.     private String bag;
    32.     public GarbageBag(String bag) {
    33.         this.bag = bag;
    34.     }
    35.  
    36.     @Override
    37.     public String toString() {
    38.         return bag;
    39.     }
    40. }
    41.  
    42. 结果：
    43. 11:05:12.166 [main] DEBUG com.multiThreads.test16.Test7 - start
    44. 11:05:12.169 [main] DEBUG com.multiThreads.test16.Test7 - 装满了垃圾
    45. 11:05:12.229 [保洁阿姨] DEBUG com.multiThreads.test16.Test7 - start...
    46. 11:05:12.229 [保洁阿姨] DEBUG com.multiThreads.test16.Test7 - 空垃圾袋
    47. 11:05:13.228 [main] DEBUG com.multiThreads.test16.Test7 - 想换一只新垃圾袋？
    48. 11:05:13.228 [main] DEBUG com.multiThreads.test16.Test7 - 换了么？false
    49. 11:05:13.228 [main] DEBUG com.multiThreads.test16.Test7 - 空垃圾袋

    5、原子数组

    • AtomicIntegerArray；
    • AtomicLongArray；
    • AtomicReferenceArray。

    有如下方法：

    1. public class Test8 {
    2.     public static void main(String[] args) {
    3.         demo(
    4.                 () -> new int[10],
    5.                 (array) -> array.length,
    6.                 (array, index) -> array[index]++,
    7.                 array -> System.out.println(Arrays.toString(array))
    8.         );
    9.  
    10.         demo(
    11.                 () -> new AtomicIntegerArray(10),
    12.                 (array) -> array.length(),
    13.                 (array, index) -> array.getAndIncrement(index),
    14.                 array -> System.out.println(array)
    15.         );
    16.     }
    17.  
    18.     /**
    19.      * 方法内会启动10个线程，并发让数组所有元素总共自增10000次
    20.      *
    21.      * @param arraySupplier：Supplier：提供者，()->结果。提供数组、可以是线程不安全数组或线程安全数组
    22.      * @param lengthFun：Function：函数，(参数)->结果，还有一种为BiFunction 为 (参数1，参数2) -> 结果。获取数组长度的方法
    23.      * @param putConsumer：Consumer：消费者,一个参数，没结果，(参数)->void。BiConsumer为(参数1，参数2)->{} 自增方法，回传array，length，index。
    24.      * @param printConsumer：打印数组的方法
    25.      * @param 
    26.      */
    27.     private static  void demo(
    28.             Supplier arraySupplier, Function lengthFun,
    29.             BiConsumer putConsumer, Consumer printConsumer) {
    30.         List ts = new ArrayList<>();
    31.         T array = arraySupplier.get();
    32.         int length = lengthFun.apply(array);
    33.         for (int i = 0; i < length; i++) {
    34.             // 每个线程对数组作 10000 次操作
    35.             ts.add(new Thread(() -> {
    36.                 for (int j = 0; j < 10000; j++) {
    37.                     putConsumer.accept(array, j % length);
    38.                 }
    39.             }));
    40.         }
    41.  
    42.         ts.forEach(t -> t.start()); // 启动所有线程
    43.         ts.forEach(t -> {
    44.             try {
    45.                 t.join();
    46.             } catch (InterruptedException e) {
    47.                 e.printStackTrace();
    48.             }
    49.         }); // 等所有线程结束
    50.         printConsumer.accept(array);
    51.     }
    52. }
    53.  
    54. 结果：
    55. [9780, 9779, 9782, 9788, 9778, 9772, 9777, 9776, 9776, 9780]
    56. [10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000, 10000]

    6、字段更新器

    • AtomicReferenceFieldUpdater // 域 字段；
    • AtomicIntegerFieldUpdater；
    • AtomicLongFieldUpdater。

    利用字段更新器，可以针对对象的某个域（Field）进行原子操作，只能配合volatile修饰的字段使用，否则会出现异常。

    1. public class Test9 {
    2.     public static void main(String[] args) {
    3.         Student stu = new Student();
    4.  
    5.         AtomicReferenceFieldUpdater updater = AtomicReferenceFieldUpdater.newUpdater(Student.class, String.class, "name");
    6.  
    7.         // 参数：要更新的对象，原来的值，更新的值
    8.         System.out.println(updater.compareAndSet(stu, null, "张三"));
    9.         System.out.println(stu);
    10.     }
    11. }
    12.  
    13. class Student {
    14.     volatile String name;
    15.  
    16.     @Override
    17.     public String toString() {
    18.         return "Student{" +
    19.                 "name='" + name + '\'' +
    20.                 '}';
    21.     }
    22. }
    23.  
    24. 结果：
    25. true
    26. Student{name='张三'}

    7、原子累加器

    累加器性能比较。

    1. public class Test10 {
    2.     public static void main(String[] args) {
    3.         // 使用原子整数
    4.         for (int i = 0; i < 5; i++) {
    5.             demo(
    6.                     () -> new AtomicLong(0),
    7.                     (adder) -> adder.getAndIncrement()
    8.             );
    9.         }
    10.  
    11.         System.out.println();
    12.  
    13.         // 使用原子累加器
    14.         for (int i = 0; i < 5; i++) {
    15.             demo(
    16.                     () -> new LongAdder(),
    17.                     adder -> adder.increment()
    18.             );
    19.         }
    20.     }
    21.  
    22.     /**
    23.      *
    24.      * @param adderSupplier：提供累加器对象
    25.      * @param action：执行累加操作
    26.      * @param 
    27.      */
    28.     private static  void demo(Supplier adderSupplier, Consumer action) {
    29.         T adder = adderSupplier.get();
    30.         ArrayList ts = new ArrayList<>();
    31.         // 4 个线程，每人累加50万
    32.         for (int i = 0; i < 4; i++) {
    33.             ts.add(new Thread(()->{
    34.                 for (int j = 0; j < 500000; j++) {
    35.                     action.accept(adder);
    36.                 }
    37.             }));
    38.         }
    39.         long start = System.nanoTime();
    40.         ts.forEach(t -> t.start());
    41.         ts.forEach(t -> {
    42.             try {
    43.                 t.join();
    44.             } catch (InterruptedException e) {
    45.                 e.printStackTrace();
    46.             }
    47.         });
    48.  
    49.         long end = System.nanoTime();
    50.         System.out.println(adder + " cost:" + (end - start)/1000_000);
    51.     }
    52. }
    53.  
    54. 结果：
    55. 2000000 cost:26
    56. 2000000 cost:26
    57. 2000000 cost:30
    58. 2000000 cost:31
    59. 2000000 cost:30
    60.  
    61. 2000000 cost:14
    62. 2000000 cost:9
    63. 2000000 cost:9
    64. 2000000 cost:9
    65. 2000000 cost:8

    性能提升的原因很简单，就是在竞争时，设置多个累加单元，Thread-0累加Cell[0]，而Thread-1累加Cell[1]...最后将结果汇总。这样它们在累加时操作不同的Cell变量，因此减少了CAS重试失败次数，从而提高性能。

    7.1、LongAdder原理分析

    LongAdder是并发大师@author Doug Lea （大哥李）的作品，设计得非常精巧。

    LongAdder类有几个关键域：

    1. // 累加单元数组，懒惰初始化
    2. transient volatile Cell[] cells;
    3.  
    4. // 基础值，如果没有竞争，则用cas累加这个域
    5. transient volatile long base;
    6.  
    7. // 在cells创建或扩容时，置为1，表示加锁
    8. transient volatile int cellsBusy;

    7.1.1、CAS锁

    1. // 实际尽量不要使用CasLock
    2. class CasLock {
    3.     private static final Logger LOGGER = LoggerFactory.getLogger(CasLock.class);
    4.     // 0:没加锁；1：加了锁
    5.     private AtomicInteger state = new AtomicInteger(0);
    6.  
    7.     public void lock() {
    8.         while(true) {
    9.             if (state.compareAndSet(0, 1)) {
    10.                 break;
    11.             }
    12.         }
    13.     }
    14.  
    15.     public void unlock() {
    16.         LOGGER.debug("unlock...");
    17.         state.set(0);
    18.     }
    19.  
    20.     public static void main(String[] args) {
    21.         CasLock lock = new CasLock();
    22.         new Thread(()->{
    23.             LOGGER.debug("begin...");
    24.             lock.lock();
    25.             LOGGER.debug("lock...");
    26.             try {
    27.                 Thread.sleep(1000);
    28.             } catch (InterruptedException e) {
    29.                 e.printStackTrace();
    30.             } finally {
    31.                 lock.unlock();
    32.             }
    33.         }).start();
    34.  
    35.         new Thread(()->{
    36.             LOGGER.debug("begin...");
    37.             lock.lock();
    38.             try {
    39.                 LOGGER.debug("lock...");
    40.             } finally {
    41.                 lock.unlock();
    42.             }
    43.         }).start();
    44.     }
    45. }
    46.  
    47. 结果：
    48. 15:35:45.783 [Thread-0] DEBUG com.multiThreads.test16.CasLock - begin...
    49. 15:35:45.783 [Thread-1] DEBUG com.multiThreads.test16.CasLock - begin...
    50. 15:35:45.786 [Thread-0] DEBUG com.multiThreads.test16.CasLock - lock...
    51. 15:35:46.786 [Thread-0] DEBUG com.multiThreads.test16.CasLock - unlock...
    52. 15:35:46.786 [Thread-1] DEBUG com.multiThreads.test16.CasLock - lock...
    53. 15:35:46.786 [Thread-1] DEBUG com.multiThreads.test16.CasLock - unlock...

    7.1.2、原理之伪共享

    其中Cell即为累加单元。

    1.     @sum.misc.Contended
    2.     static final class Cell {
    3.         volatile long value;
    4.         
    5.         Cell(long x) {value = x;}
    6.         
    7.         // 最重要的方法，用来cas方式进行累加，prev表示旧值，next表示新值
    8.         final boolean cas(long prev, long next) {
    9.             return UNSAFE.compareAndSwapLong(this, valueOffset, prev, next);
    10.         }
    11.         
    12.         // 省略不重要代码
    13.         ...
    14.     }

    因为CPU与内存的速度差异很大，需要靠预读数据至缓存来提升效率。而缓存以缓存行为单位，每个缓存行对应着一块内存，一般是64 byte（8个long），缓存的加入会造成数据副本的产生，即同一份数据会缓存在不同核心的缓存行中，CPU要保证数据的一致性，如果某个CPU核心更改了数据，其他CPU核心对应的整个缓存行必须失效。

    因为Cell是数组形式，在内存中是连续存储的，一个Cell为24字节（16字节的对象头和8字节的value），因此缓存行可以存下2个Cell对象。这样问题来了：

    • 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字节的大小的padding，从而让CPU将对象预读至缓存时占用不同的缓存行，这样，不会造成对方缓存行的失效。

    8、Unsafe

    8.1、概述

    Unsafe对象提供了非常底层的，操作内存、线程的方法，Unsafe对象不能直接调用，只能通过反射获得。

    1. class UnsafeAccessor {
    2.     static Unsafe unsafe;
    3.     
    4.     static {
    5.         try {
    6.             Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
    7.             theUnsafe.setAccessible(true); // 允许访问私有成员
    8.             unsafe = (Unsafe) theUnsafe.get(null); // 获得Unsafe对象
    9.         } catch (NoSuchFieldException | IllegalAccessException e) {
    10.             e.printStackTrace();
    11.         }
    12.     }
    13.     
    14.     static Unsafe getUnsafe() {
    15.         return unsafe;
    16.     }
    17. }

    8.2、Unsafe CAS操作

    1. public class Test13 {
    2.     public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
    3.         Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
    4.         theUnsafe.setAccessible(true); // 允许访问私有成员
    5.         Unsafe unsafe = (Unsafe) theUnsafe.get(null); // 获得Unsafe对象
    6.  
    7.         // 1、获取域的偏移地址
    8.         long idOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("id"));
    9.         long nameOffset = unsafe.objectFieldOffset(Teacher.class.getDeclaredField("name"));
    10.         // 2、执行CAS操作
    11.         Teacher t = new Teacher();
    12.         unsafe.compareAndSwapInt(t, idOffset, 0, 1); // 修改id：0->1
    13.         unsafe.compareAndSwapObject(t, nameOffset, null, "张三"); // 修改name：null->张三
    14.         // 3、验证
    15.         System.out.println(t);
    16.     }
    17. }
    18.  
    19. @Data
    20. class Teacher {
    21.     volatile int id;
    22.     volatile String name;
    23. }
    24.  
    25. 结果：
    26. Teacher(id=1, name=张三)
    1. public class Test14 {
    2.     public static void main(String[] args) {
    3.         Account.demo(new MyAtomicInteger(10000));
    4.     }
    5. }
    6.  
    7. interface Account {
    8.     // 获取余额
    9.     Integer getBalance();
    10.  
    11.     // 取款
    12.     void withdraw(Integer amount);
    13.  
    14.     /**
    15.      * 方法内部会启动1000个线程，每个线程做 -10元 的操作
    16.      * 如果初始余额为10000那么正确的结果应当是0
    17.      */
    18.     static void demo(Account account) {
    19.         List threads = new ArrayList<>();
    20.         for (int i = 0; i < 1000; i++) {
    21.             threads.add(new Thread(()->{
    22.                 account.withdraw(10);
    23.             }));
    24.         }
    25.         long start = System.nanoTime();
    26.         threads.forEach(Thread::start);
    27.         threads.forEach(t -> {
    28.             try {
    29.                 t.join();
    30.             } catch (InterruptedException e) {
    31.                 e.printStackTrace();
    32.             }
    33.         });
    34.         long end = System.nanoTime();
    35.         System.out.println(account.getBalance() + " cost: " + (end - start)/1000_000 + "ms");
    36.     }
    37. }
    38.  
    39. class UnsafeAccessor {
    40.     static Unsafe unsafe;
    41.  
    42.     static {
    43.         try {
    44.             Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
    45.             theUnsafe.setAccessible(true); // 允许访问私有成员
    46.             unsafe = (Unsafe) theUnsafe.get(null); // 获得Unsafe对象
    47.         } catch (NoSuchFieldException | IllegalAccessException e) {
    48.             e.printStackTrace();
    49.         }
    50.     }
    51.  
    52.     static Unsafe getUnsafe() {
    53.         return unsafe;
    54.     }
    55. }
    56.  
    57. class MyAtomicInteger implements Account{
    58.     private static final Unsafe UNSAFE;
    59.     private static final long valueOffset;
    60.     private volatile int value;
    61.     static {
    62.         try {
    63.             UNSAFE = UnsafeAccessor.getUnsafe();
    64.             valueOffset = UNSAFE.objectFieldOffset(MyAtomicInteger.class.getDeclaredField("value"));
    65.         } catch (NoSuchFieldException e) {
    66.             e.printStackTrace();
    67.             throw new RuntimeException(e);
    68.         }
    69.     }
    70.  
    71.     public MyAtomicInteger(int value) {
    72.         this.value = value;
    73.     }
    74.  
    75.     public int getValue() {
    76.         return value;
    77.     }
    78.  
    79.     public void decrement(int amount) {
    80.         while(true) {
    81.             int prev = this.value;
    82.             int next = prev - amount;
    83.             if (UNSAFE.compareAndSwapInt(this, valueOffset, prev, next)) {
    84.                 break;
    85.             }
    86.         }
    87.     }
    88.  
    89.     @Override
    90.     public Integer getBalance() {
    91.         return getValue();
    92.     }
    93.  
    94.     @Override
    95.     public void withdraw(Integer amount) {
    96.         decrement(amount);
    97.     }
    98. }
    99.  
    100. 结果：
    101. 0 cost: 111ms

  • 相关阅读:
    复数go版本的下载和管理
    [机缘参悟-119] ：反者道之动与阴阳太极
    1024发博客！
    JAVA的由来：要不要干一杯95年的Java
    STM32实现USB转TTL串口工具
    MySQL学习笔记（十二）锁
    23062C++&QTday5
    【openstack】卷迁移导致的虚机热迁移失败问题排查及解决
    OSPF的防止环路的机制
    Web基础与HTTP协议
  • 原文地址：https://blog.csdn.net/weixin_44623055/article/details/126235236