电商减库存存在的安全问题

@Override
public void deductStock(Map<Long, Integer> skuMap) {for (Map.Entry<Long, Integer> entry : skuMap.entrySet()) {Long skuId = entry.getKey();Integer num = entry.getValue();// 查询skuSku sku = getById(skuId);//  判断库存是否充足if (sku.getStock() < num) {// 如果不足，抛出异常throw new LyException(400, "库存不足！");}// 如果充足，扣减库存 update tb_sku set stock = stock - 1, sold = sold + 1  where id = 1Map<String,Object> param = new HashMap<>();param.put("id", skuId);param.put("num", num);getBaseMapper().deductStock(param);}
}

上面这样的操作存在安全风险，因为我们的代码是允许多线程的环境，当多个用户并发访问时，先判断库存是否充足，会出现一种情况：

• 判断的时候，库存是充足的，但是在减库存之前，有其它线程抢先一步，扣减库存，导致库存不足了，此时就会出现超卖现象！

思路1，同步锁
按照以往的思路，我们应该怎么做？

• 我们一般需要加同步锁，synchronized,目的是让多线程执行，从而保证线程安全，但是加Synchronized只能保证当前jvm内的线程安全。
• 如果搭建一个微服务集群，同步锁synchronized就失效了，原因是因为线程所，在进程时会失效，因为每个进程都有自己的锁。
• 解决多进行安全的问题，必须使用进程锁（分布式锁）：
• 

标题：分布式锁

分布式锁其实可以这样理解为：控制分布式系统有序的去对共享资源进行操作，通过互斥保持一致性
举个不恰当的例子：假设共享资源就是一个房子，里面有各种各样的书，分布式系统就是要进屋子里看书的那个人，分布式锁就是保证这房子里面，只有一个门，并且一次只能一个人进去，而且门只有一把钥匙，然后许多人进去看书，可以，排队，第二个人没有钥匙，那就等着，等第一个人出来，然后你在拿这钥匙进去，就这样以此类推。

二：实现原理

• 互斥性
• 保证同一时间只要一个客户端可以拿到锁，也就是可以对共享资源进行操作
• 安全性
• 只有加锁的服务才能有解锁的权限，也就是不能让a加的锁，bcd都可以解锁，如果都能解锁，那分布式就没有意义了
• 可能出现的情况就是a去查询发现持有锁，就在准备解锁，这时候突然a持有的锁过期了，然后b就去获取锁，因为a锁过期，b拿到锁，这时候a继续执行第二部进行解锁如果不加校验，就将b持有的锁就给删除了
  避免死锁
• 出现死锁就会导致后续的任何服务都拿不到锁，不能在对共享资源进行任何操作了
• 保证加锁与解锁操作是原子性操作
• 这个其实属于是实现分布式锁的问题，假设a用redis实现分布式锁
• 假设枷锁操作，操作步骤分为两步
• 设置key set（key ,value) 2：给key设置过期时间
• 假设现在a刚实现set后，程序崩了就导致了没给key设置过期，时间就导致key一直存在就发生了死锁

三.使用redis实现分布式锁

首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足一下四个条件

• 互斥性，在任意时刻，只要一个客户端能持有锁

• 不会发生死锁，即使有一个客户端持有锁的时候崩溃而没有主动解锁，也能保证后续其他客户端能加锁

• 具有容错性，只要大部分的redis节点正常允许，客户段就可以加锁和解锁

• 解铃还须系铃人，枷锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

  可以看到，我们加锁就一行代码：jedis.set(String key, String value, String nxxx, String expx, int time)，这个set()方法一共有五个形参：

• 第一个为key，我们使用key来当锁，因为key是唯一的。

• 第二个为value，我们传的是requestId，很多童鞋可能不明白，有key作为锁不就够了吗，为什么还要用到value？原因就是我们在上面讲到可靠性时，分布式锁要满足第四个条件解铃还须系铃人，通过给value赋值为requestId，我们就知道这把锁是哪个请求加的了，在解锁的时候就可以有依据。requestId可以使用UUID.randomUUID().toString()方法生成。

• 第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作；

• 第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定。

• 第五个为time，与第四个参数相呼应，代表key的过期时间。

总的来说，执行上面的set()方法就只会导致两种结果：1. 当前没有锁（key不存在），那么就进行加锁操作，并对锁设置个有效期，同时value表示加锁的客户端。2. 已有锁存在，不做任何操作。

心细的童鞋就会发现了，我们的加锁代码满足我们可靠性里描述的三个条件。首先，set()加入了NX参数，可以保证如果已有key存在，则函数不会调用成功，也就是只有一个客户端能持有锁，满足互斥性。其次，由于我们对锁设置了过期时间，即使锁的持有者后续发生崩溃而没有解锁，锁也会因为到了过期时间而自动解锁（即key被删除），不会发生死锁。最后，因为我们将value赋值为requestId，代表加锁的客户端请求标识，那么在客户端在解锁的时候就可以进行校验是否是同一个客户端。由于我们只考虑Redis单机部署的场景，所以容错性我们暂不考虑。

思路2，数据库排它锁

数据库锁简单来说有两种：

• 共享锁：读操作时会开启共享锁，此时大家都可以查询
• 排他锁（互斥锁）：一般是写操作会开启排它锁，此时其他事务无法获取共享锁或排它锁，会阻塞

要保证安全，必须排它锁
但是我们之前的业务是先查询sku(读），然后判断是否充足，然后减库存（写），这样就会导致多个请求同时查询到一样的库存，减库存还是有安全问题
我们必须在查询时加排他锁，怎么办

• 可以通过select …for update语法来开启，但时我们要加锁的商品不止一个，此时加锁就是范围锁，甚至时表锁，性能会有较大的影响

思路3：乐观锁

上述思路1和思路2都是枷锁，实现互斥，保证线程的安全，我们称之为悲观锁。

• 悲观锁：认为线程安全问题一定会发生，因此会加锁保证线程串行执行，从而保证安全，我们为了追求性能，可以使用乐观锁的机制
• 乐观锁，认为线程安全的问题一定会发生，因为允许许多线程并执行，一般会在执行那一刻进行判断和比较，然后根据是否存在风险来决定是否执行操作
• 举例说明，可以给库存表加一个字段，叫version

id  stock    version
10	10			1

执行更新前，先查询库存及version
然后判断库存是否充足，如果充足，执行sql

update tb_stock set stock = stock - #{num}, version = version + 1 WHERE id = #{id} AND version = 1

乐观锁就先比较执行的思路，其实就是CAS(compare and set)的思想。
CAS的思想在很多地方都使用，例如

• JDK的JUC包下的AtomicInteger,AtomicLong等等
• Redis的watch,也是乐观锁,CAS原理

简化：我们在减库存中，可以用stock来代替version,执行sql时判断stock是否跟自己查询到一样

update tb_stock set stock = stock - 1 WHERE id = 10 AND stock = 10

思路4：继续简化
我们可以不查询库存，直接执行sql,在sql语句中做判断
语句是这样的

update tb_stock set stock = stock - 1 WHERE id = 10 AND stock >= 1

思路5：继续简化
我们最终的目的是，库存不能超卖，不能为负数，因为我们可以设置stock字段为无符号整数，数据库自动会写入数据字段，如果为负，会抛出异常，我们就无需枷锁或任何其他判断了