基于数据库方式和Redis方法实现乐观锁

朱雀 2023-10-11 15:03 10阅读 0赞

### 一、乐观锁意义 ###

乐观锁是相对于悲观锁而言的，**悲观锁**总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁，关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。Java中synchronized和ReentrantLock等独占锁就是悲观锁思想的实现。

**乐观锁**总是假设最好的情况，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号机制和CAS算法实现。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库提供的类似于write\_condition机制，其实都是提供的乐观锁。在Java中java.util.concurrent.atomic包下面的原子变量类就是使用了乐观锁的一种实现方式CAS实现的。

总结：悲观锁只要在其他线程要获取某个资源时，就会加锁，无论是读还是写，都会占有锁，而乐观锁只会在写数据的时候加锁，读数据的时候不加锁，适用于读多写少的情况。

### 二、乐观锁使用场景 ###

*  高并发下的库存扣减  
    使用一下代码，模拟1000个线程来同时下单20号商品，每个线程扣减3个库存，20号商品的初始库存为100个。

![在这里插入图片描述][20201122145337966.png_pic_center]  
数据库设计非常简单，只有以上三个字段

service代码

package com.wk.seckill.service;
    
    import com.wk.seckill.bean.Stock;
    import com.wk.seckill.dao.ProductDao;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Service;
    
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.atomic.AtomicInteger;
    
    @SuppressWarnings("all")
    @Service
    public class ProdService {
        
    
        @Autowired
        private ProductDao productDao;
    
        private AtomicInteger atomicInteger = new AtomicInteger();
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        /**
         * 扣减方法，悲观锁实现同步，会将整个代码块锁定，如果同时有1000个请求，那么
         * 只能一个一个轮流进入方法执行，如果我们每次扣减3个，最多只有33个线程可以扣减成功
         * 剩余的其实已经没有意义了，但剩余的967个线程还是会阻塞查询一遍，造成资源浪费
         */
        public boolean reduce(Integer pid,Integer buys){
        
            //获取商品对象
            Stock stock = productDao.selectById(pid);
            Integer stockCount = stock.getStock();
            if(stockCount < buys){
        
                return false;
            }
            //更新库存
            stock.setStock(stock.getStock() - buys);
            int i = productDao.updateById(stock);
            return i != 0;
        }
    
    
        /**
         * 上面的悲观锁方式，锁住了整个代码，造成效率低下，由此，我们可以使用乐观锁
         * 只在更新的代码块加锁，这样 判断库存是否还有剩余的代码块就不用加锁，如果判断小于就直接返回了
         * 提高了性能
         */
        public boolean reduceOptimisticLock(Integer pid,Integer buys){
        
            while (true){
        
                //获取商品对象
                Stock stock = productDao.selectById(pid);
                Integer stockCount = stock.getStock();
                if(stockCount < buys){
        
                    return false;
                }
                //获取版本号
                Long version = stock.getVersion();
                //更新库存
                if (productDao.updateCountByVersion(pid,buys,version) > 0){
        
                    //更新成功
                    return true;
                }
    
                //更新失败，当前线程随机休眠
                waitForLock();
            }
        }
    
        /**
         * 由于是whlie(true)，所以上面的方法会一直循环执行，直到一直return为止
         */
        private void waitForLock() {
        
            try {
        
                Thread.sleep(new Random().nextInt(10) + 1);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    
    
        /**
         *  redis实现乐观锁，watch机制
         */
        public boolean reduceRedisWatch(Integer pid,Integer buys){
        
            while (true){
        
                //1.开启事务之前watch监视
                redisTemplate.watch(String.valueOf(pid));
               try {
        
                   //2.获取库存
                   String stock = redisTemplate.opsForValue().get(String.valueOf(pid));
                   int count = Integer.parseInt(stock);
                   if (count < buys){
        
                       return false;
                   }
                   //3.开启事务，扣减库存
                   redisTemplate.multi();
                   //更新库存
                   redisTemplate.opsForValue().set(String.valueOf(pid),String.valueOf(count - buys));
                   //4.提交事务
                   List<Object> results = redisTemplate.exec();
                   if(results == null || results.size() == 0){
        
                       //事务执行失败
                       waitForLock();
                       continue;
                   }else {
        
                       return true;
                   }
               }finally {
        
                   redisTemplate.unwatch(); //重新监视
               }
            }
        }
    
        public Integer getLeftCount(Integer pid) {
        
            return productDao.selectById(pid).getStock();
        }
    
        public void updateStock(Stock stock) {
        
            productDao.updateById(stock);
        }
    }

测试代码

import com.wk.seckill.AppSeckill;
    import com.wk.seckill.bean.Stock;
    import com.wk.seckill.dao.ProductDao;
    import com.wk.seckill.service.ProdService;
    import org.junit.Before;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.CyclicBarrier;
    
    @SuppressWarnings("all")
    @SpringBootTest(classes = AppSeckill.class)
    @RunWith(SpringRunner.class)
    public class StockTest {
        
    
        @Autowired
        private ProdService prodService;
    
        private Integer pid = 20;
    
        private Integer buys = 3;
    
        private Integer userNum = 2000;
    
        private CountDownLatch countDownLatch = new CountDownLatch(userNum);
    
        private CyclicBarrier cyc = new CyclicBarrier(1+userNum);
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        //记录购买成功的人数
        private int successPerson;
    
        //记录卖出去的商品个数
        private int saleOutNum;
    
        @Before
        public void init(){
        
            Stock stock = new Stock();
            stock.setId(20);
            stock.setStock(100);
            stock.setVersion(0l);
            prodService.updateStock(stock);
        }
    
        /**
         * 测试库存扣减
         *
         *   悲观锁情况下：2000个线程执行需要 1.6秒左右
         *
         *   乐观锁情况下：2000个线程执行需要 1.3秒左右
         *
         */
        @Test
        public void testReduce() throws InterruptedException, BrokenBarrierException {
        
            long start = System.currentTimeMillis();
            for (int i = 0; i < userNum; i++) {
        
                new Thread(new ProdCallable()).start();
                countDownLatch.countDown(); //倒计时器-1
            }
            cyc.await();
            //Thread.sleep(3000);
            long end = System.currentTimeMillis();
            System.out.println("秒杀用时：" + (end - start));
            System.out.println("购买成功的人数为：" + successPerson);
            System.out.println("卖出去的商品个数为：" + saleOutNum);
            System.out.println("剩余的商品个数为：" + prodService.getLeftCount(pid));
        }
    
        class ProdCallable implements Runnable {
        
    
            @Override
            public void run() {
        
                try {
        
                    countDownLatch.await();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
    
                if (prodService.reduce(pid, buys)) {
        
                    synchronized (countDownLatch) {
        
                        successPerson++;
                        saleOutNum += buys;
                    }
                }
                try {
        
                    cyc.await();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
        
                    e.printStackTrace();
                }
    
            }
        }
    
    }

在不使用任何锁的情况下运行测试  
![在这里插入图片描述][20201122145649182.png_pic_center]很明显，在不使用任何锁的情况下出现了 **超卖**现象

解决

*  使用同步锁  
    在 reduce方法上，加入同步关键字  
    ![在这里插入图片描述][20201122145833659.png_pic_center]

发现加锁之后测试成功，但前面说了synchronized是悲观锁，会将整个代码块锁定。但其实这是没必要的，试想，悲观锁实现同步，会将整个代码块锁定，如果同时有1000个请求，那么只能一个一个轮流进入方法执行，如果我们每次扣减3个，最多只有33个线程可以扣减成功 剩余的其实已经没有意义了，但剩余的967个线程还是会阻塞查询一遍，造成资源浪费。下面使用基于数据库方式的乐观锁机制。

*  mysql实现乐观锁  
    mysql实现乐观锁就是给表记录加一个字段 version，每次修改这行表记录，就让version + 1。

public boolean reduceOptimisticLock(Integer pid,Integer buys){
        
            while (true){
        
                //获取商品对象
                Stock stock = productDao.selectById(pid);
                Integer stockCount = stock.getStock();
                if(stockCount < buys){
        
                    return false;
                }
                //获取版本号
                Long version = stock.getVersion();
                //更新库存
                if (productDao.updateCountByVersion(pid,buys,version) > 0){
        
                    //更新成功
                    return true;
                }
    
                //更新失败，当前线程随机休眠
                waitForLock();
            }
        }
    
        /**
         * 由于是whlie(true)，所以上面的方法会一直循环执行，直到一直return为止
         */
        private void waitForLock() {
        
            try {
        
                Thread.sleep(new Random().nextInt(10) + 1);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }

<update id="updateCountByVersion">
            UPDATE stock
            set stock = stock - #{buys} ,version = version +1
            where id = #{pid}
            and version = #{version}
        </update>

分析数据库实现乐观锁的过程

1.首先我们摒弃了synchronized这种悲观锁的使用  
2.根据商品的id查询商品，并获取当前商品更新的版本号version，  
在更新的时候带着版本号去更新，如果数据库中的版本号和当前查出来的版本号一致，则可以更新，否则扣减库存失败，让当前线程随机休眠几毫米，再次进入循环，直到扣减成功或者库存不足为止。  
3.假设我们开启1000个线程，有100个线程进入这个方法，获取到了库存，此时version在没更新之前默认都是0，虽然有100个线程可以同时进入该方法，但只能有一个线程执行sql，当第一个线程执行完sql，此时version字段变成1，第一个线程更新成功，这时第二个线程来执行sql，发现在执行sql时数据库的version变成了1，而此时该线程查询到的version是0，这就造成了version不一致，更新失败。  
这样我们只在更新数据时加锁，假设前100个线程将库存扣减成功，则后900个线程也不用同步等待，在执行到`if(stockCount < buys){ return false; }`就被返回了。  
![在这里插入图片描述][20201122151948471.png_pic_center]  
此时也可以运行成功，但是效率提高了。

*  Redis实现乐观锁

如果用redis实现乐观锁就要先了解一下redis的事务和watch机制，  
使用 multi 开启事务，使用exec提交事务

redis-server:0>multi
    "OK"
    
    redis-server:0>set k1 v1 
    "QUEUED"
    
    redis-server:0>set k2 v2
    "QUEUED"
    
    redis-server:0>exec
     1)  "OK"
     2)  "OK"

watch用于监视一个key，watch一个key，然后开启事务，如果在事务执行的过程中，这个key被其他客户端修改了，则此次事务提交失败，原来和mysql的version机制差不多，watch \[key\]就是相当于获取这个key的version，然后开启事务更新库存，在提交事务的时候检查watch的这个key的version是否发生了变化，如果发生了变化，则证明在此期间有其他客户端修改了这个key，我们放弃修改，否则更新成功。

注意：在SpringBoot整合redis中，要开启支持事务  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]

public boolean reduceRedisWatch(Integer pid,Integer buys){
        
            while (true){
        
                //1.开启事务之前watch监视
                redisTemplate.watch(String.valueOf(pid));
               try {
        
                   //2.获取库存
                   String stock = redisTemplate.opsForValue().get(String.valueOf(pid));
                   int count = Integer.parseInt(stock);
                   if (count < buys){
        
                       return false;
                   }
                   //3.开启事务，扣减库存
                   redisTemplate.multi();
                   //更新库存
                   redisTemplate.opsForValue().set(String.valueOf(pid),String.valueOf(count - buys));
                   //4.提交事务
                   List<Object> results = redisTemplate.exec();
                   if(results == null || results.size() == 0){
        
                       //事务执行失败
                       waitForLock();
                       continue;
                   }else {
        
                       return true;
                   }
               }finally {
        
                   redisTemplate.unwatch(); //重新监视
               }
            }
        }

如果事务执行成功，集合的size就会大于1，否则就为0。如果为0，让当前线程休眠几秒，再次执行，以此来达到乐观锁机制。

[20201122145337966.png_pic_center]: https://img-blog.csdnimg.cn/20201122145337966.png#pic_center
[20201122145649182.png_pic_center]: https://img-blog.csdnimg.cn/20201122145649182.png#pic_center
[20201122145833659.png_pic_center]: https://img-blog.csdnimg.cn/20201122145833659.png#pic_center
[20201122151948471.png_pic_center]: https://img-blog.csdnimg.cn/20201122151948471.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201122163246687.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2,size_16,color_FFFFFF,t_70#pic_center