Java并发编程深度解析：掌握CAS、Synchronized与Callable的高效并发之道

Java并发编程

本篇文章主要带大家深入了解Java并发编程，了解CAS，Synchronized原理以及Callable接口的概念及使用。

1.CAS

CAS: 全称 Compare and swap ，字面意思 :” 比较并交换 “ ，一个 CAS 涉及到以下操作：

我们假设内存中的原数据 V ，旧的预期值 A ，需要修改的新值 B 。 1. 比较 A 与 V 是否相等。（比较） 2. 如果比较相等，将 B 写入 V 。（交换） 3. 返回操作是否成功。

CAS是一个具有原子性的硬件指令集，当多个线程同时对某个资源进行CAS操作，只能有一个线程操作成功，但是并不会阻塞其他线程,其他线程只会收到操作失败的信号。

针对不同的操作系统，JVM用到了不同的原理在Java中也实现了CAS。

CAS的具体应用

实现原子类

标准库中提供了 java.util.concurrent.atomic 包 , 里面的类都是基于这种方式来实现的 . 典型的就是 AtomicInteger 类 . 其中的 getAndIncrement 相当于 i++ 操作 .

AtomicInteger atomicInteger = new AtomicInteger(0);
// 相当于 i++
atomicInteger.getAndIncrement();

假设两个线程同时调用 getAndIncrement。有以下几个步骤：

1. 两个线程都读取 value 的值到 oldValue 中. (oldValue 是一个局部变量, 在栈上. 每个线程有自己的栈)
2. 线程1 先执行 CAS 操作. 由于 oldValue 和 value 的值相同, 直接进行对 value 赋值.
3. 线程2 再执行 CAS 操作, 第一次 CAS 的时候发现 oldValue 和 value 不相等, 不能进行赋值. 因此需要进入循环. 在循环里重新读取 value 的值赋给 oldValue.
4. 线程2 接下来第二次执行 CAS, 此时 oldValue 和 value 相同, 于是直接执行赋值操作.
5. 线程1 和 线程2 返回各自的 oldValue 的值即可.

通过形如上述代码就可以实现一个原子类. 不需要使用重量级锁, 就可以高效的完成多线程的自增操作.

实现自旋锁

基于 CAS 实现更灵活的锁 , 获取到更多的控制权 .

自旋锁伪代码：

public class SpinLock {
    private Thread owner = null;
    public void lock(){
        // 通过 CAS 看当前锁是否被某个线程持有. 
        // 如果这个锁已经被别的线程持有, 那么就自旋等待. 
        // 如果这个锁没有被别的线程持有, 那么就把 owner 设为当前尝试加锁的线程. 
        while(!CAS(this.owner, null, Thread.currentThread())){
       }
   }
    public void unlock (){
        this.owner = null;
   }
}

CAS的ABA问题

什么是ABA问题

ABA 的问题 :

假设存在两个线程 t1 和 t2. 有一个共享变量 num, 初始值为 A.

接下来 , 线程 t1 想使用 CAS 把 num 值改成 Z, 那么就需要：

• 先读取 num 的值, 记录到 oldNum 变量中.
• 使用 CAS 判定当前 num 的值是否为 A, 如果为 A, 就修改成 Z.

但是 , 在 t1 执行这两个操作之间 , t2 线程可能把 num 的值从 A 改成了 B, 又从 B 改成了 A. t1 线程无法区分当前这个变量始终是 A, 还是经历了一个变化过程 .如下图所示：

ABA问题引起的bug以及解决方案

上述的ABA问题就有可能在一些对业务准确性要求较高的领域引起难以接受的bug。

比如：笔友有100元存款，滑稽想从 ATM 取 50 块钱. 取款机创建了两个线程, 并发的来执行 -50操作. 我们期望一个线程执行 -50 成功, 另一个线程 -50 失败. 如果使用 CAS 的方式来完成这个扣款过程就可能出现问题. 正常的过程 1) 存款 100. 线程1 获取到当前存款值为 100, 期望更新为 50; 线程2 获取到当前存款值为 100, 期望更新为 50. 2) 线程1 执行扣款成功, 存款被改成 50. 线程2 阻塞等待中. 3) 轮到线程2 执行了, 发现当前存款为 50, 和之前读到的 100 不相同, 执行失败. 异常的过程 1) 存款 100. 线程1 获取到当前存款值为 100, 期望更新为 50; 线程2 获取到当前存款值为 100, 期望更新为 50. 2) 线程1 执行扣款成功, 存款被改成 50. 线程2 阻塞等待中. 3) 在线程2 执行之前, 滑稽的朋友正好给滑稽转账 50, 账户余额变成 100 !! 4) 轮到线程2 执行了, 发现当前存款为 100, 和之前读到的 100 相同, 再次执行扣款操作. 这个时候, 扣款操作被执行了两次!!!

解决方案：

给要修改的值 , 引入版本号 . 在 CAS 比较数据当前值和旧值的同时 , 也要比较版本号是否符合预期 .

CAS 操作在读取旧值的同时 , 也要读取版本号，真正修改的时候：

• 如果当前版本号和读到的版本号相同, 则修改数据, 并把版本号 + 1.
• 如果当前版本号高于读到的版本号. 就操作失败(认为数据已经被修改过了).

2.Synchronized原理

上篇文章中，我分享了锁策略，里面提到了synchronized具有的特性（在JDK1.8中），如下：

1. 开始时是乐观锁 , 如果锁冲突频繁 , 就转换为悲观锁 .

2. 开始是轻量级锁实现 , 如果锁被持有的时间较长 , 就转换成重量级锁 .

3. 实现轻量级锁的时候大概率用到的自旋锁策略

4. 是一种不公平锁

5. 是一种可重入锁

6. 不是读写锁

JVM在运行synchronized时采用了非常多的优化操作。JVM 将 synchronized 锁分为 无锁、偏向锁、轻量级锁、重量级锁 状态。会根据情况，进行依升级。

什么是偏向锁？

偏向锁不是真的 "加锁", 只是给对象头中做一个 "偏向锁的标记", 记录这个锁属于哪个线程. 如果后续没有其他线程来竞争该锁, 那么就不用进行其他同步操作了(避免了加锁解锁的开销)如果后续有其他线程来竞争该锁(刚才已经在锁对象中记录了当前锁属于哪个线程了, 很容易识别当前申请锁的线程是不是之前记录的线程), 那就取消原来的偏向锁状态, 进入一般的轻量级锁状态。偏向锁本质上相当于 "延迟加锁" 。能不加锁就不加锁, 尽量来避免不必要的加锁开销。但是该做的标记还是得做的, 否则无法区分何时需要真正加锁。

什么是轻量级锁？

随着其他线程进入竞争 , 偏向锁状态被消除 , 进入轻量级锁状态 ( 自适应的自旋锁 ). 此处的轻量级锁就是通过 CAS 来实现：

• 通过 CAS 检查并更新一块内存 (比如 null => 该线程引用)
• 如果更新成功, 则认为加锁成功
• 如果更新失败, 则认为锁被占用, 继续自旋式的等待(并不放弃 CPU).

什么是重量级锁？

如果竞争进一步激烈 , 自旋不能快速获取到锁状态 , 就会膨胀为重量级锁 此处的重量级锁就是指用到内核提供的 mutex。

• 执行加锁操作, 先进入内核态.
• 在内核态判定当前锁是否已经被占用
• 如果该锁没有占用, 则加锁成功, 并切换回用户态.
• 如果该锁被占用, 则加锁失败. 此时线程进入锁的等待队列, 挂起. 等待被操作系统唤醒.
• 经历了一系列的沧海桑田, 这个锁被其他线程释放了, 操作系统也想起了这个挂起的线程, 于是唤醒这个线程, 尝试重新获取锁.

锁消除

编译器 +JVM会 判断锁是否可消除 . 如果可以 , 就直接消除。

有些应用程序的代码中, 用到了 synchronized, 但其实没有在多线程环境下. (例如 StringBuffer)

此时每个 append 的调用都会涉及加锁和解锁. 但如果只是在单线程中执行这个代码, 那么这些加锁解锁操作是没有必要的, 白白浪费了一些资源开销.

锁粗化

一段逻辑中如果出现多次加锁解锁, 编译器 + JVM 会自动进行锁的粗化。如下图所示：

3.Callable接口

Callable 是一个 interface. 相当于把线程封装了一个 " 返回值 ". 方便程序猿借助多线程的方式计算结果 .

例如：我现在要创建一个线程计算1 + 2 + 3 + ... + 1000。

不使用Callable接口，代码如下：

static class Result {
    public int sum = 0;
    public Object lock = new Object();
}
public static void main(String[] args) throws InterruptedException {
    Result result = new Result();
    Thread t = new Thread() {
        @Override
        public void run() {
            int sum = 0;
            for (int i = 1; i <= 1000; i++) {
                sum += i;
           }
            synchronized (result.lock) {
                result.sum = sum;
                result.lock.notify();
           }
       }
   };
    t.start();
    synchronized (result.lock) {
        while (result.sum == 0) {
            result.lock.wait();
       }
        System.out.println(result.sum);
   }
}

可以看到, 上述代码需要一个辅助类 Result, 还需要使用一系列的加锁和 wait notify 操作, 代码复杂, 容易出错.

下面使用Callable来解决这个问题：

Callable<Integer> callable = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        int sum = 0;
        for (int i = 1; i <= 1000; i++) {
            sum += i;
       }
        return sum;
   }
};
FutureTask<Integer> futureTask = new FutureTask<>(callable);
Thread t = new Thread(futureTask);
t.start();
int result = futureTask.get();
System.out.println(result);

可以看到, 使用 Callable 和 FutureTask 之后, 代码简化了很多, 也不必手动写线程同步代码了.

Callable 和 Runnable 相对 , 都是描述一个 " 任务 ". Callable 描述的是带有返回值的任务 , Runnable 描述的是不带返回值的任务 . Callable 通常需要搭配 FutureTask 来使用 . FutureTask 用来保存 Callable 的返回结果 . 因为Callable 往往是在另一个线程中执行的 , 啥时候执行完并不确定 . FutureTask 就可以负责这个等待结果出来的工作 .