【程序员的必修课】并发编程--原子性保证_并发读写操作的原子性保证

并发编程-原子性保证

一、前言

关于并发编程的三大问题：原子性、有序性、可见性，我们在第一次的分享中就提到过了

今天，我们就来聊聊原子性相关的问题

这里关于显示锁 synchronized，我就不再赘述了，感兴趣的同学，可以翻阅我之前写的介绍 synchronized 的文章哦：https://mp.weixin.qq.com/s/L7VZ1ELlWG5tHZ6hMgbGSg

那这篇文章我们还能讲什么东西呢？之前我也提过了，原子性问题，我们要关注的不是原子性如何实现，而是怎么确保锁的效率和避免死锁

二、限定锁的范围

虽然我们前面提到，我们需要注意的问题，是确保锁的效率和避免死锁，但是我还是想提一下锁范围这个问题

在讲这个问题之前，我想先引出我会在全篇使用的一个例子，那就是银行转账

public class Account {
    private long balance;
    private String bankID;

    public Account(String bankID) {
        this.bankID = bankID;
    }

    // 转账方法
  	// 注：目前这个状态是不具备一点线程安全性的
    public boolean transfer(String targetID,long amount) {
	      if (this.balance<amount) return false;
        this.balance-=amount;
        // 将要为被转用户设置的新 balance 数
        long balanceToBeSet = target.getBalance() + amount;
        // 在转账和被转账账户之间，设置睡眠，模拟各大银行之间的转账延迟
        try {
            Thread.sleep(delay);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        target.setBalance(balanceToBeSet);
        return true;
    }
  
  	// getter and setter
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Account 类包括两个成员变量，一个是表示用户的 bankID，一个表示用户当前拥有的钱数 balance

这里我们有一个需求，就是希望账户A 转钱给账户B，账户B 转钱给 账户C

测试用例如下：

package top.faroz.back_transfer;

import java.util.HashMap;

public class TransferDemo {
    public static void main(String[] args) throws InterruptedException {
        Account A = new Account("A");
        Account B = new Account("B");
        Account C = new Account("C");
        A.setBalance(200);
        B.setBalance(200);
        C.setBalance(200);
        TransferThread t1 = new TransferThread(A, B, 100,1000);
        TransferThread t2 = new TransferThread(B, C, 100,100);
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println("A->B，B->C 的转账操作执行完毕");
        System.out.println("A 剩余："+A.getBalance());
        System.out.println("B 剩余："+B.getBalance());
        System.out.println("C 剩余："+C.getBalance());
    }
}

class TransferThread extends Thread {
    private Account account1;
    private Account account2;
    private long amt;
    private int delay;

    public TransferThread(Account account1, Account account2,long amt,int delay) {
        this.account1 = account1;
        this.account2 = account2;
        this.amt = amt;
        this.delay = delay;
    }

    @Override
    public void run() {
        account1.transfer(account2,amt,delay);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

最后的预期结果，是 A剩100，B剩200，C剩300

我们先来执行以下我们的测试用例：

可以发现，B 账户本应该是200余额，变成了 300，说明这个银行系统中，莫名其妙多出了 100 元，这是在金融系统中，绝对不允许出现的情况，如果出现了，写这段逻辑的程序员可以被拉去祭天了

我们先来分析一下，为什么我们上面这段（没有加锁的）代码，会出现这个问题

• 首先，在转账逻辑中，我在更新被转用户之前，先记录了被转用户更新后的钱数

• 随后，我们又模拟了转账延迟

并且，我还故意将 A->B 的延迟数设置的比 B->C 的延迟时间长，保证 B先把钱扣了（B这时理应是剩余100元）

但是因为我们在 t1 线程提前存储了B被转后应该修改的金额（300），导致 t2 线程修改完后的 B 的值（100），被直接覆盖成了 300

这，就是不加锁导致的线程安全问题，再细化一点，这里的问题，是线程安全中的可见性问题

1、无脑方法锁

这个时候有同学可能会说了：“直接在 transfor 方法前加上 synchronized 不就好了？”

有这种想法的同学，一般是刚学并发编程的小白，咱们可以来看看，只对 transfor 方法加 synchronized ，能不能解决问题

可以发现，运行结果仍然是错误的

咱们分析一下错误原因，这里可不是因为 synchronized 失效了，其实是因为咱们锁住的对象不完整

我们转账，是要对两个对象加锁的，一个账户 A，一个账户 B

这里对非静态方法加上方法锁，其实只是对其中一个对象加了锁（对象A），对象 B 仍然是我行我素，一样会在 t2 线程没执行完的时候，被 t1 线程修改，达不到我们保证线程安全的目的

2、锁公有锁

我们可以改造一下 Account 类，添加一个公有锁对象，synchronized 加锁的时候，锁这个对象就行

代码我就不写了，其原理和我下面要介绍的方法三很类似

3、锁类对象

锁类对象，就是锁 Account.class

一般要锁类对象，我们有两个方法：

**1、**将 transfor 改造成静态方法，加 synchroinized

但是这样，成员变量也要修改成 static ，与业务不符，这里我们不采用

**2、**transfor 内部 synchronized 锁直接锁 Account.class

因为在 java 中，一个类只有一个类对象，所以只要是调用了这个方法的对象，其操作的锁，都是同一个锁

其执行结果，也是正确的了

三、提高锁的效率

上面的代码因为有对象锁的加持，已经安全无比了，西卡西，又出现了一个新的问题------我们的代码变成了串行化的了

说得更直白一点，B->C 的转账过程，必须要等 A->B 的转账过程结束了，才可以执行

就别说两个有共同对象的转账操作了，就算是两个操作分别是 A->B ，C->D，后面一个也要等前面一个完成，才能执行，这种等待是完全没有必要的，因为我们知道，两笔转账操作没有一点数据交集，等待纯粹就是浪费 CPU 资源

假设是支付宝级别的支付量，日平均支付单数在 8亿，平均下来，每秒就有将近 1W 比交易

但是我们串行化的交易效率，是完全支撑不了的

所以所，要优化

1、改用细粒度锁

这里，我们常见的优化方式是这样的，既然对象锁的粒度太大，我们是不是要想办法将锁粒度缩小？

这里，缩的最小的范围，就是我们要操作的两个对象了

这样，思路就来了，方法中每次要进行转账操作，我们必须要获取转账和被转账的对象锁，才能执行

public boolean transfer(Account target,long amount,int delay) {
        synchronized (this) {
            synchronized (target) {
                if (this.balance<amount) return false;
                this.balance-=amount;
                // 将要为被转用户设置的新 balance 数
                long balanceToBeSet = target.getBalance() + amount;
                // 在转账和被转账账户之间，设置睡眠，模拟各大银行之间的转账延迟
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                target.setBalance(balanceToBeSet);
                return true;
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

对 synchronized 原理不太了解的同学，可能对这种做法有什么好处还不太了解，这里我再稍微说明一下

前门我们所 class 对象的方法，是让这个方法，在同一时间段内，只能对两个对象进行操作，再说的直白一点，就是在 A->B 转账的过程中， C->D 的转账操作是无法完成的

但是改用我们只锁两个对象的细粒度锁，A->B 转账时，因为 C->D 转账不存在数据的冲突，所以照样可以同时执行，这样，其效率就大大增加了

再来看一下运行结果：

这个方法看似天衣无缝了，但其实，隐藏着巨大的隐患…

四、避免死锁

1、死锁复现

第三小结我们提到，细粒度锁看似天衣无缝，但是隐藏着巨大的隐患

具体什么隐患？就是我们的死锁问题了

在操作系统中，出现死锁，还有鸵鸟策略，随机杀死策略等方式处理

但是在后端程序中，一旦某个方法出现死锁，如果没有超时释放机制的话，那么这两个线程就会一直卡死，只有重启应用或者服务器才能解决，这，是在生产中绝对不允许发生的，写出这样的代码，程序员就离被拉去祭天不远了

我们这就来复现一下上面代码的死锁情况，不过这里我们的业务要改成 A->B 转账，B->A 转账

• 修改转账代码：

• 业务改成 A B 互转：

这时，我们再来运行看看结果：

可以看到，我跑了好久了，还是没有结果，说明两个方法阻塞了

这里死锁的情况，我们可以用一个资源图表示：

可以看到 t1 获取了资源 A 想要占有资源B，t2 获取了资源B，想要占有资源A

就这样双方互不相让，导致了死锁的产生

2、死锁出现的条件

既然已经知道会产生死锁了，那么只有知道为什么会产生死锁，我们才能对症下药

有 OS 知识的同学可能了解，死锁的从产生条件有如下四个：

1）互斥

这里的互斥，指的是资源要互斥

比如说我们的线程t1 获取了资源A 后，线程 t2 就不能获取资源A 了

2）占有且等待

线程 T1 已经取得共享资源 X，在等待共享资源 Y 的时候，不释放共享资源 X。这个名词强调占有，即在等待其他资源的时候，手头的资源绝对不能放掉

3）不可抢占

其他线程不能强行抢占线程 T1 占有的资源

4）循环等待

线程 T1 等待线程 T2 占有的资源，线程 T2 等待线程 T1 占有的资源，就是循环等待

可能上面的四个概念有点抽象，但是我举个例子，各位可能就能比较好的理解和记忆了

假设双十一，线下店要卖 ps5 和 xbox ，而且各只有一台，你和你的舍友两台都向买到

双十一一大早，你和舍友早早的冲进店里，你买到了 ps5，舍友买到了 xbox，但是因为都只有一台，你就没法买到 xbox 了，你的舍友也就买不到 ps5了**（资源互斥）**

这个时候你就开始和你的舍友协商了，希望把他手里的 xbox 搞到手，但是你心里想的是，自己手上的 ps5 绝对不能给舍友**（占有且等待）**

你和舍友两个人都是身强体壮，你知道你打不过他，他也打不过你，所以硬从他手上抢走 xbox 是没办法的，但是舍友也没法把你手上的 ps5 抢走**（不可抢占）**

你俩在收银台前发誓，不把两台主机拿到，誓不结账回家，就这样你能我松口我等你松口，谁也不让谁，后面等着结账的人都快等死了（线程阻塞），你俩也不肯防守，就等着对方撒手**（循环等待）**

说实话，即使是这样，在面试的时候这四个必要条件我也经常会忘记一两个，不过在向我班上的那些做题家咨询之后，我了解到了一个记忆的好法子，就是把每个关键点总结成一两个字，然后将所有字串联起来形成一句话进行记忆，最好在把这句话具象化为一个画面

（斥占等不循）-> 赤脚站着等，不询问路该怎么走 <可以想象十字路口有个迷路的人赤脚站着在等待，但是不 询问路该怎么走>

我为什么要大费周章的强调记住这四个条件的重要性，还提出这么多奇葩的记忆方法，这是因为记住他们，对我们后面解决死锁问题十分重要

3、避免死锁

上面我们提到的四个死锁的产生条件，都是必要条件

也就是说，其中任何一个条件没有满足死锁就不会产生

那么，避免死锁的思路就来了

1）破坏互斥？？

互斥条件是没法破坏的，毕竟你也不能凭空再变出个资源出来吧

所以这个死锁条件，我们是无法破坏的

2）破坏占有且等待

对于“占用且等待”这个条件，我们可以一次性申请所有的资源，这样就不存在等待了。

有过操作系统学习经验的同学，也许对这个解决方式十分熟悉，没错，这就是经典的哲学家就餐问题其中的一种解决方法

所谓哲学家就餐问题，就是有5个哲学家，5盘菜，5把叉子，每个哲学家在吃菜的时候，必须使用左右手两把叉子（至于为什么要使用两把叉子，这里暂就不深究了），如果在同一时间内，五个哲学家同时拿起自己左手边的叉子，那么，所有人右手边就都没有叉子了，又因为没有人愿意放下手中的叉子，所以大伙就都吃不了饭了（类似于我们的死锁情况）

对于这种问题的解决方式，我们是分配一个服务生，当有哲学家想要吃东西的时候，只能向服务生申请叉子，只有在这个哲学家左右都有叉子的时候，服务生才会同意哲学家就餐，并为他分配叉子。通过这种方式，我们就避免了只获取部分资源而等待的情况（占有且等待），从而避免了死锁

那在代码中该怎么实现呢？

我们可以借助微服务中注册中心的概念，在 Account 类中，新增一个 Allocator 的单例，专门用来处理对象的申请

有这个 Allocator 存在，假设某个转账线程要进行 A->B 的转账，会先走 Allocator ，如果 Allocator 发现当前只有资源 A，没有资源B，那就回拒绝这次操作，线程就会进入等待状态

这里要注意一点，微服务中，注册中心是只有一个的，所以咱们的 Allocator 也必须是单例的

• Allocator：

注意这里 Allocator 的单例写法

class Allocator {
    // 存储在这个队列里的对象，表示正在被某个线程使用
    private final Set<Object> usingQueue = new HashSet<>();
    private static Allocator instance = new Allocator();

    private Allocator() {}

    public static Allocator getInstance() {
        return instance;
    }

    public boolean apply(Object o1,Object o2) {
        if (usingQueue.contains(o1) || usingQueue.contains(o2)) {
            return false;
        }
        usingQueue.add(o1);
        usingQueue.add(o2);
        return true;
    }

    public void free(Object o1,Object o2) {
        usingQueue.remove(o1);
        usingQueue.remove(o2);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

这里写 Allocator 必须注意两个问题：

**1、**咱们的使用对象队列 usingQueue 照理说也应该是线程安全的（JUC里有对应的线程安全类，这里为了演示方便，我们就先用 HashSet 代替一下了）

**2、**这里因为会对 Account 对象的属性值进行修改（其实只会修改余额 balance），所以在重写 equals 和 hashcode 的时候要注意，只能对唯一且不可变的对象属性进行重写（即 bankID 属性）

如果不这么做，会在修改 balance 的值之后，导致 free 方法中对集合的 remove()方法无法执行（内存泄露，hashcode 和 equals 都不相等）

public void free(Object o1,Object o2) {
  usingQueue.remove(o1);
  usingQueue.remove(o2);
}
1
2
3
4

对第二点还不太清楚的小伙伴，建议看一下 HashSet 和 HashMap 的原理，这里推荐文章我也贴在后面了：https://blog.csdn.net/fmwind/article/details/76460681

• 有了上面的准备，对应的 transfor 方法就该这么写：

// 转账方法
    public boolean transfer(Account target,long amount,int delay) {
        // 当无法申请到资源的时候，当前线程持续等待
        while (!allocator.apply(this,target));
        try {
            synchronized (this) {
                // 模拟获取第一个资源的时候，线程切换
                // 尝试达成达到死锁的目的
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (target) {
                    if (this.balance<amount) return false;
                    this.balance-=amount;
                    // 将要为被转用户设置的新 balance 数
                    long balanceToBeSet = target.getBalance() + amount;
                    // 在转账和被转账账户之间，设置睡眠，模拟各大银行之间的转账延迟
                    try {
                        Thread.sleep(delay);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    target.setBalance(balanceToBeSet);
                    //System.out.println(Thread.currentThread()+"转账完毕，this 剩余余额为："+this.balance+"，target剩余余额为："+target.balance);
                }
            }
        } finally {
            allocator.free(this,target);
        }
        return true;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

这个例子中，有个地方可能写的不是太妥当：

在真实业务中，我们不应该让某个线程或者请求一直等待资源，而是应该对其设置一个超时时间

3）破坏不可抢占

对于“不可抢占”这个条件，占用部分资源的线程进一步申请其他资源时，如果申请不到，可以主动释放它占有的资源，这样不可抢占这个条件就破坏掉了。

对于这个方法的实现，我们用 synchronized 是无法实现的，因为 synchronized 是自动加锁解锁的，在线程执行结束之前，是不会释放锁的

其实，对于这个方法的实现，在 java SDK 层面是没有实现的，但是在我们后面的 J.U.C 中，会有对这种方法的实现，这里就暂时不表了

4）破坏循环等待

对于“循环等待”这个条件，可以靠按序申请资源来预防。所谓按序申请，是指资源是有线性顺序的，申请的时候可以先申请资源序号小的，再申请资源序号大的，这样线性化后自然就不存在循环了

我们之前在占有且等待实现中提到过哲学家就餐问题，但其还有一种解决方案，那就是对叉子进行编号

对叉子有小到大进行编号，哲学家就餐的时候，必须先拿编号小的，再拿编号大的叉子，吃完后，必须先放下编号大的，再放下编号小的叉子

对于我们银行转账的业务，我们可以对 bankID 进行大小比较

public class Account {
    private long balance;
    private String bankID;

    public Account(String bankID) {
        this.bankID = bankID;
    }

    // 转账方法
    public boolean transfer(Account target,long amount,int delay) {
        Account left = this;
        Account right = target;
        // 保证左小右大
        if (this.bankID.compareTo(right.bankID)==1) {
            left=target;
            right=this;
        }
        synchronized (left) {
            // 模拟获取第一个资源的时候，线程切换
            // 尝试达成达到死锁的目的
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            synchronized (right) {
                if (this.balance<amount) return false;
                this.balance-=amount;
                // 将要为被转用户设置的新 balance 数
                long balanceToBeSet = target.getBalance() + amount;
                // 在转账和被转账账户之间，设置睡眠，模拟各大银行之间的转账延迟
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                target.setBalance(balanceToBeSet);
                //System.out.println(Thread.currentThread()+"转账完毕，this 剩余余额为："+this.balance+"，target剩余余额为："+target.balance);
            }
        }
        return true;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

这里因为 synchrpnized 释放锁的顺序是从最后一个加锁的部分开始释放的，所以也满足我们从小到大加锁，从大到小解锁的顺序

• 最后运行成功，没有出现死锁：

不过这里我要说了，虽然这种按照循序获取锁的方式不需要像处理占有且等待那样维护一个分配资源的单例，但是其也有其弊端

我们假设有10个线程，10个资源，线程1需要资源1，2；线程2需要资源2，3；线程3需要资源3，4…以此类推。假设这个请求是同时发生的，那么线程1-9都会出现循环等待，只有在到线程10时，因为其需要的资源是资源 10 和 1，因为资源1已经被线程1占有了，所以这个时候，线程10不会占有资源10，那么线程9就可以同时获取资源9和10，从而完成任务，这样依次往后，线程8，7，6，…2，1，10才可以完成任务。那么，我们本来期望的并发，就变成了串行，效率没有任何提升

而如果使用破坏占有且等待的方法，这种情况下，线程1，3，5，7，9就可以并发执行，理论情况下，其效率是破坏循环等待这种方法的5倍

所以，现实情况中，对资源编号避免死锁的方式因为效率不高，我们很少用

五、小结

让我们回顾一下今天的知识，今天，我们借由银行转账这个案例，按照如下顺序，由浅入深的探讨了遇到并发问题，该如何思考

**1、**首先要确定加锁范围，起码要保证并发安全性吧（方法锁，类锁）

**2、**在最基础的保证并发安全性的情况下，咱就要开始思考效率问题（避免串行化，使用细粒度锁）

**3、**但是在使用细粒度锁的同时，又可能引发死锁问题，我们又通过死锁的四个必要条件，反向思考该如何避免死锁（资源互斥、占有且等待、不可抢占、循环等待）

由此，我们就完成了一个线程安全、净量高效、不会死锁的转账方法（当然目前这个方案还是有些地方欠考虑，我们在之后的文章中会继续对这个案例进行优化的）

希望这种由浅入深的方式，一是能帮助各位更好的理解、阅读本文，二是希望这种思考问题的方式，能对各位日后处理各种问题带来启发

还请各位不要小瞧这个银行转账的案例，这可以说是并发问题中的经典案例，笔者本人就是因为当初对这类问题的不熟悉，败在了蔚来一面上（当初面试的时候，甚至连问题场景都没有改，就是银行转账），所以还是希望各位能重视起来