一、优惠券秒杀

1.1 全局唯一ID

1.2 Redis实现全局唯一ID

1.3 添加优惠卷

1.4 实现秒杀下单

1.5 库存超卖问题分析

1.6 代码实现乐观锁解决超卖问题

1.7 优惠券秒杀-一人一单

1.8 集群环境下的并发问题

二、分布式锁

2.1 基本原理和实现方式对比

2.2 Redis分布式锁的实现核心思路

2.3 实现分布式锁版本一

2.4 Redis分布式锁误删情况说明

2.5 解决Redis分布式锁误删问题

2.6 分布式锁的原子性问题

2.7 Lua脚本解决多条命令原子性问题

2.8 利用Java代码调用Lua脚本改造分布式锁

三、分布式锁-redission

3.1 分布式锁-redission功能介绍

3.2 分布式锁-Redission快速入门

3.3 分布式锁-redission可重入锁原理

3.4 分布式锁-redission锁重试和WatchDog机制

3.5 分布式锁-redission锁的MutiLock原理

一、优惠券秒杀

1.1 全局唯一ID

每个店铺都可以发布优惠券：

当用户抢购时，就会生成订单并保存到tb_voucher_order这张表中，而订单表如果使用数据库自增ID就存在一些问题：

id的规律性太明显（会给客户暴露一些信息）
受单表数据量的限制（如果订单量过大，单表无法保存如此多的时候，只能分表，mysql它的自增长是每张表各自计算自己的自增长，那么必然会有重复id的问题）

场景分析一：如果我们的id具有太明显的规则，用户或者说商业对手很容易猜测出来我们的一些敏感信息，比如商城在一天时间内，卖出了多少单，这明显不合适。

场景分析二：随着我们商城规模越来越大，mysql的单表的容量不宜超过500W，数据量过大之后，我们要进行拆库拆表，但拆分表了之后，他们从逻辑上讲他们是同一张表，所以他们的id是不能一样的，于是乎我们需要保证id的唯一性。

全局ID生成器：是一种在分布式系统下用来生成全局唯一ID的工具，一般要满足下列特性：

解决：Redis就可以用作全局ID生成器使用。

唯一性：使用incr命令进行自增，因为Redis是独立于数据库之外的，不管你有几张表，Redis只有一个，都是唯一的。
高可用：Redis的集群方案、主从方案、哨兵模式，都可以确保它的高可用。
高性能：这个不用多说，我们Redis就是以高性能著称的。
递增性：使用incr命令进行自增。
安全性：虽然可以用incr自增，但是这样还是跟数据库自增一样，从1开始加，那这样就太容易被人猜出规律，所以我们可以拼接一些其它信息。

ID的组成部分：符号位：1bit，永远为0

时间戳：31bit，以秒为单位，可以使用69年

序列号：32bit，秒内的计数器，支持每秒产生2^32个不同ID

1.2 Redis实现全局唯一ID

@Component
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.自增长long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);// 3.拼接并返回return timestamp << COUNT_BITS | count;}
}

1.3 添加优惠卷

每个店铺都可以发布优惠券，分为平价券和特价券。平价券可以任意购买，而特价券需要秒杀抢购：

tb_voucher：优惠券的基本信息，优惠金额、使用规则等
tb_seckill_voucher：优惠券的库存、开始抢购时间，结束抢购时间。特价优惠券才需要填写这些信息

平价卷由于优惠力度并不是很大，所以是可以任意领取

而代金券由于优惠力度大，所以像第二种卷，就得限制数量，从表结构上也能看出，特价卷除了具有优惠卷的基本信息以外，还具有库存，抢购时间，结束时间等等字段

新增普通卷代码：VoucherController

/*** 新增普通券* @param voucher 优惠券信息* @return 优惠券id*/
@PostMapping
public Result addVoucher(@RequestBody Voucher voucher) {voucherService.save(voucher);return Result.ok(voucher.getId());
}

新增秒杀卷代码：VoucherController

/*** 新增秒杀券* @param voucher 优惠券信息，包含秒杀信息* @return 优惠券id*/
@PostMapping("seckill")
public Result addSeckillVoucher(@RequestBody Voucher voucher) {voucherService.addSeckillVoucher(voucher);return Result.ok(voucher.getId());
}

VoucherServiceImpl

@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {// 保存优惠券save(voucher);// 保存秒杀信息SeckillVoucher seckillVoucher = new SeckillVoucher();seckillVoucher.setVoucherId(voucher.getId());seckillVoucher.setStock(voucher.getStock());seckillVoucher.setBeginTime(voucher.getBeginTime());seckillVoucher.setEndTime(voucher.getEndTime());seckillVoucherService.save(seckillVoucher);// 保存秒杀库存到Redis中stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
}

1.4 实现秒杀下单

下单核心思路：当我们点击抢购时，会触发右侧的请求，我们只需要编写对应的controller即可

秒杀下单应该思考的内容：

下单时需要判断两点：

秒杀是否开始或结束，如果尚未开始或已经结束则无法下单
库存是否充足，不足则无法下单

下单核心逻辑分析：

当用户开始进行下单，我们应当去查询优惠卷信息，查询到优惠卷信息，判断是否满足秒杀条件

比如时间是否充足，如果时间充足，则进一步判断库存是否足够，如果两者都满足，则扣减库存，创建订单，然后返回订单id，如果有一个条件不满足则直接结束。

VoucherOrderServiceImpl

@Override
@Transactional
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}//5，扣减库存boolean success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).update();if (!success) {//扣减库存return Result.fail("库存不足！");}//6.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 6.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 6.2.用户idLong userId = UserHolder.getUser().getId();voucherOrder.setUserId(userId);// 6.3.代金券idvoucherOrder.setVoucherId(voucherId);voucherOrderService.save(voucherOrder);return Result.ok(orderId);}

1.5 库存超卖问题分析

有关超卖问题分析：在我们原有代码中是这么写的

if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");
}
//5，扣减库存
boolean success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).update();
if (!success) {//扣减库存return Result.fail("库存不足！");
}

假设线程1过来查询库存，判断出来库存大于1，正准备去扣减库存，但是还没有来得及去扣减，此时线程2过来，线程2也去查询库存，发现这个数量一定也大于1，那么这两个线程都会去扣减库存，最终多个线程相当于一起去扣减库存，此时就会出现库存的超卖问题。

超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是加锁：而对于加锁，我们通常有两种解决方案：见下图：

悲观锁：

悲观锁可以实现对于数据的串行化执行，比如syn，和lock都是悲观锁的代表，同时，悲观锁中又可以再细分为公平锁，非公平锁，可重入锁，等等。

乐观锁：

乐观锁：会有一个版本号，每次操作数据会对版本号+1，再提交回数据时，会去校验是否比之前的版本大1 ，如果大1 ，则进行操作成功，这套机制的核心逻辑在于，如果在操作过程中，版本号只比原来大1 ，那么就意味着操作过程中没有人对他进行过修改，他的操作就是安全的，如果不大1，则数据被修改过，当然乐观锁还有一些变种的处理方式比如cas。

乐观锁的典型代表：就是cas，利用cas进行无锁化机制加锁，var5 是操作前读取的内存值，while中的var1+var2 是预估值，如果预估值 == 内存值，则代表中间没有被人修改过，此时就将新值去替换内存值。

其中do while 是为了在操作失败时，再次进行自旋操作，即把之前的逻辑再操作一次。

白话讲解乐观锁：

我们现在库中优惠券库存只剩下1个，版本号现在是1，当线程1开始访问，先查出库存和版本号（都是1），这时线程2抢到cpu执行权，也查出库存和版本号（都是1），这时线程1又抢到执行权，使用update order set stock = stock - 1,version = version+1 where id = 10 and version = 1;执行成功，线程2又接着执行，update order set stock = stock - 1,version = version+1 where id = 10 and version = 1;执行失败，因为此时version已经等于2了，无法通过where version = 1这个条件。

上面版本号的方式确实可以实现乐观锁，但是还有个问题，我们何必多次一举加个版本号呢，我直接用库存数不就得了吗？update order set stock = stock - 1 where id = 10 and stock = 1; 这就是CAS。

1.6 代码实现乐观锁解决超卖问题

修改代码方案一、

VoucherOrderServiceImpl 在扣减库存时，改为：

boolean success = seckillVoucherService.update().setSql("stock= stock - 1") //set stock = stock -1.eq("voucher_id", voucherId).eq("stock",voucher.getStock()).update(); //where id = ？ and stock = ?

以上逻辑的核心含义是：只要我扣减库存时的库存和之前我查询到的库存是一样的，就意味着没有人在中间修改过库存，那么此时就是安全的，但是以上这种方式通过测试发现会有很多失败的情况，失败的原因在于：在使用乐观锁过程中假设100个线程同时都拿到了100的库存，然后大家一起去进行扣减，但是100个人中只有1个人能扣减成功，其他的人在处理时，他们在扣减时，库存已经被修改过了，所以此时其他线程都会失败。

修改代码方案二、

之前的方式要修改前后都保持一致，但是这样我们分析过，成功的概率太低，所以我们的乐观锁需要变一下，改成stock大于0 即可。

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

1.7 优惠券秒杀-一人一单

需求：修改秒杀业务，要求同一个优惠券，一个用户只能下一单

现在的问题在于：

优惠卷是为了引流，但是目前的情况是，一个人可以无限制的抢这个优惠卷，所以我们应当增加一层逻辑，让一个用户只能下一个单，而不是让一个用户下多个单

具体操作逻辑如下：比如时间是否充足，如果时间充足，则进一步判断库存是否足够，然后再根据优惠卷id和用户id查询是否已经下过这个订单，如果下过这个订单，则不再下单，否则进行下单

VoucherOrderServiceImpl

初步代码：增加一人一单逻辑

@Override
@Transactional
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}// 5.一人一单逻辑// 5.1.用户idLong userId = UserHolder.getUser().getId();int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}//6，扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();if (!success) {//扣减库存return Result.fail("库存不足！");}//7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);voucherOrderService.save(voucherOrder);return Result.ok(orderId);
}

存在问题：现在的问题还是和之前一样，并发过来，查询数据库，都不存在订单，所以我们还是需要加锁，但是乐观锁比较适合更新数据，而现在是插入数据，所以我们需要使用悲观锁操作。

比如有的黄牛弄100台电脑，同时登录一个账号，这时一起抢券，咱这段代码就出问题了，比如这100个线程同时查出本账号没有下过订单，那这100个线程都会绕过判断，直接创建100单，所以就导致了1个账号抢到100个券的情况，这就是为什么现在大部分app都做账号限制登录设备，最多只能同时登录1个设备。

注意：在这里提到了非常多的问题，我们需要慢慢的来思考，首先我们的初始方案是封装了一个createVoucherOrder方法，同时为了确保他线程安全，在方法上添加了一把synchronized锁。

@Override
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}return createVoucherOrder(voucherId);
}@Transactional
public synchronized Result createVoucherOrder(Long voucherId){// 5.一人一单逻辑// 5.1.用户idLong userId = UserHolder.getUser().getId();int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}//6，扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).update();if (!success) {//扣减库存return Result.fail("库存不足！");}//7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);voucherOrderService.save(voucherOrder);return Result.ok(orderId);
}

但是这样添加锁，锁的粒度太粗了，因为我们使用的是同步方法方式加锁，那锁的是this本身，就导致哪怕是不同账号也会被锁住，我张三获得锁，只锁住张三账号即可，不用连李四也锁住啊。在使用锁过程中，控制锁粒度是一个非常重要的事情，因为如果锁的粒度太大，会导致每个线程进来都会锁住，所以我们需要去控制锁的粒度，以下这段代码需要修改为：intern() 这个方法是从常量池中拿到数据，如果我们直接使用userId.toString() 他拿到的对象实际上是不同的对象，new出来的对象，我们使用锁必须保证锁必须是同一把，所以我们需要使用intern()方法

@Transactional
public  Result createVoucherOrder(Long voucherId) {Long userId = UserHolder.getUser().getId();synchronized(userId.toString().intern()){// 5.1.查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}// 6.扣减库存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("库存不足！");}// 7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 7.2.用户idvoucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);voucherOrderService.save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);}
}

但是以上代码还是存在问题，问题的原因在于当前方法被spring的事务控制，如果你在方法内部加锁，可能会导致当前方法事务还没有提交，但是锁已经释放也会导致问题，所以我们选择将当前方法整体包裹起来，确保事务不会出现问题：如下：

在seckillVoucher 方法中，添加以下逻辑，这样就能保证事务的特性，同时也控制了锁的粒度

@Override
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}Long userId = UserHolder.getUser().getId();synchronized(userId.toString().intern()) {return this.createVoucherOrder(voucherId);}
}@Transactional
public  Result createVoucherOrder(Long voucherId) {Long userId = UserHolder.getUser().getId();// 5.1.查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2.判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("用户已经购买过一次！");}// 6.扣减库存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("库存不足！");}// 7.创建订单VoucherOrder voucherOrder = new VoucherOrder();// 7.1.订单idlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);// 7.2.用户idvoucherOrder.setUserId(userId);// 7.3.代金券idvoucherOrder.setVoucherId(voucherId);voucherOrderService.save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);
}

但是以上做法依然有问题，因为你调用的方法，其实是this.的方式调用的，this是当前对象，而不是代理对象，我们都知道，如果事务想要生效，是spring对当前这个类做了动态代理，所以这个地方，我们需要获得原始的事务对象，来操作事务

Long userId = UserHolder.getUser().getId();
synchronized(userId.toString().intern()) {IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);
}

1.8 集群环境下的并发问题

通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。

1、我们将服务启动两份，端口分别为8081和8082：

如何开启两个服务呢？

2、然后修改nginx的conf目录下的nginx.conf文件，配置反向代理和负载均衡：

有关锁失效原因分析

由于现在我们部署了多个tomcat，每个tomcat都有一个属于自己的jvm，那么假设在服务器A的tomcat内部，有两个线程，这两个线程由于使用的是同一份代码，那么他们的锁对象是同一个，是可以实现互斥的，但是如果现在是服务器B的tomcat内部，又有两个线程，但是他们的锁对象写的虽然和服务器A一样，但是锁对象却不是同一个，所以线程3和线程4可以实现互斥，但是却无法和线程1和线程2实现互斥，这就是集群环境下，syn锁失效的原因，在这种情况下，我们就需要使用分布式锁来解决这个问题。

二、分布式锁

2.1 基本原理和实现方式对比

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

分布式锁的核心思想就是让大家都使用同一把锁，只要大家使用的是同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路。

那么分布式锁他应该满足一些什么样的条件呢？

可见性：多个线程都能看到相同的结果，注意：这个地方说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思。

互斥：互斥是分布式锁的最基本的条件，使得程序串行执行。

高可用：程序不易崩溃，时时刻刻都保证较高的可用性。

高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能。

安全性：安全也是程序中必不可少的一环。

常见的分布式锁有三种

Mysql：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少见。

Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用redis或者zookeeper作为分布式锁，利用setnx这个方法，如果插入key成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁。

Zookeeper：zookeeper也是企业级开发中较好的一个实现分布式锁的方案，由于本套视频并不讲解zookeeper的原理和分布式锁的实现，所以不过多阐述。

2.2 Redis分布式锁的实现核心思路

实现分布式锁时需要实现的两个基本方法：

获取锁：
- 互斥：确保只能有一个线程获取锁
- 非阻塞：尝试一次，成功返回true，失败返回false
释放锁：
- 手动释放
- 超时释放：获取锁时添加一个超时时间

核心思路：

我们利用redis的setnx方法，当有多个线程进入时，我们就利用该方法，第一个线程进入时，redis 中就有这个key 了，返回了1，如果结果是1，则表示他抢到了锁，那么他去执行业务，然后再删除锁，退出锁逻辑，没有抢到锁的哥们，等待一定时间后重试即可。

插曲：我们都知道应该先用setnx命令获取分布式锁，然后使用expire命令为锁设置时间，防止redis宕机无法释放锁，那么有没有这样一种情况，那就是我用setnx命令获取分布式锁，还没等执行expire命令呢，redis就宕机了？那么如何解决呢？

解决方式：两条命令合二为一变成原子操作，使用 set lock thread1 ex 10 nx；其中10是10秒。

2.3 实现分布式锁版本一

基本锁接口

public interface ILock {/*** 尝试获取锁* @param timeoutSec 锁持有的超时时间，过期后自动释放* @return true 代表锁获取成功；false 代表锁获取失败*/boolean tryLock(Long timeoutSec);/*** 释放锁*/void unlock();}

实现类

利用setnx方法进行加锁，同时增加过期时间，防止死锁，此方法可以保证加锁和增加过期时间具有原子性。

public class SimpleRedisLock implements ILock{private String name;private StringRedisTemplate stringRedisTemplate;private static final String KEY_PREFIX = "lock:";public SimpleRedisLock(String name,StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}@Overridepublic boolean tryLock(Long timeoutSec) {// 获取线程标示long threadId = Thread.currentThread().getId();// 获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(success);}@Overridepublic void unlock() {//通过del删除锁stringRedisTemplate.delete(KEY_PREFIX + name);}
}

修改一人一单的业务代码

@Override
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}Long userId = UserHolder.getUser().getId();//创建锁对象SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);//获取锁boolean isLock = lock.tryLock(1200L);//判断获取锁是否成功if (!isLock) {//获取锁失败，返回错误或重试return Result.fail("不允许重复下单");}try {IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);}finally {//释放锁lock.unlock();}
}

2.4 Redis分布式锁误删情况说明

逻辑说明：

持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明。

解决方案：解决方案就是在每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果不属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁。

白话解释：单位卫生间只有一个坑位，A和B这时都肚子痛，但A快了一步抢到了坑位（锁），开始蹲坑，B就只能在外面等着，单位卫生间比较智能，为了防止摸鱼，换了只能门锁，只要蹲坑时间超过10分钟（锁过期时间10分钟），自动打开门锁，但由于A处理时间较长，早已超过10分钟，这时锁已经被打开了（释放），虽然已经打开了，但A还在蹲着，这时B因为拿到了锁，也进到坑位里执行他的业务，这时A完事了，开始执行del lock操作，但其实他的锁早就被删除释放了，这时他del lock操作删除的是B的锁，这就是误删操作。而解决办法就是，把自己的锁打上标识（线程ID，也可以是UUID），等自己做del lock操作前，先判断一下是不是自己的锁，如果不是自己的锁，则不进行锁的删除。

2.5 解决Redis分布式锁误删问题

需求：修改之前的分布式锁实现，满足：在获取锁时存入线程标示（可以用UUID表示）
在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致

- 如果一致则释放锁
- 如果不一致则不释放锁

核心逻辑：在存入锁时，放入自己线程的标识，在删除锁时，判断当前这把锁的标识是不是自己存入的，如果是，则进行删除，如果不是，则不进行删除。

public class SimpleRedisLock implements ILock{private String name;private StringRedisTemplate stringRedisTemplate;private static final String KEY_PREFIX = "lock:";private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";public SimpleRedisLock(String name,StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}@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() {//获取线程标识String threadId = ID_PREFIX + Thread.currentThread().getId();//获取锁中的标识String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);if (threadId.equals(id)) {//通过del删除锁stringRedisTemplate.delete(KEY_PREFIX + name);}}
}

有关代码实操说明：

在我们修改完此处代码后，我们重启工程，然后启动两个线程，第一个线程持有锁后，手动释放锁，第二个线程此时进入到锁内部，再放行第一个线程，此时第一个线程由于锁的value值并非是自己，所以不能释放锁，也就无法删除别人的锁，此时第二个线程能够正确释放锁，通过这个案例初步说明我们解决了锁误删的问题。

2.6 分布式锁的原子性问题

更为极端的误删逻辑说明：

线程1现在持有锁之后，在执行业务逻辑过程中，他正准备删除锁，而且已经走到了条件判断的过程中，比如他已经拿到了当前这把锁确实是属于他自己的，正准备删除锁，但是此时他的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，我们要防止刚才的情况发生。

这块一旦阻塞，阻塞期间分布式锁还超时删除了，那这时线程2拿到锁执行业务，线程2正执行业务呢，这时线程1不阻塞了，开始执行del删除锁操作，那不还是把线程2的锁删了吗，所以还是有问题，所以我们必须确保判断锁标识的动作和释放锁的动作得成原子性。

2.7 Lua脚本解决多条命令原子性问题

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：Lua 教程 | 菜鸟教程，这里重点介绍Redis提供的调用函数，我们可以使用lua去操作redis，又能保证他的原子性，这样就可以实现拿锁比锁删锁是一个原子性动作了，作为Java程序员这一块并不作一个简单要求，并不需要大家过于精通，只需要知道他有什么作用即可。

这里重点介绍Redis提供的调用函数，语法如下：

redis.call('命令名称', 'key', '其它参数', ...)

例如，我们要执行set name jack，则脚本是这样：

# 执行 set name jack
redis.call('set', 'name', 'jack')

例如，我们要先执行set name Rose，再执行get name，则脚本如下：

# 先执行 set name jack
redis.call('set', 'name', 'Rose')
# 再执行 get name
local name = redis.call('get', 'name')
# 返回
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

例如，我们要执行 redis.call('set', 'name', 'jack') 这个脚本，语法如下：

如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

接下来我们来回一下我们释放锁的逻辑：

释放锁的业务流程是这样的

1、获取锁中的线程标示

2、判断是否与指定的标示（当前线程标示）一致

3、如果一致则释放锁（删除）

4、如果不一致则什么都不做

如果用Lua脚本来表示则是这样的：

最终我们操作redis的拿锁比锁删锁的lua脚本就会变成这样

-- 这里的 KEYS[1] 就是锁的key，这里的ARGV[1] 就是当前线程标示
-- 获取锁中的标示，判断是否与当前线程标示一致
if (redis.call('GET', KEYS[1]) == ARGV[1]) then-- 一致，则删除锁return redis.call('DEL', KEYS[1])
end
-- 不一致，则直接返回
return 0

2.8 利用Java代码调用Lua脚本改造分布式锁

我们的RedisTemplate中，可以利用execute方法去执行lua脚本，参数对应关系就如下图：

Java代码

private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}public void unlock() {// 调用lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),ID_PREFIX + Thread.currentThread().getId());
}
经过以上代码改造后，我们就能够实现 拿锁比锁删锁的原子性动作了~

小总结：

基于Redis的分布式锁实现思路：

利用set nx ex获取锁，并设置过期时间，保存线程标示
释放锁时先判断线程标示是否与自己一致，一致则删除锁
- 特性：
  - 利用set nx满足互斥性
  - 利用set ex保证故障时锁依然能释放，避免死锁，提高安全性
  - 利用Redis集群保证高可用和高并发特性

笔者总结：我们一路走来，利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题我们开始是利用删之前通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是我们没法保证拿锁比锁删锁是一个原子性的动作，最后通过lua表达式来解决这个问题

但是目前还剩下一个问题锁不住，什么是锁不住呢，你想一想，如果当过期时间到了之后，我们可以给他续期一下，比如续个30s，就好像是网吧上网，网费到了之后，然后说，来，网管，再给我来10块的，是不是后边的问题都不会发生了，那么续期问题怎么解决呢，可以依赖于我们接下来要学习redission啦

测试逻辑：

第一个线程进来，得到了锁，手动删除锁，模拟锁超时了，其他线程会执行lua来抢锁，当第一天线程利用lua删除锁时，lua能保证他不能删除他的锁，第二个线程删除锁时，利用lua同样可以保证不会删除别人的锁，同时还能保证原子性。

三、分布式锁-redission

3.1 分布式锁-redission功能介绍

基于setnx实现的分布式锁存在下面的问题：

重入问题：重入问题是指获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如HashTable这样的代码中，他的方法都是使用synchronized修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

不可重试：是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。

超时释放：我们在加锁时增加了过期时间，这样的我们可以防止死锁，但是如果卡顿的时间超长，虽然我们采用了lua表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患

主从一致性： 如果Redis提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

那么什么是Redission呢？

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

3.2 分布式锁-Redission快速入门

引入依赖：

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.6</version>
</dependency>

配置Redisson客户端：

@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient(){// 配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");// 创建RedissonClient对象return Redisson.create(config);}
}

改造一人一单代码：

@Resource
private RedissonClient redissonClient;@Override
public Result seckillVoucher(Long voucherId) {// 1.查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2.判断秒杀是否开始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始！");}// 3.判断秒杀是否已经结束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀已经结束！");}// 4.判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足！");}Long userId = UserHolder.getUser().getId();//创建锁对象 这个代码不用了，因为我们现在要使用分布式锁//SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);RLock lock = redissonClient.getLock("lock:order:" + userId);//获取锁对象boolean isLock = lock.tryLock();//加锁失败if (!isLock) {return Result.fail("不允许重复下单");}try {//获取代理对象(事务)IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//释放锁lock.unlock();}}

3.3 分布式锁-redission可重入锁原理

在Lock锁中，他是借助于底层的一个voaltile的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state=0，假如有人持有这把锁，那么state=1，如果持有这把锁的人再次持有这把锁，那么state就会+1 ，如果是对于synchronized而言，他在c语言代码中会有一个count，原理和state类似，也是重入一次就加一，释放一次就-1 ，直到减少成0 时，表示当前这把锁没有被人持有。

在redission中，我们的也支持支持可重入锁

在分布式锁中，他采用hash结构用来存储锁，其中大key表示表示这把锁是否存在，用小key表示当前这把锁被哪个线程持有，所以接下来我们一起分析一下当前的这个lua表达式

这个地方一共有3个参数

KEYS[1] ：锁名称

ARGV[1]：锁失效时间

ARGV[2]： id + ":" + threadId; 锁的小key

exists: 判断数据是否存在 name：是lock是否存在,如果==0，就表示当前这把锁不存在

redis.call('hset', KEYS[1], ARGV[2], 1);此时他就开始往redis里边去写数据，写成一个hash结构

Lock{

id + ":" + threadId : 1

}

如果当前这把锁存在，则第一个条件不满足，再判断

redis.call('hexists', KEYS[1], ARGV[2]) == 1

此时需要通过大key+小key判断当前这把锁是否是属于自己的，如果是自己的，则进行

redis.call('hincrby', KEYS[1], ARGV[2], 1)

将当前这个锁的value进行+1 ，redis.call('pexpire', KEYS[1], ARGV[1]); 然后再对其设置过期时间，如果以上两个条件都不满足，则表示当前这把锁抢锁失败，最后返回pttl，即为当前这把锁的失效时间

如果小伙帮们看了前边的源码，你会发现他会去判断当前这个方法的返回值是否为null，如果是null，则对应则前两个if对应的条件，退出抢锁逻辑，如果返回的不是null，即走了第三个分支，在源码处会进行while(true)的自旋抢锁。

"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hset', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +"return redis.call('pttl', KEYS[1]);"

3.4 分布式锁-redission锁重试和WatchDog机制

说明：由于课程中已经说明了有关tryLock的源码解析以及其看门狗原理，所以笔者在这里给大家分析lock()方法的源码解析，希望大家在学习过程中，能够掌握更多的知识。

抢锁过程中，获得当前线程，通过tryAcquire进行抢锁，该抢锁逻辑和之前逻辑相同。

1、先判断当前这把锁是否存在，如果不存在，插入一把锁，返回null。

2、判断当前这把锁是否是属于当前线程，如果是，则返回null。

所以如果返回是null，则代表着当前这哥们已经抢锁完毕，或者可重入完毕，但是如果以上两个条件都不满足，则进入到第三个条件，返回的是锁的失效时间，同学们可以自行往下翻一点点，你能发现有个while( true) 再次进行tryAcquire进行抢锁。

long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {return;
}

接下来会有一个条件分支，因为lock方法有重载方法，一个是带参数，一个是不带参数，如果带带参数传入的值是-1，如果传入参数，则leaseTime是他本身，所以如果传入了参数，此时leaseTime != -1 则会进去抢锁，抢锁的逻辑就是之前说的那三个逻辑。

if (leaseTime != -1) {return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}

如果是没有传入时间，则此时也会进行抢锁，而且抢锁时间是默认看门狗时间 commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()

ttlRemainingFuture.onComplete((ttlRemaining, e) 这句话相当于对以上抢锁进行了监听，也就是说当上边抢锁完毕后，此方法会被调用，具体调用的逻辑就是去后台开启一个线程，进行续约逻辑，也就是看门狗线程。

RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {if (e != null) {return;}// lock acquiredif (ttlRemaining == null) {scheduleExpirationRenewal(threadId);}
});
return ttlRemainingFuture;

此逻辑就是续约逻辑，注意看commandExecutor.getConnectionManager().newTimeout（）此方法。

Method( new TimerTask() {},参数2 ，参数3 )

指的是：通过参数2，参数3 去描述什么时候去做参数1的事情，现在的情况是：10s之后去做参数一的事情。

因为锁的失效时间是30s，当10s之后，此时这个timeTask 就触发了，他就去进行续约，把当前这把锁续约成30s，如果操作成功，那么此时就会递归调用自己，再重新设置一个timeTask()，于是再过10s后又再设置一个timerTask，完成不停的续约。

那么大家可以想一想，假设我们的线程出现了宕机他还会续约吗？当然不会，因为没有人再去调用renewExpiration这个方法，所以等到时间之后自然就释放了。

private void renewExpiration() {ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee == null) {return;}Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent == null) {return;}Long threadId = ent.getFirstThreadId();if (threadId == null) {return;}RFuture<Boolean> future = renewExpirationAsync(threadId);future.onComplete((res, e) -> {if (e != null) {log.error("Can't update lock " + getName() + " expiration", e);return;}if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}

3.5 分布式锁-redission锁的MutiLock原理

为了提高redis的可用性，我们会搭建集群或者主从，现在以主从为例

此时我们去写命令，写在主机上，主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的，这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。

那么MutiLock 加锁原理是什么呢？笔者画了一幅图来说明

当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功，那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试。

黑马Redis视频教程实战篇（三）相关推荐

黑马Redis视频教程实战篇（六）
目录一.附近商户 1.1.GEO数据结构的基本用法 1.2.导入店铺数据到GEO 1.3.实现附近商户功能二.用户签到 2.1.BitMap功能演示 2.2.实现签到功能 2.3.签到统计 2.4 ...
黑马Redis视频教程实战篇（五）
目录一.达人探店 1.1.发布探店笔记 1.2.查看探店笔记 1.3.点赞功能 1.4.点赞排行榜二.好友关注 2.1.关注和取消关注 2.2.共同关注 2.3.Feed流实现方案 2.4.推送到 ...
黑马Redis学习——实战篇（4）
目录 4.分布式锁 4.1 .基本原理和实现方式对比 4.2 .Redis分布式锁的实现核心思路 4.3 实现分布式锁版本一 4.4 Redis分布式锁误删情况说明 4.5 解决Redis分布式锁误删 ...
黑马Redis笔记高级篇 | 多级缓存
黑马Redis笔记高级篇 | 多级缓存(黑马教程云服务器踩坑记录) 1.JVM进程缓存(tomcat服务内部) 1.1 导入商品案例 1.2 初识Caffeine 1.3 实现进程缓存 2.Lua语法 ...
Vue实战篇三十三：实现新闻的浏览历史
系列文章目录 Vue基础篇一:编写第一个Vue程序 Vue基础篇二:Vue组件的核心概念 Vue基础篇三:Vue的计算属性与侦听器 Vue基础篇四:Vue的生命周期(秒杀案例实战) Vue基础篇五:V ...
Vue实战篇三十一：实现一个改进版的头条新闻
系列文章目录 Vue基础篇一:编写第一个Vue程序 Vue基础篇二:Vue组件的核心概念 Vue基础篇三:Vue的计算属性与侦听器 Vue基础篇四:Vue的生命周期(秒杀案例实战) Vue基础篇五:V ...
Vue实战篇三十：实现一个简易版的头条新闻
系列文章目录 Vue基础篇一:编写第一个Vue程序 Vue基础篇二:Vue组件的核心概念 Vue基础篇三:Vue的计算属性与侦听器 Vue基础篇四:Vue的生命周期(秒杀案例实战) Vue基础篇五:V ...
[知识图谱实战篇] 三.Python提取JSON数据、HTML+D3构建基本可视化布局
前面作者讲解了很多知识图谱原理知识,包括知识图谱相关技术.Neo4j绘制关系图谱等,但仍缺少一个系统全面的实例.为了加深自己对知识图谱构建的认识,为后续创建贵州旅游知识图谱打下基础,作者深入学习了张宏 ...
用Kotlin撸一个图片压缩插件-实战篇(三)
简述: 由于个人原因,已经有很长一段时间没有写过文章,有句话是那么说的只要开始就不会太晚,所以我们开始<用Kotlin撸一个图片压缩插件>系列文章最后一篇实战篇.实际上我已经把源码发布到了 ...

黑马Redis视频教程实战篇（三）

一、优惠券秒杀

1.1 全局唯一ID

1.2 Redis实现全局唯一ID

1.3 添加优惠卷

1.4 实现秒杀下单

1.5 库存超卖问题分析

1.6 代码实现乐观锁解决超卖问题

1.7 优惠券秒杀-一人一单

1.8 集群环境下的并发问题

二、分布式锁

2.1 基本原理和实现方式对比

2.2 Redis分布式锁的实现核心思路

2.3 实现分布式锁版本一

2.4 Redis分布式锁误删情况说明

2.5 解决Redis分布式锁误删问题

2.6 分布式锁的原子性问题

2.7 Lua脚本解决多条命令原子性问题

2.8 利用Java代码调用Lua脚本改造分布式锁

三、分布式锁-redission

3.1 分布式锁-redission功能介绍

3.2 分布式锁-Redission快速入门

3.3 分布式锁-redission可重入锁原理

3.4 分布式锁-redission锁重试和WatchDog机制

3.5 分布式锁-redission锁的MutiLock原理

黑马Redis视频教程实战篇（三）相关推荐

最新文章

热门文章