随着互联网的疾速倒退，商品秒杀的场景咱们并不少见；秒杀是一种供不应求的，高并发的场景，它外面蕴含了很多技术点，把握了其中的技术点，虽不肯定能让你面试立马胜利，但那也必是一个闪耀的点！
前言
假如咱们当初有一个商城零碎，外面上线了一个商品秒杀的模块，那么这个模块咱们要怎么设计呢？
秒杀模块又会有哪些不同的需要呢？
全局惟一 ID
商品秒杀实质上其实还是商品购买，所以咱们须要筹备一张订单表来记录对应的秒杀订单。
这里就波及到了一个订单 id 的问题了，咱们是否能够像其余表一样应用数据库本身的自增 id 呢？
数据库自增 id 的毛病
订单表如果应用数据库自增 id，则会存在一些问题：
- id 的法则太显著了 因为咱们的订单 id 是须要回显给用户查看的，如果是 id 法则太显著的话，会裸露一些信息，比方第一天下单的 id = 10，第二天下单的 id = 11，这就阐明这两单之间基本没有其余用户下单
- 受单表数据量的限度 在高并发场景下，产生上百万个订单都是有可能的，而咱们都晓得 MySQL 的单张表基本不可能包容这么多数据（性能等起因的限度）；如果是将单表拆成多表，还是用数据库自增 id 的话，就存在了订单 id 反复的状况了，很显然这是业务不容许的。
基于以上两个问题，咱们能够晓得订单表的 id 须要是一个全局惟一的 ID，而且还不能存在显著的法则。
全局 ID 生成器
全局 ID 生成器，是一种在分布式系统下用来生成全局惟一 ID 的工具，个别要满足下列个性：
这里咱们思考一下是否能够用 Redis 中的自增计数来作为全局 id 生成器呢？
能不能次要是看它是否满足上述 5 个条件：
- 唯一性，每个订单都是来 Redis 这里生成订单 id 的，所以唯一性能够保障
- 高可用，Redis 能够由主从、集群等模式保障可用性
- 高性能，Redis 是基于内存的，原本就是以性能自称的
- 递增性，increment 原本就是递增的
- 安全性。。。这个就麻烦了点了，因为 Redis 的 increment 也是递增的，法则太显著了。。。
综上，Redis 的 increment 并不能满足安全性，所以咱们不能单纯应用它来做全局 id 生成器。
然而——
咱们能够应用它，再和其余货色拼接起来~
举个栗子：
ID 的组成部分：
- 符号位：1bit，永远为 0
- 工夫戳：31bit，以秒为单位，能够应用 69 年
- 序列号：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 策略：
- 每天一个 key，不便统计订单量
- ID 结构是 工夫戳 + 计数器
扩大
全局惟一 ID 生成策略：
- UUID
Redis 自增（须要额定拼接）snowflake 算法- 数据库自增
超卖问题的产生
解决方案
超卖问题是典型的多线程平安问题，针对这一问题的常见解决方案就是加锁：
锁有两种：
一，乐观锁： 认为线程平安问题肯定会产生，因而在操作数据之前先获取锁，确保线程串行执行。例如 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();并发平安问题
因为上述的实现是分成两步的：
- 判断以后用户在数据库中并没有订单
- 执行扣除操作，并生成订单
也正因为是分成了两步，所以才引发了线程平安问题： 能够是同一个用户的多个申请线程都同时判断没有订单，后续则大家都执行了扣除操作。
要解决这个问题，也很简略，只有让这两步串行执行即可，也就是加锁！
在办法头上加 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命令（如果不存在，则设置）
次要要实现两个性能：
- 获取锁（设置一个 key）
- 开释锁（删除 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 设计一个超时工夫，让它超时生效；然而这个超时工夫的长短却不好确定：
- 设置过短，会导致其余线程提前取得锁，引发线程平安问题
- 设置过长，线程须要额定期待
锁的误删
超时工夫是一个十分不好把握的货色，因为业务线程的阻塞工夫是不可预估的，在极其状况下，它总能阻塞到 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 曾经判断以后锁是它的锁了，正筹备开释锁，可偏偏这时候它阻塞了（可能是 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 0Java 中调用执行：
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 设计出一种生产可用的分布式锁：
实现思路：
- 利用
set nx ex获取锁，并设置过期工夫，保留线程标示- 开释锁时先判断线程标示是否与本人统一，统一则删除锁 (Lua 脚本保障原子性)
有哪些个性？
- 利用
set nx满足互斥性- 利用
set ex保障故障时锁仍然能开释，防止死锁，进步安全性- 利用
Redis集群保障高可用和高并发个性
目前还有待欠缺的点：
- 不可重入，同一个线程无奈屡次获取同一把锁
- 不可重试，获取锁只尝试一次就返回 false，没有重试机制
- 超时开释，锁超时开释尽管能够防止死锁，但如果是业务执行耗时较长，也会导致锁开释，存在安全隐患（尽管曾经解决了误删问题，然而依然可能存在未知问题）
- 主从一致性，如果 Redis 提供了主从集群，主从同步存在提早，当主宕机时，在主节点中的锁数据并没有及时同步到从节点中，则会导致其余线程也能取得锁，引发线程平安问题（延迟时间是在毫秒以下的，所以这种状况概率极低）