秒杀场景下的事务整理——Redis分布式锁的优化

随着互联网的快速发展，产品秒杀的场景咱们并不少见；秒杀是一种供不应求的，高并发的场景，它里边包含了许多技术点，掌握了其间的技术点，虽不一定能让你面试立马成功，但那也必是一个闪耀的点！

前语

假定咱们现在有一个商城体系，里边上线了一个产品秒杀的模块，那么这个模块咱们要怎样规划呢？

秒杀模块又会有哪些不同的需求呢？

大局仅有 ID

产品秒杀本质上其实仍是产品购买，所以咱们需求预备一张订单表来记录对应的秒杀订单。

这儿就涉及到了一个订单 id 的问题了，咱们是否能够像其他表相同运用数据库本身的自增 id 呢？

数据库自增 id 的缺点

订单表假定运用数据库自增 id ，则会存在一些问题：

1. id 的规则太显着了 由于咱们的订单 id 是需求回显给用户查看的，假定是 id 规则太显着的话，会暴露一些信息，比方第一天下单的 id = 10 ， 第二天下单的 id = 11，这就阐明这两单之间底子没有其他用户下单
2. 受单表数据量的约束 在高并发场景下，发生上百万个订单都是有或许的，而咱们都知道 MySQL 的单张表底子不或许容纳这么多数据（功用等原因的约束）；假定是将单表拆成多表，仍是用数据库自增 id 的话，就存在了订单 id 重复的情况了，很明显这是事务不允许的。

根据以上两个问题，咱们能够知道订单表的 id 需求是一个大局仅有的 ID，并且还不能存在显着的规则。

大局 ID 生成器

大局ID生成器，是一种在分布式体系下用来生成大局仅有ID的东西，一般要满意下列特性：

这儿咱们考虑一下是否能够用 Redis 中的自增计数来作为大局 id 生成器呢？

能不能首要是看它是否满意上述 5 个条件：

1. 仅有性，每个订单都是来 Redis 这儿生成订单 id 的，所以仅有功用够确保
2. 高可用，Redis 能够由主从、集群等形式确保可用性
3. 高功用，Redis 是根据内存的，原本便是以功用自称的
4. 递加性，increment 原本便是递加的
5. 安全性。。。这个就费事了点了，由于 Redis 的 increment 也是递加的，规则太显着了。。。

综上，Redis 的 increment 并不能满意安全性，所以咱们不能单纯运用它来做大局 id 生成器。

可是——

咱们能够运用它，再和其他东西拼接起来~

举个栗子：

ID的组成部分：

1. 符号位：1bit，永远为0
2. 时刻戳：31bit，以秒为单位，能够运用69年
3. 序列号：32bit，秒内的计数器，支持每秒发生2^32个不同ID

上面的时刻戳便是用来增加复杂性的

下面给出代码样例：

public class RedisIdWorker {
  /**
   * 开端时刻戳
   */
  private static final long BEGIN_TIMESTAMP = 1640995200L;
  /**
   * 序列号的位数
   */
  private static final int COUNT_BITS = 32;
​
  private StringRedisTemplate stringRedisTemplate;
​
  public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
    this.stringRedisTemplate = stringRedisTemplate;
   }
​
  public long nextId(String keyPrefix) {
    // 1.生成时刻戳
    LocalDateTime now = LocalDateTime.now();
    long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
    long timestamp = nowSecond - BEGIN_TIMESTAMP;
​
    // 2.生成序列号
    // 2.1.获取当时日期，精确到天
    String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
    // 2.2.自增加
    // 每天一个key
    long count = stringRedisTemplate.opsForValue()
                                   .increment("icr:" + keyPrefix + ":" + date);
​
    // 3.拼接并回来
    return timestamp << COUNT_BITS | count;
   }
}

Redis自增ID战略：

1. 每天一个key，方便统计订单量
2. ID构造是 时刻戳 + 计数器

扩展

大局仅有ID生成战略：

1. UUID
2. Redis自增（需求额外拼接）
3. snowflake算法
4. 数据库自增

超卖问题的发生

处理方案

超卖问题是典型的多线程安全问题，针对这一问题的常见处理方案便是加锁：

锁有两种：

一，失望锁： 认为线程安全问题一定会发生，因此在操作数据之前先获取锁，确保线程串行履行。例如Synchronized、Lock都归于失望锁；

二，达观锁： 认为线程安全问题不一定会发生，因此不加锁，只是在更新数据时去判别有没有其它线程对数据做了修正。

假定没有修正则认为是安全的，自己才更新数据。 假定现已被其它线程修正阐明发生了安全问题，此刻能够重试或异常。

达观锁的两种完结

下面介绍达观锁的两种完结：

第一种，添加版本号：

每扣减一次就更改一下版本号，每次进行扣减之前需求查询一下版本号，只有在扣减时的版本号和之前的版本号相一起，才进行扣减。

第二种，CAS法

由于每扣减一次，库存量都会发生改动的，所以咱们完全能够用库存量来做标志，标志当时库存量是否被其他线程更改正（在这种情况下，库存量的功用和版本号类似）

下面给出 CAS 法扣除库存时，针对超卖问题的处理方案：

  // 扣减库存
  boolean success = seckillVoucherService.update()
           .setSql("stock = stock - 1") // set stock = stock - 1
           .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
           .update();

请注意上述的 CAS 判别有所优化了的，并不是判别刚查询的库存和扣除时的库存是否持平，而是判别当时库存是否大于 0。

由于 判别刚查询的库存和扣除时的库存是否持平会出现问题：假定多个线程都判别到不持平了，那它们都停止了扣减，这时候就会出现没办法买完了。

而 判别当时库存是否大于 0，则能够很好地处理上述问题！

一人一单的需求

一般来说秒杀的产品都是优惠力度很大的，所以或许存在一种需求——渠道只允许一个用户购买一个产品。

关于秒杀场景下的这种需求，咱们应该怎样去规划呢？

很显着，咱们需求在履行扣除库存的操作之前，先去查查数据库是否现已有了该用户的订单了；假定有了，阐明该用户现已下单过了，不能再购买；假定没有，则履行扣除操作并生成订单。

// 查询订单
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
// 判别是否存在
if (count > 0) {
  // 用户现已购买过了
  return Result.fail("用户现已购买过一次！");
}
​
// 扣减库存
boolean success = seckillVoucherService.update()
     .setSql("stock = stock - 1") // set stock = stock - 1
     .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
     .update();

并发安全问题

由于上述的完结是分成两步的：

1. 判别当时用户在数据库中并没有订单
2. 履行扣除操作，并生成订单

也正由于是分成了两步，所以才引发了线程安全问题： 能够是同一个用户的多个恳求线程都一起判别没有订单，后续则咱们都履行了扣除操作。

要处理这个问题，也很简单，只要让这两步串行履行即可，也便是加锁！

在办法头上加 synchronized

很明显这种会锁住整个办法，锁的范围太大了，并且会对一切恳求线程作出约束；而咱们的需求只是同一个用户的恳求线程串行就能够了；明显有些大材小用了~

@Transactional
public synchronized Result createVoucherOrder(Long voucherId) {
  // 一人一单
  Long userId = UserHolder.getUser().getId
   // 查询订单
   int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
   // 判别是否存在
   if (count > 0) {
     // 用户现已购买过了
     return Result.fail("用户现已购买过一次！");
  
   // 扣减库存
   boolean success = seckillVoucherService.update()
       .setSql("stock = stock - 1") // set stock = stock - 1
       .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0
       .update();
   if (!success) {
     // 扣减失败
     return Result.fail("库存不足！");
  
   // 创立订单
   VoucherOrder voucherOrder = new VoucherOrder();
   .....
   return Result.ok(orderId);
}

锁住同一用户 id 的 String 目标

@Transactional
public Result createVoucherOrder(Long voucherId) {
  // 一人一单
  Long userId = UserHolder.getUser().getId
  
  // 锁住同一用户 id 的 String 目标
  synchronized (userId.toString().intern()) {
      // 查询订单
      int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
      // 判别是否存在
       ......
      
      // 扣减库存
     ......
   
      // 创立订单
      ......
   }
   return Result.ok(orderId);
}

上述办法开启了事务，可是synchronized (userId.toString().intern())锁住的却不是整个办法（先开释锁，再提交事务，写入订单），那就存在一个问题——假定一个线程的事务还没提交（也便是还没写入订单），这时候其他线程来了却能够获得锁，它判别数据库中订单为0 ，又能够再次创立订单。。。。

为了处理这个问题，咱们需求先提交事务，再开释锁：

 // 锁住同一用户 id 的 String 目标
 synchronized (userId.toString().intern()) {
   ......
    createVoucherOrder(voucherId);
   ......
 }
​
@Transactional
public Result createVoucherOrder(Long voucherId) {
  // 一人一单
  Long userId = UserHolder.getUser().getId
  
  
      // 查询订单
      int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
      // 判别是否存在
       ......
      
      // 扣减库存
     ......
   
      // 创立订单
      ......
   
   return Result.ok(orderId);
}

集群形式下的并发安全问题

刚刚评论的那些都默许是单机结点的，可是现在假定放在了集群形式下的话就会出现一下问题。

刚刚的加锁现已处理了单机节点下的线程安全问题，可是却不能处理集群下多节点的线程安全问题：

由于 synchronized 锁的是对应 JVM 内的锁监视器，可是不同的结点有不同的 JVM，不同的 JVM 又有不同的锁监视器，所以刚刚的规划在集群形式下锁住的其实仍是不同的目标，即无法处理线程安全问题。

知道问题发生的原因，咱们应该很快就想到了处理办法了：

已然是由于集群导致了锁不同，那咱们就从头规划一下，让他们都运用同一把锁即可！

分布式锁

分布式锁：满意分布式体系或集群形式下多进程可见并且互斥的锁。

分布式锁的完结

分布式锁的核心是完结多进程之间互斥，而满意这一点的方式有许多，常见的有三种：

MySQL	Redis	Zookeeper
互斥	使用mysql本身的互斥锁机制	使用setnx这样的互斥指令	使用节点的仅有性和有序性完结互斥
高可用	好	好	好
高功用	一般	好	一般
安全性	断开衔接，主动开释锁	使用锁超时时刻，到期开释	暂时节点，断开衔接主动开释

根据 Redis 的分布式锁

用 Redis 完结分布式锁，首要应用到的是 SETNX key value指令（假定不存在，则设置）

首要要完结两个功用：

1. 获取锁（设置一个 key）
2. 开释锁 （删去 key）

基本思想是履行了 SETNX指令的线程获得锁，在完结操作后，需求删去 key，开释锁。

加锁：

@Override
public boolean tryLock(long timeoutSec) {
  // 获取线程标明
  String threadId = ID_PREFIX + Thread.currentThread().getId();
  // 获取锁
  Boolean success = stringRedisTemplate.opsForValue()
       .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
  return Boolean.TRUE.equals(success);
}

开释锁：

@Override
public void unlock() {
  // 获取线程标明
  String threadId = ID_PREFIX + Thread.currentThread().getId();
  // 获取锁中的标明
  String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
  // 开释锁
  stringRedisTemplate.delete(KEY_PREFIX + name);
}

可是这儿会存在一个隐患——假定该线程发生堵塞（或者其他问题），一直不开释锁（删去 key）这可怎样办？

为了处理这个问题，咱们需求为 key 规划一个超时时刻，让它超时失效；可是这个超时时刻的长短却不好确定：

1. 设置过短，会导致其他线程提早获得锁，引发线程安全问题
2. 设置过长，线程需求额外等候

锁的误删

超时时刻是一个十分不好把握的东西，由于事务线程的堵塞时刻是不可预估的，在极点情况下，它总能堵塞到 lock 超时失效，正如上图中的线程1，锁超时开释了，导致线程2也进来了，这时候 lock 是 线程2的锁了（key 相同，value不同，value一般是线程仅有标识）；假定这时候，线程1突然不堵塞了，它要开释锁，假定按照刚刚的代码逻辑的话，它会开释掉线程2的锁；线程2的锁被开释掉之后，又会导致其他线程进来（线程3），如此往复。。。

为了处理这个问题，需求在开释锁时多加一个判别，每个线程只开释自己的锁，不能开释别人的锁！

开释锁

@Override
public void unlock() {
  // 获取线程标明
  String threadId = ID_PREFIX + Thread.currentThread().getId();
  // 获取锁中的标明
  String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
  
  // 判别标明是否共同
  if(threadId.equals(id)) {
    // 开释锁
    stringRedisTemplate.delete(KEY_PREFIX + name);
   }
}

原子性问题

刚刚咱们谈论的开释锁的逻辑：

1. 判别当时锁是当时线程的锁
2. 当时线程开释锁

能够看到开释锁是分两步完结的，假定你是对并发比较有感觉的话，应该一会儿就知道这儿会存在问题了。

分步履行，并发问题！

假定 线程1 现已判别当时锁是它的锁了，正预备开释锁，可偏偏这时候它堵塞了（或许是 FULL GC 引起的），锁超时失效，线程2来加锁，这时候锁是线程2的了；可是假定线程1这时候醒过来，由于它现已履行了过程1了的，所以这时候它会直接直接过程2，开释锁（可是此刻的锁不是线程1的了）

其实这便是一个原子性的问题，刚刚开释锁的两步应该是原子的，不可分的！

要使得其满意原子性，则需求在 Redis 中运用 Lua 脚本了。

引入 Lua 脚本保持原子性

lua 脚本：

-- 比较线程标明与锁中的标明是否共同
if(redis.call('get', KEYS[1]) == ARGV[1]) then
  -- 开释锁 del key
  return redis.call('del', KEYS[1])
end
return 0

Java 中调用履行：

public class SimpleRedisLock implements ILock {
​
  private String name;
  private StringRedisTemplate stringRedisTemplate;
​
  public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
    this.name = name;
    this.stringRedisTemplate = stringRedisTemplate;
   }
​
  private static final String KEY_PREFIX = "lock:";
  private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
  private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
  static {
    UNLOCK_SCRIPT = new DefaultRedisScript<>();
    UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    UNLOCK_SCRIPT.setResultType(Long.class);
   }
​
  @Override
  public boolean tryLock(long timeoutSec) {
    // 获取线程标明
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    // 获取锁
    Boolean success = stringRedisTemplate.opsForValue()
         .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
    return Boolean.TRUE.equals(success);
   }
​
  @Override
  public void unlock() {
    // 调用lua脚本
    stringRedisTemplate.execute(
        UNLOCK_SCRIPT,
        Collections.singletonList(KEY_PREFIX + name),
        ID_PREFIX + Thread.currentThread().getId());
   }
}
​

到了目前为止，咱们规划的 Redis 分布式锁现已是生产可用的，相对完善的分布式锁了。

总结

这一次咱们从秒杀场景的事务需求动身，一步步地使用 Redis 规划出一种生产可用的分布式锁：

完结思路：

1. 使用set nx ex获取锁，并设置过期时刻，保存线程标明
2. 开释锁时先判别线程标明是否与自己共同，共同则删去锁 (Lua 脚本确保原子性)

有哪些特性？

1. 使用set nx满意互斥性
2. 使用set ex确保毛病时锁依然能开释，防止死锁，进步安全性
3. 使用Redis集群确保高可用和高并发特性

目前还有待完善的点：

1. 不可重入，同一个线程无法屡次获取同一把锁
2. 不可重试，获取锁只尝试一次就回来false，没有重试机制
3. 超时开释，锁超时开释虽然能够防止死锁，但假定是事务履行耗时较长，也会导致锁开释，存在安全隐患（虽然现已处理了误删问题，可是仍然或许存在不知道问题）
4. 主从共同性，假定Redis提供了主从集群，主从同步存在推迟，当主宕机时，在主节点中的锁数据并没有及时同步到从节点中，则会导致其他线程也能获得锁，引发线程安全问题（推迟时刻是在毫秒以下的，所以这种情况概率极低）