Redis解决秒杀下单

秒杀接口

基础下单实现

controller层实现

/**
 * 秒杀下单业务
 */
@RestController
@RequestMapping("/voucher-order")
public class VoucherOrderController {


    @Resource
    private IVoucherOrderService voucherOrderService;

    @PostMapping("seckill/{id}")
    public Result seckillVoucher(@PathVariable("id") Long voucherId) {
        return voucherOrderService.seckillVoucher(voucherId);
    }
}

service层实现下单【未涉及下单模块】

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    /**
     * 实现优惠卷下单
     * @param voucherId
     * @return
     */
    @Transactional  //添加事务
    @Override
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券id
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 查询优惠卷信息
        //3. 判断秒杀是否开启
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            //否 返回异常， 结束
            return Result.fail("秒杀未开始！");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已经结束！");
        }
        //4. 是，判断库存是否充足
            //否 返回异常， 结束
        Integer stock = voucher.getStock();
        if (stock < 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.订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        // 6.2.用户id
        Long userId = UserHolder.getUser().getId();
        voucherOrder.setUserId(userId);
        // 6.3.代金券id
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);

        return Result.ok(orderId);
    }
}

当我们点击限时抢购时 ，如果所有条件允许，就会下单成功

数据库优惠卷数量也会减1

订单表也会添加订单

上述就是实现最基本的优惠卷下单功能。当然真实的业务场景绝对不会是向我们这么简单的。

在同一时间会有上万的用户同时点击限时抢购 按钮，此刻的并发量就会达到非常大。就会出现一系列的安全问题。

比如: 超卖问题、一人一单问题、集群模式下线程安全问题….. 。下面我们就需要解决这些问题

库存超卖问题

在高并发的场景下会出现的情况

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 是为了在操作失败时，再次进行自旋操作，即把之前的逻辑再操作一次。

乐观锁方案

方案一:

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

​ 但是以上这种方式通过测试发现会有很多失败的情况，失败的原因在于：在使用乐观锁过程中假设100个线程同时都拿到了100的库存，然后大家一起去进行扣减，但是100个人中只有1个人能扣减成功，其他的人在处理时，他们在扣减时，库存已经被修改过了，所以此时其他线程都会失败

方案二：

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

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

一人一单问题

::: 要求同一个优惠券，一个用户只能下一单

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

intern() 这个方法是从常量池中拿到数据，如果我们直接使用userId.toString() 他拿到的对象实际上是不同的对象，new出来的对象，我们使用锁必须保证锁必须是同一把，所以我们需要使用intern()方法

加锁

synchronized (UserId.toString().intern()){
            IVoucherOrderService orderService = (IVoucherOrderService) AopContext.currentProxy();
            return orderService.createVoucherOrder(voucherId);

整个代码实现

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    /**
     * 实现优惠卷秒杀下单
     * @param voucherId
     * @return
     */
//    @Transactional  //添加事务
    @Override
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券id
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 查询优惠卷信息
        //3. 判断秒杀是否开启
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            //否 返回异常， 结束
            return Result.fail("秒杀未开始！");
        }
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已经结束！");
        }
        //4. 是，判断库存是否充足
            //否 返回异常， 结束
        Integer stock = voucher.getStock();
        if (stock < 1){
            return Result.fail("已经被抢完了！");
        }
//todo 需要给当前对象加锁操作----------------------
        Long UserId = UserHolder.getUser().getId();
        /**
         * 获取互斥锁，只允许一个进入
         */
 
        synchronized (UserId.toString().intern()){
            IVoucherOrderService orderService = (IVoucherOrderService) AopContext.currentProxy();
            return orderService.createVoucherOrder(voucherId);
            //关闭锁
           // lock.unLock();
        }
//todo------------------------------
    }

    /**
     * 对于一人一单加安全锁
     * @param voucherId
     * @return
     */
    @Transactional
    public Result createVoucherOrder(Long voucherId){
        //判断用户是否购买过
        Long UserId = UserHolder.getUser().getId();
        /**
         * 一人一单解决,加锁
         */
        Integer count = query().eq("user_id", UserId).eq("voucher_id", voucherId).count();
        if(count > 0){
            return Result.fail("该用户已经购买过了！");
        }
        //5. 库存充足，扣减库存
        boolean update = seckillVoucherService.update().setSql("stock = stock - 1")
                .eq("voucher_id", voucherId).eq("stock",0) //对乐观锁的判断
                .update();
        if(!update){
            return Result.fail("库存不足！");
        }

        VoucherOrder order = new VoucherOrder();
        //创建用户id，代金卷id ，订单id
        long orderId = redisIdWorker.nextId("order");
        order.setId(orderId);

        //6. 创建订单  .返回订单信息
        order.setUserId(UserId);
        order.setVoucherId(voucherId);
        save(order);
        return Result.ok(voucherId);
    }
}

**使用切面代理需要注意的点 **： 在项目启动的地方，暴露代理对象

@MapperScan("com.hmdp.mapper")
@SpringBootApplication
//暴露代理对象
@EnableAspectJAutoProxy(exposeProxy = true)
public class HmDianPingApplication {

    public static void main(String[] args) {
        SpringApplication.run(HmDianPingApplication.class, args);
        System.out.println("Local :" + "http://localhost:8081/");
    }
}

测试结果

一个用户数量只会减少一个

以上的一人一单方法只适合单体情况下，如果在集群模式下就会失败

通过idea提供的功能，自己开启集群。操作如下:

通过以下设置覆盖yaml文件中的端口

锁的原理: 在我们当前的jvm内部维护了一个锁监控器对象 ，我们这里用的是userId，userId在常量池中存储

在一个jvm中，维护了一个线程池，所以当id相同时 ，他永远都是一个锁（锁的监视器是同一个）。 但是如果是集群模式下就是多个jvm，多个jvm中的锁监视器是多个tomcat ，多个jvm，多个常量池。而常量池中的userId只是存储在jvm1的常量池中，而非同时几个都存在。

所以另一个就会成功。所以还会出现线程安全问题

需要使用实现跨jvm的锁 ，也就是 分布式锁

分布式锁

​

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

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

分布式锁满足的条件

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

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

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

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

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

常见的三种分布式锁

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

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

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

基于redis实现分布式锁

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

• 获取锁：

  • 互斥：确保只能有一个线程获取锁
  • 非阻塞：尝试一次，成功返回true，失败返回false

• 释放锁：

  • 手动释放
  • 超时释放：获取锁时添加一个超时时间

  

核心思路：

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

• 同时可以解决误删锁的问题

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)+"=";




    //todo 获取分布式锁
    @Override
    public boolean tryLock(long timeOutSec) {
        //获取线程的表示
        String value = ID_PREFIX +  Thread.currentThread().getId();
        //获取锁
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, value, timeOutSec, TimeUnit.MINUTES);
        //注意自动拆箱出现的空指针错误
        return Boolean.TRUE.equals(aBoolean);
    }

    //todo 释放锁
    @Override
    public void unLock() {

        //获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();

        //获取锁中的标识
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);

        //判断两种锁的 标识是否一致
        if(id.equals(threadId)){
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }
}

service层执行

try {
    IVoucherOrderService orderService = (IVoucherOrderService) AopContext.currentProxy();
    return orderService.createVoucherOrder(voucherId);
}  finally {
    lock.unLock();
}

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

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：https://www.runoob.com/lua/lua-tutorial.html

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命令来调用脚本，调用脚本的常见命令如下 ：

用Lua编写下列业务流程

java调用Lua脚本改进分布式锁

1. 写lua脚本

2. 在idea中插入

3. 加载脚本

4. 调用脚本