超越并发瓶颈：CAS与乐观锁的智慧应用

目录

一、走进CAS

（一）基本知识快速回顾

（二）CAS 的原子性实际上是硬件 CPU 直接保证的

Java 层次

JVM 层次

硬件层次

总结

（三）性能检测说明

二、理解乐观锁

（一）直观的理解

（二）乐观锁适用于读多写少的情况的原因

（三）悲观锁在读多写少的情况下也有冲突少的特点，为什么不适合呢？

（四）案例：乐观锁实现余额更新

三、整体理解

（一）CAS机制和乐观锁思想

（二）深入源码扩展

​​​​​​​干货分享，感谢您的阅读！​​​​​​​

synchronized 是 Java 内置的同步机制，依赖 JVM 实现，通过进入和退出监视器锁（Monitor Lock）来保证线程的安全性。在高并发情况下，线程可能会频繁地在 BLOCKED 状态和 RUNNABLE 状态之间切换，导致用户态和内核态的频繁切换，从而影响性能。

而Lock（如 ReentrantLock）是基于 AQS 实现，通过使用自旋锁和非阻塞算法，减少了用户态和内核态的切换，提高了性能。

​与 synchronized 的实现方式不同，AQS中很多数据结构的变化，都是依赖 CAS 进行操作的，而CAS 就是乐观锁的一种实现。

一、走进CAS

这里我们直接简单理解即可，如果想深入可见：揭秘CAS：深入理解与应用解析-CSDN博客。

（一）基本知识快速回顾

CAS（Compare And Swap）技术是一种常用的并发编程技术，用于解决多线程环境下的数据竞争问题。它是无锁（lock-free）算法的一种实现方式，通过原子操作实现了线程安全的数据更新。

CAS操作通常包括三个参数：内存地址（或变量），期望值和新值。

​它的执行过程如下：

  • 比较：CAS操作首先比较内存地址中的值与期望值是否相等。
  • 交换：如果相等，CAS操作将新值写入内存地址，否则不进行任何操作。
  • 返回：CAS操作返回内存地址中的旧值。

CAS操作在执行过程中具有原子性，即在比较和交换的过程中不会被其他线程干扰。这样可以确保多线程环境下数据的一致性和正确性。

分类具体说明解释
优点无锁特性CAS操作不需要使用传统的锁机制，避免了锁带来的开销和竞争问题，提高了并发性能和可伸缩性。
原子性CAS操作是原子的，不会发生线程间的竞争和数据不一致的问题。
适用范围广CAS操作可以用于实现各种同步原语和数据结构，如原子操作、乐观锁、无锁队列等。
缺点ABA问题CAS操作无法检测到其他线程在执行过程中对值的修改。例如，一个值原先是A，然后被修改为B，最后又被修改回A，这时CAS操作无法区分这种情况，会错误地认为值没有被修改过。
自旋开销如果CAS操作失败，会进入自旋状态，不断重试直到成功。自旋会占用CPU资源，当自旋次数过多时，会导致性能下降。
并发度限制CAS操作在并发度较高的情况下，由于竞争激烈，可能会导致大量的CAS操作失败，进而影响性能。

在实际应用中，CAS技术常用于实现线程安全的数据结构和算法，例如并发队列、计数器、自旋锁等。它在一些高并发场景下能够提供较好的性能和可伸缩性，但需要注意处理ABA问题和合理控制自旋次数，以及根据具体场景评估并发度的限制。

（二）CAS 的原子性实际上是硬件 CPU 直接保证的

直接以AtomicInteger 中 CAS 操作的原子性保证来进行理解。

Java 层次

AtomicInteger 类中的 compareAndSet 方法用于执行 CAS 操作，其代码如下：

  1. public final boolean compareAndSet(int expectedValue, int newValue) {
  2.     return U.compareAndSetInt(this, VALUE, expectedValue, newValue);
  3. }

这里的 U 是 Unsafe 类的实例，VALUE 是内存偏移量。compareAndSetInt 是 Unsafe 类中的一个本地方法，直接调用底层的硬件指令来实现原子操作。

public final native boolean compareAndSetInt(Object o, long offset, int expected, int x);

JVM 层次

Unsafe 类中 compareAndSetInt 方法的实现会调用 weakCompareAndSetInt 方法，该方法通过自旋重试实现CAS操作：

  1. public final int getAndAddInt(Object o, long offset, int delta) {
  2.     int v;
  3.     do {
  4.         v = getIntVolatile(o, offset);
  5.     } while (!weakCompareAndSetInt(o, offset, v, v + delta));
  6.     return v;
  7. }

在这个方法中，getIntVolatile 获取当前值，weakCompareAndSetInt 尝试更新值，如果更新失败，则重复上述过程，直到成功，即自旋重试。

硬件层次

在 Linux 系统的 x86 架构上，CAS 操作最终会映射到 cmpxchgl 汇编指令，这是由 os_cpu/linux_x86/atomic_linux_x86.hpp 文件中的代码实现的：

  1. template<>
  2. template<typename T>
  3. inline T Atomic::PlatformCmpxchg<4>::operator()(T exchange_value,
  4.                                                 T volatile* dest,
  5.                                                 T compare_value,
  6.                                                 atomic_memory_order /* order */) const {
  7.     STATIC_ASSERT(4 == sizeof(T));
  8.     __asm__ volatile ("lock cmpxchgl %1,(%3)"
  9.                       : "=a" (exchange_value)
  10.                       : "r" (exchange_value), "a" (compare_value), "r" (dest)
  11.                       : "cc", "memory");
  12.     return exchange_value;
  13. }

这里的 cmpxchgl 指令是关键。这条汇编指令的作用是：

  • 比较寄存器 EAX 中的值（compare_value）和内存地址 dest 中的值。
  • 如果两者相等，则将 exchange_value 存储到 dest 中。
  • 如果不相等，则将 dest 中的值加载到 EAX 中。

lock 前缀确保了操作的原子性，这意味着在多处理器系统中，该指令在执行时会锁住总线或使用缓存一致性协议，保证其他处理器不能访问内存地址，直到操作完成。

在不同的硬件平台上，支持CAS操作的指令可能不同，但其基本原理是一致的：

  • x86 平台：x86处理器提供了 CMPXCHG 指令来实现CAS操作。这个指令是原子的，即在执行过程中，不会被其他指令中断。
  • PowerPC 平台：PowerPC处理器提供了 lwarx 和 stwcx. 指令组合来实现CAS操作，这些指令也确保了操作的原子性。
  • ARM 平台：ARM处理器提供了 LDREX 和 STREX 指令组合来实现CAS操作。

总结

硬件指令 cmpxchgl 结合 lock 前缀保证了在多处理器环境下的原子性，即整个比较和替换操作是不可分割的，这就是 CAS 操作能够实现原子性的原因。

（三）性能检测说明

为了验证原子类的性能优势，可以编写一个简单的测试程序，分别使用 AtomicInteger（CAS 实现）和 synchronized 关键字来实现一个计数器，并进行多线程并发访问，最后比较它们的性能。具体代码如下：

  1. package org.zyf.javabasic.thread.lock.opti;
  2.  
  3. import java.util.concurrent.atomic.AtomicInteger;
  4.  
  5. /**
  6.  * @program: zyfboot-javabasic
  7.  * @description: 比较使用 AtomicInteger 和 synchronized 关键字的性能
  8.  * @author: zhangyanfeng
  9.  * @create: 2024-06-08 11:10
  10.  **/
  11. public class AtomicVsSynchronized {
  12.     private static final int THREAD_COUNT = 100;
  13.     private static final int ITERATIONS = 1000000;
  14.  
  15.     // 使用 AtomicInteger 实现计数器
  16.     private static AtomicInteger atomicCounter = new AtomicInteger(0);
  17.  
  18.     // 使用 synchronized 实现计数器
  19.     private static int synchronizedCounter = 0;
  20.  
  21.     // 使用 AtomicInteger 进行计数
  22.     private static class AtomicCounterRunnable implements Runnable {
  23.         @Override
  24.         public void run() {
  25.             for (int i = 0; i < ITERATIONS; i++) {
  26.                 atomicCounter.incrementAndGet();
  27.             }
  28.         }
  29.     }
  30.  
  31.     // 使用 synchronized 进行计数
  32.     private static class SynchronizedCounterRunnable implements Runnable {
  33.         @Override
  34.         public void run() {
  35.             for (int i = 0; i < ITERATIONS; i++) {
  36.                 synchronized (AtomicVsSynchronized.class) {
  37.                     synchronizedCounter++;
  38.                 }
  39.             }
  40.         }
  41.     }
  42.  
  43.     public static void main(String[] args) throws InterruptedException {
  44.         long startTime;
  45.         long endTime;
  46.  
  47.         // 测试使用 AtomicInteger 的性能
  48.         startTime = System.currentTimeMillis();
  49.         Thread[] atomicThreads = new Thread[THREAD_COUNT];
  50.         for (int i = 0; i < THREAD_COUNT; i++) {
  51.             atomicThreads[i] = new Thread(new AtomicCounterRunnable());
  52.             atomicThreads[i].start();
  53.         }
  54.         for (int i = 0; i < THREAD_COUNT; i++) {
  55.             atomicThreads[i].join();
  56.         }
  57.         endTime = System.currentTimeMillis();
  58.         System.out.println("AtomicInteger 总耗时：" + (endTime - startTime) + " 毫秒");
  59.         System.out.println("AtomicInteger 计数结果：" + atomicCounter.get());
  60.  
  61.         // 测试使用 synchronized 的性能
  62.         startTime = System.currentTimeMillis();
  63.         Thread[] syncThreads = new Thread[THREAD_COUNT];
  64.         for (int i = 0; i < THREAD_COUNT; i++) {
  65.             syncThreads[i] = new Thread(new SynchronizedCounterRunnable());
  66.             syncThreads[i].start();
  67.         }
  68.         for (int i = 0; i < THREAD_COUNT; i++) {
  69.             syncThreads[i].join();
  70.         }
  71.         endTime = System.currentTimeMillis();
  72.         System.out.println("synchronized 总耗时：" + (endTime - startTime) + " 毫秒");
  73.         System.out.println("synchronized 计数结果：" + synchronizedCounter);
  74.     }
  75. }

运行结果如下：

  • AtomicInteger 总耗时：6754 毫秒 ，AtomicInteger 计数结果：100000000
  • synchronized 总耗时：2409 毫秒，synchronized 计数结果：100000000

数据说明使用 synchronized 关键字的方式比使用 AtomicInteger 类的方式具有更好的性能。

二、理解乐观锁

（一）直观的理解

CAS（Compare-And-Swap）可以被看作是乐观锁的一种实现方式。

乐观锁认为自己在使用数据时不会有别的线程修改数据，所以不会添加锁，只是在更新数据的时候去判断之前有没有别的线程更新了这个数据。如果这个数据没有被更新，当前线程将自己修改的数据成功写入。如果数据已经被其他线程更新，则根据不同的实现方式执行不同的操作（例如报错或者自动重试）

在Java中是通过使用无锁编程来实现，最常采用的是CAS算法，Java原子类中的递增操作就通过CAS自旋实现的。（ 可见Java中常用的锁总结与理解_java锁用来解决什么-CSDN博客）

（二）乐观锁适用于读多写少的情况的原因

  • 乐观锁不阻塞读操作：乐观锁在读取数据时并不会进行加锁操作，而是先读取数据，然后在更新数据时检查是否被其他线程修改过。因此，即使有写操作正在进行，读操作也不会被阻塞，从而可以实现读操作的并发执行。

  • 乐观锁的冲突少：在读多写少的情况下，写操作的频率较低，因此冲突的概率也相对较低。乐观锁的重试操作是在发生冲突时进行的，因此在冲突较少的情况下，重试的概率也较低，从而可以更高效地处理并发冲突。

  • 乐观锁的开销低：乐观锁的实现通常比较轻量级，不需要频繁地进行加锁和解锁操作，因此在读多写少的情况下，乐观锁的性能通常会更好。

（三）悲观锁在读多写少的情况下也有冲突少的特点，为什么不适合呢？

尽管悲观锁在读多写少的情况下可能会有较少的冲突，但它的主要问题在于加锁这个动作上：

  • 悲观锁在读取数据时通常会对共享资源进行加锁，这会导致其他线程无法同时进行读操作。即使读操作之间并没有冲突，也会由于加锁操作而导致不必要的阻塞。
  • 悲观锁在进行加锁和解锁操作时会引入额外的开销，尤其是在并发量较大的情况下，频繁的加锁和解锁操作会降低系统的性能。

尽管悲观锁在一些情况下也能够处理并发问题，但在读多写少的情况下，乐观锁更适合，因为它更符合读多写少的特点，可以更好地实现读操作的并发执行，提高系统的性能。

（四）案例：乐观锁实现余额更新

假设我们有一个账户类 Account，其中包含一个余额字段 balance，我们希望使用乐观锁实现对余额的更新操作。下面是一个简单的示例代码：

  1. package org.zyf.javabasic.thread.lock.opti;
  2.  
  3. import java.util.concurrent.atomic.AtomicReference;
  4.  
  5. /**
  6.  * @program: zyfboot-javabasic
  7.  * @description: 账户类 Account
  8.  * @author: zhangyanfeng
  9.  * @create: 2024-06-08 12:33
  10.  **/
  11. public class Account {
  12.     private AtomicReference balance;
  13.  
  14.     public Account(double initialBalance) {
  15.         this.balance = new AtomicReference<>(initialBalance);
  16.     }
  17.  
  18.     public void updateBalance(double amount) {
  19.         boolean success = false;
  20.         do {
  21.             Double currentBalance = balance.get();
  22.             Double newBalance = currentBalance + amount;
  23.             success = balance.compareAndSet(currentBalance, newBalance);
  24.         } while (!success);
  25.     }
  26.  
  27.     public double getBalance() {
  28.         return balance.get();
  29.     }
  30. }

比较使用乐观锁和悲观锁（使用 synchronized 关键字）两种方式实现余额更新的性能表现：

  1. package org.zyf.javabasic.thread.lock.opti;
  2.  
  3. /**
  4.  * @program: zyfboot-javabasic
  5.  * @description: 乐观锁和悲观锁
  6.  * @author: zhangyanfeng
  7.  * @create: 2024-06-08 12:34
  8.  **/
  9. public class AccountPerformanceTest {
  10.     private static final int THREAD_COUNT = 100;
  11.     private static final int ITERATIONS = 1000000;
  12.  
  13.     public static void main(String[] args) throws InterruptedException {
  14.         Account optimisticAccount = new Account(1000);
  15.         Account pessimisticAccount = new Account(1000);
  16.  
  17.         long startTime;
  18.         long endTime;
  19.  
  20.         // 测试乐观锁性能
  21.         startTime = System.currentTimeMillis();
  22.         Thread[] optimisticThreads = new Thread[THREAD_COUNT];
  23.         for (int i = 0; i < THREAD_COUNT; i++) {
  24.             Account account = new Account(1000); // 创建新的账户对象
  25.             optimisticThreads[i] = new Thread(() -> {
  26.                 for (int j = 0; j < ITERATIONS; j++) {
  27.                     account.updateBalance(10);
  28.                 }
  29.             });
  30.             optimisticThreads[i].start();
  31.         }
  32.         for (int i = 0; i < THREAD_COUNT; i++) {
  33.             optimisticThreads[i].join();
  34.         }
  35.         endTime = System.currentTimeMillis();
  36.         System.out.println("乐观锁总耗时：" + (endTime - startTime) + " 毫秒");
  37.         System.out.println("乐观锁最终余额：" + optimisticAccount.getBalance());
  38.  
  39.         // 测试悲观锁性能
  40.         startTime = System.currentTimeMillis();
  41.         Thread[] pessimisticThreads = new Thread[THREAD_COUNT];
  42.         for (int i = 0; i < THREAD_COUNT; i++) {
  43.             Account account = new Account(1000); // 创建新的账户对象
  44.             pessimisticThreads[i] = new Thread(() -> {
  45.                 for (int j = 0; j < ITERATIONS; j++) {
  46.                     synchronized (account) {
  47.                         account.updateBalance(10);
  48.                     }
  49.                 }
  50.             });
  51.             pessimisticThreads[i].start();
  52.         }
  53.         for (int i = 0; i < THREAD_COUNT; i++) {
  54.             pessimisticThreads[i].join();
  55.         }
  56.         endTime = System.currentTimeMillis();
  57.         System.out.println("悲观锁总耗时：" + (endTime - startTime) + " 毫秒");
  58.         System.out.println("悲观锁最终余额：" + pessimisticAccount.getBalance());
  59.     }
  60. }

运行结果如下：

  • 乐观锁总耗时：1293 毫秒，乐观锁最终余额：1000.0
  • 悲观锁总耗时：2954 毫秒，悲观锁最终余额：1000.0

根据测试结果，乐观锁的总耗时明显比悲观锁少，而且最终余额也保持了正确的值。这与乐观锁的特性相符合，因为乐观锁在读多写少的场景下性能通常会更好，而且不会导致线程阻塞，从而提高了并发性能。

三、整体理解

（一）CAS机制和乐观锁思想

CAS（Compare-And-Swap）是一种基于硬件原子操作的乐观锁实现机制。它适用于读多写少的场景，在并发操作中，线程先读取共享资源的值，然后尝试进行更新操作。

CAS操作分为比较和替换两个阶段，如果在替换阶段发现共享资源的值与预期值相同，则更新操作成功，否则需要重新尝试。CAS机制利用硬件提供的原子操作确保了比较和替换这两个步骤的原子性，避免了锁带来的性能开销和线程阻塞。

乐观锁的设计思想是“先写后检查”，相信操作会顺利进行，只有在发生冲突时才进行处理。CAS机制是乐观锁的一种实现方式，它通过乐观地假设没有其他线程在修改共享资源的情况下进行操作，避免了不必要的阻塞和等待，提高了系统的并发性能。

（二）深入源码扩展

结合了CAS机制和乐观锁的思想来体会源码的话最直接的就是Java的并发包（java.util.concurrent），感兴趣可直接扩展以下基本部分：

  1. AtomicInteger/AtomicLong/AtomicReference等原子类：利用CAS机制实现了一系列原子操作，如增加、减少、更新等，提供了一种无锁的方式进行原子操作，避免了使用传统的锁机制带来的性能开销。

  2. StampedLock类：StampedLock是Java 8中新增的类，结合了乐观锁和悲观锁的思想。它提供了三种模式：读模式、写模式和乐观读模式。在读多写少的场景下，StampedLock使用乐观锁来进行读操作，只有在乐观读失败时才使用悲观锁来升级为读锁。

  3. ConcurrentHashMap类：ConcurrentHashMap是一个高效的线程安全的哈希表实现，它采用了分段锁的方式来实现并发访问。在Java 8及以上版本中，ConcurrentHashMap利用了CAS操作来进行无锁的并发更新操作，从而提高了性能。

以上如果感兴趣的请自行扩展阅读。

文章知识点与官方知识档案匹配，可进一步学习相关知识
Java技能树并发并发的定义148946 人正在系统学习中
注：本文转载自blog.csdn.net的张彦峰ZYF的文章"https://blog.csdn.net/xiaofeng10330111/article/details/124414624"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！