分布式锁之防止超卖 --mysql原子操作，乐观锁，redis事务，乐观锁

读未提交解决分布式超卖

（单体项目）

读已提交的代码

@Service
public class StockServiceImpl extends ServiceImpl<StockMapper, Stock>
    implements StockService{

    ReentrantLock lock = new ReentrantLock() ;

    @Transactional
    public void deduct(){
   // 上锁
        lock.lock();
         // 查库存 扣库存
          LambdaQueryWrapper<Stock> queryWrapper = new LambdaQueryWrapper<>();
          queryWrapper.eq(Stock::getProductCode,"1001") ;

          Stock stock = this.getOne(queryWrapper);
          if (stock!=null && stock.getCount() > 0){
              System.out.println("当前库存："+  stock.getCount());
              stock.setCount(stock.getCount()-1);
              this.updateById(stock) ;
          }

         lock.unlock();
    }

}
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

发生超卖现象
原因： 事务2读取到事务1 未提交的数据 ，两个事务会同时读到同一个数据 ，然后减一其实只减少了一次
解决：改为 @Transactional(isolation = Isolation.READ_UNCOMMITTED),

JVM 本地锁

可以解决 多例模式 的超卖，但是不能解决 集群部署下的

Mysql锁

sql语句解决超卖问题

集群环境下

 <update id="updateStock">
       update demo_03.db_stock set count = count - #{count}
       <where>
           product_code = #{productCode}
           and
           demo_03.db_stock.count > #{count}
       </where>
    </update>
1
2
3
4
5
6
7
8

局限：

1. 锁范围：选择表锁还是行级锁
2. 仓库位置写死了 ，不能动态的 从不同仓库取货，而且这样会导致间隙锁（间隙锁产生的原因就是 因为按照商品code 进行更新 直接锁住了 所有的数据，导致都不能进行添加。（这里不是表锁））
   间隙锁的产生原因：是索引：1. 范围更新
   不是索引：1.范围更新 2. 对不是索引的字段进行更新
   
   
   客户端2无法更新商品 1002，因为商品code 不是索引 ，导致间隙锁
   
   
   

4. 无法记录其他的一些数据,例如仓库扣减记录

mysql悲观锁解决并发问题

# 查询并锁住 1001 商品 ，其他客户端无法对1001商品进行操作 for update也是一种行级锁
select * from db_stock where product_code = '1001' for update ;
1
2

  @Select("select * from demo_03.db_stock where product_code = #{productCode} for update ")
    List<Stock> queryStock(String productCode);

    @Transactional
    public void deduct(){
        // 查询并且 锁定仓库
     List<Stock> list = stockMapper.queryStock("1001") ;

      //判空 选取并且 扣减一个仓库
        Stock stock = list.get(0);
        //扣减库存
        if (stock!=null && stock.getCount() >0){
            stock.setCount(stock.getCount()-1);
        log.info("当前库:{},当前库存数量:{}",stock.getWarehouse(),stock.getCount());
            this.stockMapper.updateById(stock) ;
        }

    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

先查询库存，然后锁定，然后 分析 ，最后减库存 。
缺点：1.加锁顺序不一样会导致死锁，1加锁 2加锁 ，1想获得2 的锁会被 阻塞
2.select for update 对一条数据进行绑定 ，就不能使用普通的select

mysql 自旋锁 解决并发问题

select version   -> version = 1
update  ....where  version = 1
1
2

问题

1.并发量很低。更新压测失败率很高
为了确保成功 可以 递归调用该方法 或者while

取消事务注解 和 增加try cache避免递归 堆溢出，但是不停重试浪费cpu资源

2. ABA 问题 ：
3. 读写分离导致乐观锁 不可靠，主从集群，由于io操作阻塞，主里面是新数据，但是从里面是旧数据。会导致并发性问题

mysql锁总结

性能：一个sql > 悲观锁 > jvm锁 > 乐观锁

如果追求极致性能、业务场景简单并且不需要记录数据前后变化的情况下。

​ 优先选择：一个sql

如果写并发量较低（多读），争抢不是很激烈的情况下优先选择：乐观锁

如果写并发量较高，一般会经常冲突，此时选择乐观锁的话，会导致业务代码不间断的重试。

​ 优先选择：mysql悲观锁

不推荐jvm本地锁。

redis乐观锁

redis开始事务和监听，当 别的客户端修改了数值的时候 ，此事务会执行失败
watch 机制和原理（用来监听key）

127.0.0.1:6379> watch stock1 
OK
127.0.0.1:6379> multi
OK
// 此时客户端2 执行 set stock1 15

127.0.0.1:6379(TX)> set stock1 13
QUEUED
127.0.0.1:6379(TX)> exec 
(nil)  // 执行失败 

1
2
3
4
5
6
7
8
9
10
11

记住 ： 乐观锁就是不上锁 ，有个类似版本号的东西，这里的版本号就是来监控变化的 这里 watch自动监控了。

基本和mysql乐观锁一样的操作，但是失败率太高，失败就要重试 ，由于受最大连接数限制，你在重试，新的连接进进不来，并发量很低。

java 实现

1. 这里执行需要引入回调函数

/**
 * Callback executing all operations against a surrogate 'session' (basically against the same underlying Redis
 * connection). Allows 'transactions' to take place through the use of multi/discard/exec/watch/unwatch commands.
 *
 * @author Costin Leau
 */
public interface SessionCallback<T> {

	/**
	 * Executes all the given operations inside the same session.
	 *
	 * @param operations Redis operations
	 * @return return value
	 */
	@Nullable
	<K, V> T execute(RedisOperations<K, V> operations) throws DataAccessException;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Executes all the given operations inside the same session.： 在一个会话里执行一些列操作

 redisTemplate.execute(new SessionCallback<Object>() {

            // 执行事务 + 回调
            @Override
            public  Object execute(RedisOperations  operations) throws DataAccessException {

                // 具体事务
                // watch
                operations.watch(stockS);
                // check stock1
                String stockCount = operations.opsForValue().get(stockS).toString();
                // multi
                operations.multi();
                if (stockCount!=null && stockCount.length() > 0){
                    int now  = Integer.valueOf(stockCount).intValue();
                    operations.opsForValue().set(stockS , String.valueOf(--now)) ;
                }
                // exec
                List exec = operations.exec();

                // 执行事务结果集为空 是 失败，继续重试
                if (exec==null || exec.size() == 0){
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    deduct();
                }
                return null;
            }
        }) ;

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

我的文章还有关于redisson 和zeepkooper 的分布式锁
zoopkeeper分布式锁
redis分布式锁

ReentrantLock 的原理