利用双重检查锁定和CAS算法：解决并发下数据库的一致性问题

刺骨的言语ヽ痛彻心扉 2024-04-17 23:16 25阅读 0赞

### 背景 ###

最近有一个场景遇到了数据库的并发问题。现在先由我来抽象一下，去掉不必要的繁杂业务。  
 数据库表book存储着每本书的阅读量，一开始数据库是空的，不存在任何的数据。当用户访问接口的时候，判断表book是否存在此书的记录，如果不存在，即插入一条新记录，而且阅读量设置为1；当下个用户再阅读此书时，再调用接口就直接将此书的阅读量增加1，而不用再插入新记录。

### 并发下出现的问题 ###

下面看一下伪代码：

public void addOrUpdateBook{ 
        Book oldBook = this.bookMapper.selectByBookName(bookName);
        if (oldBook == null){  // 1、判断书本记录是否为空
            this.bookMapper.insertSelective(b);  // 2、新增书本记录
        }else{ 
            Integer updateCount = this.bookMapper.updateReadFrequency(book); // 3、更新阅读量
        }
    }

1、插入多条相同的书本记录：

假如此时线程A在【1】处判断出书本记录为空，然后在【2】处进行创建书本记录；如果线程B在线程A提交前又进来【1】处判断书本是否为空，因为线程A还没创建完和提交事务，数据库中的书本记录还是为空，所以线程B也开始进行书本记录的创建。最后的结果：数据库中存在两条此书的记录。

2、书本的阅读量更新后不准确

假如书本《Java并发编程的艺术》的阅读量为10，然后此时线程A进行阅读量加1的操作，但是在线程A提交事务前，线程B也进行阅读量加1的操作，那么就是不管是线程A还是线程B，都是在阅读量10的基础上加1，最后当两个线程都执行完时，《Java并发编程的艺术》的阅读量就是11，这和正确结果阅读量12是不一样的！

### 思考 ###

此时我们都知道了，并发会导致数据库数据的一致性问题。那么，我们该怎么解决呢？

1、方法加锁

最简单的做法，就是直接在方法那里加synchronized关键字，将整个方法锁起来，每个请求只能一个一个按照顺序执行，那么插入和更新的并发问题都不存在了。

public synchronized void addOrUpdateBook{ 
        Book oldBook = this.bookMapper.selectByBookName(bookName);
        if (oldBook == null){  // 1、判断书本记录是否为空
            this.bookMapper.insertSelective(b);  // 2、新增书本记录
        }else{ 
            Integer updateCount = this.bookMapper.updateReadFrequency(book); // 3、更新阅读量
        }
    }

2、双重检查锁定

上面的方法性能是最低的，未获取到锁的所有请求都会被阻塞在方法外面（不管是否存在并发问题）。其实，我们可以利用双重检查锁定来解决这个锁性能低问题，做法也是非常的简单，就是我们只管对可能出现插入并发问题的代码进行上锁就行了，就是说从同步方法改为同步块，不再锁住不必要的代码。

public void addOrUpdateBook{ 
    	Book oldBook = this.bookMapper.selectByBookName(bookName);
    	if (oldBook == null){  // 1、判断书本记录是否为空
    	    // 加锁
    	    synchronized (this){ 
    	        oldBook = this.bookMapper.selectByBookName(bookName);
    	        if (oldBook == null){ 
    	            BeanUtil.copyProperties(query,b);
    	            this.bookMapper.insertSelective(b); // 2、新增书本记录
    	        }else{ 
    	            Integer updateCount = this.bookMapper.updateReadFrequency(book); // 3、更新阅读量 
    	        }
    	    }
    	}else{ 
    	    Integer updateCount = this.bookMapper.updateReadFrequency(book); // 4、更新阅读量
    	}
    }

我们可以看到：假如线程A此时在【2】处进行记录新增，而此时线程B也在【1】处判断书本记录为空，然后被阻塞在同步块外，当线程A执行完释放锁后，线程B获取到锁，但是此时书本记录不再为空，线程B就直接更新阅读量而不再插入书本记录。当然了，后续的所有请求在第一重就能判断出书本记录不为空，然后直接更新阅读量。

那么就是说，我们的插入并发问题解决啦，而且性能比同步方法高不少。但是呢，更新的并发问题还没解决，为什么我没有同时也锁住更新阅读量的代码呢？因为我觉得没啥必要，因为为了减少线程上下文的切换，我们都推荐无锁并发编程，那么我们能怎么做呢？下面将介绍如何参考CAS算法来防止更新的并发问题。

3、CAS算法

首先，我们的表需要增加一个字段：版本号。对，这时候大家可能想到了乐观锁。没错啦~CAS算法其实就是一种乐观锁。它的原理是：比较再交换；更新时，当我们保存的旧值和数据库的值一致时，我们就能将旧值更新为我们的新值。最后，我们利用死循环来不断循环保证最后能更新成功。大家可能会疑惑死循环会不会导致性能很低？其实还好啦，起码能避免了线程的上下文切换，而且，一般同时的请求量也不会这么离谱（我们公司），并发量很大可能要做其他的方案了。下面我们先上一下代码：

public Boolean addOrUpdateBook(BookQuery query) { 
         boolean flag = true;
         // 使用双重检查锁定来处理新增的并发问题
         Book b = Book.builder().bookName(query.getBookName()).build();
         Book oldBook = this.bookMapper.selectOne(b);
         if (oldBook == null){ 
             // 加锁
             synchronized (this){ 
                 oldBook = this.bookMapper.selectOne(b);
                 if (oldBook == null){ 
                     BeanUtil.copyProperties(query,b);
                     this.bookMapper.insertSelective(b);
                 }else{ 
                     updateBook(query);
                 }
             }
         }else{ 
             updateBook(query);
         }
         return flag;
     }
    
     /** * 参考CAS的无锁算法来处理更新的并发问题（利用死循环+版本号） * @param query */
     private void updateBook(BookQuery query){ 
         // 参考cas
         for (;;){ 
             // 获取当前记录的版本号
             Integer version = this.bookMapper.getVersionByBookName(query.getBookName());
             query.setVersion(version);
             // 根据书名和版本号进行阅读量更新
             Integer updateCount = this.bookMapper.updateBookByVersion(query);
             if (updateCount != null && updateCount.equals(1)){ 
                 // 如果更新成功就跳出循环
                 break;
             }
         }
     }

更新阅读量的sql：

/** * 根据书名和版本号更新book * @param book * @return */
    @Update("update book set version = version+1,read_frequency = #{readFrequency} where book_name = #{bookName} and version = #{version}")
    Integer updateBookByVersion(BookQuery book);

这里的代码就不多做解释了，看一下注释就知道是什么原理，是非常简单易懂滴~

4、利用Redis做分布式锁

再回到加锁的那里，我们可以发现，当我们做微服务时，一般每个服务都会是多个实例，或者是单体应用的实例部署，我们的锁就只能针对单体应用有效了，而多个实例还是会导致插入的并发问题。这时候我们必须想到：分布式锁！

我知道的现在主要做分布式锁有两种方式，一种是基于Redis的分布式锁，另外一种是基于Zookeeper的分布式锁。简单分析一下上面两种锁的优缺点：从可靠性上来说，Zookeeper分布式锁有好于Redis分布式锁；而从性能上来说，Redis分布式锁要好于Zookeeper分布式锁，毕竟Redis是纯内存操作的，性能是想当的好，号称每秒可以处理10万次读写操作呢，所以我最终选择了基于Redis的分布式锁。

不过还有一个问题就是：大家玩过redis的都知道，redis只保证单个操作是具有事务的，是原子性的，多个操作就不能保证原子性了。因为我们一般加锁的做法都是，使锁具有超时的特征，避免一个请求无条件的等待锁，一直的阻塞导致系统CPU飙升。所以单单利用redis的方法我们做不到具有超时特征的分布式锁。当然了，现在有两种比较好的解决方案：一种是利用redis+lua脚本，一种是利用开源的框架Redisson。当然了，有简单的必须就利用简单的了，下面开始介绍如何利用Redisson开发分布式锁。

1）加入redis和redisson依赖

<dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
          <groupId>org.redisson</groupId>
          <artifactId>redisson</artifactId>
          <version>3.11.3</version>
    </dependency>

2）创建Redisson的工具类

package com.hyf.utils;
    
    import org.redisson.Redisson;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    import org.redisson.config.Config;
    
    /** * @author Howinfun * @desc Redisson工具类 * @date 2019/9/2 */
    public class RedissonUtil { 
    
        private static RedissonClient redissonClient;
    
        private RedissonUtil(){ 
            // 构造redisson实现分布式锁必要的Config
            Config config = new Config();
            config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456").setDatabase(1);
            // 构造RedissonClient
            redissonClient = Redisson.create(config);
        }
    
        public static final RedissonUtil INSTANCE = new RedissonUtil();
    
        /** * 设定锁定资源名称，返回锁 * @param name * @return */
        public RLock getLock(String name){ 
            return redissonClient.getLock(name);
        }
    }

3)修改业务代码

public Boolean addOrUpdateBook(BookQuery query) { 
        boolean flag = true;
        // 使用双重检查锁定来处理新增的并发问题
        Book b = Book.builder().bookName(query.getBookName()).build();
        Book oldBook = this.bookMapper.selectOne(b);
        if (oldBook == null){ 
            RLock lock = RedissonUtil.INSTANCE.getLock("addOrUpdateBook");
            try { 
                // 尝试获取锁，最多等待10000毫秒，获取锁后1000毫秒自动释放锁
                if (lock.tryLock(10000, 10000, TimeUnit.MILLISECONDS)){ 
                    oldBook = this.bookMapper.selectOne(b);
                    if (oldBook == null){ 
                        BeanUtil.copyProperties(query,b);
                        this.bookMapper.insertSelective(b);
                    }else{ 
                        updateBook(query);
                    }
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
                return false;
            }finally { 
                // 最后记得释放锁
                lock.unlock();
            }
        }else{ 
            updateBook(query);
        }
        return flag;
    }
    
    /** * 参考CAS的无锁算法来处理更新的并发问题（利用死循环+版本号） * @param query */
    private void updateBook(BookQuery query){ 
        // 参考cas
        for (;;){ 
            Integer version = this.bookMapper.getVersionByBookName(query.getBookName());
            query.setVersion(version);
            Integer updateCount = this.bookMapper.updateBookByVersion(query);
            if (updateCount != null && updateCount.equals(1)){ 
                break;
            }
        }
    }

到这里，我们的代码已经比较好使的了，不但能预防数据库插入和更新的并发问题，还能在分布式环境下也好使！

我猜大家是不怎么相信的了，那么下面我将用JMeter来测试一下。

1）首先创建一个线程组：一共有10个线程，执行1次。

![format_png][]

2）创建Http请求：消息体中的数据是循环读取CSV数据文件里的数据的![format_png 1][]

3）创建Http信息头管理：加上Content-Type

![format_png 2][]

4）创建并指定CSV数据文件

![format_png 3][]

![format_png 4][]

5）点击启动。

观察结果树：可以发现所有的请求都是请求成功的，没有报错。

![format_png 5][]

再观察一下控制台，可以发现只有一条插入sql和九条更新sql：

2019-09-03 09:28:39.888 DEBUG 8020 --- [nio-8888-exec-7] com.hyf.mapper.BookMapper.selectOne      : ==> Parameters: Java并发编程的艺术(String)
    2019-09-03 09:28:39.889 DEBUG 8020 --- [nio-8888-exec-7] com.hyf.mapper.BookMapper.selectOne      : <==      Total: 0
    2019-09-03 09:28:39.937 DEBUG 8020 --- [nio-8888-exec-7] c.hyf.mapper.BookMapper.insertSelective  : ==>  Preparing: INSERT INTO book ( id,book_name ) VALUES( ?,? ) 
    2019-09-03 09:28:39.937 DEBUG 8020 --- [nio-8888-exec-7] c.hyf.mapper.BookMapper.insertSelective  : ==> Parameters: null, Java并发编程的艺术(String)
    2019-09-03 09:28:39.939 DEBUG 8020 --- [nio-8888-exec-7] c.hyf.mapper.BookMapper.insertSelective  : <==    Updates: 1
    2019-09-03 09:28:39.950 DEBUG 8020 --- [nio-8888-exec-6] c.h.m.BookMapper.updateBookByVersion     : ==>  Preparing: update book set version = version+1,read_frequency = read_frequency+1 where book_name = ? and version = ? 
    2019-09-03 09:28:39.951 DEBUG 8020 --- [nio-8888-exec-6] c.h.m.BookMapper.updateBookByVersion     : ==> Parameters: Java并发编程的艺术(String), 1(Integer)
    2019-09-03 09:28:39.953 DEBUG 8020 --- [nio-8888-exec-6] c.h.m.BookMapper.updateBookByVersion     : <==    Updates: 1
    2019-09-03 09:28:39.957 DEBUG 8020 --- [nio-8888-exec-5] c.h.m.BookMapper.updateBookByVersion     : ==>  Preparing: update book set version = version+1,read_frequency = read_frequency+1 where book_name = ? and version = ? 
    2019-09-03 09:28:39.957 DEBUG 8020 --- [nio-8888-exec-5] c.h.m.BookMapper.updateBookByVersion     : ==> Parameters: Java并发编程的艺术(String), 2(Integer)
    2019-09-03 09:28:39.959 DEBUG 8020 --- [nio-8888-exec-5] c.h.m.BookMapper.updateBookByVersion     : <==    Updates: 1
    。。。。。。。

最后再看一下数据库：我们可以看到只有一条记录，而且阅读量为10，非常的准确！![format_png 6][]

如果大家对此demo感兴趣的话，可以到github上和码云上拉取项目，项目里头还包含JMeter的测试用例噢：  
[GitHub][]  
[码云][Link 1]

### 总结 ###

平时我们程序猿真的要多看书，虽然我自己也没看多少书，也没能好好坚持，但是从上个月开始我就下定决心好好看书了。也是因为最近在阅读《Java并发编程的艺术》，所以才有上面的思考和方案！

如果大家对我的读书笔记和思维导图感兴趣，可以到这里看看：[Java并发编程的艺术-阅读笔记和思维导图][Java_-]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvMS5wbmc?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvMi5wbmc?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvMy5wbmc?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvNC5qcGc?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvNS5wbmc?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvNi5wbmc?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vSG93aW5mdW4vaW5mb3JtYWwtZXNzYXkvcmF3L21hc3Rlci90aGlua2luZy9pbWFnZXMvNy5wbmc?x-oss-process=image/format,png
[GitHub]: https://github.com/Howinfun/Database-Concurrency-Problem
[Link 1]: https://gitee.com/Howinfun/Database-Concurrency-Problem
[Java_-]: https://blog.csdn.net/Howinfun/article/details/100037479