秒杀场景下的业务梳理——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.自增长// 每天一个keylong 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 keyreturn 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);}
​@Overridepublic 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);}
​@Overridepublic 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提供了主从集群，主从同步存在延迟，当主宕机时，在主节点中的锁数据并没有及时同步到从节点中，则会导致其他线程也能获得锁，引发线程安全问题（延迟时间是在毫秒以下的，所以这种情况概率极低）