Redis 分布式锁+秒杀异步优化

文章目录

问题
思路
setnx实现
- 锁误删问题和解决方案
- Redis Lua脚本
- - 问题引出
  - 解决方案
setnx实现的问题
Redission
- 快速入门
- redission可重入锁原理
秒杀优化(异步优化)
- 异步秒杀思路
- 秒杀资格判断
- Redis消息队列

问题

比如我们两个机器都部署了我们项目，这里nginx使用轮询的方式去做负载均衡
在这里插入图片描述
这里用postman测试

出问题了：两个请求都会进入到synchronized，这两个请求是同一个用户同一个商品，按理来说是一人一单不能同时进入synchronized代码块
但是我们采用了分布式，我们请求1进入的8081端口，请求2进入的8082端口
相同的代码，部署在不同的机器上，synchronized锁不住其他机器/端口上的代码
这样的话还是实现不了一人一单
在这里插入图片描述
原理图
一个JVM一个锁监视器，只能保证请求该JVM的线程互斥，保证不了其他JVM的，所以分布式情况下也可能两个线程（userid相同）进入同步代码块，导致线程安全问题

思路

在JVM外单起一个服务，所有节点都会找这个服务去获取锁，这样的话就可以实现不同JVM同一代码块的锁
在这里插入图片描述
下面是一些实现方案
方案对应上文提到的服务

下面我们主要讲redis的解决方案

setnx实现

set lock thread1 NX EX 10
NX代表set命令是setnx（不存在可以添加返回1，存在可以失败返回0）
EX（expire）后面代表过期时间为了防止
setnx lock thread1
expire lock 10
两个命令之间redis宕机导致锁设置了但是没有过期时间
我们使用set lock thread1 NX EX 10来同时设置key和过期时间

在这里插入图片描述
初版Java代码
锁对象以及一些方法

public interface ILock {  /**  * 尝试获取锁  * @param timeoutSec 锁持有的超时时间，过期后自动释放  * @return true代表获取锁成功；false代表获取锁失败  */  boolean tryLock(long timeoutSec);  /**  * 释放锁  */  void unlock();  
}  public class SimpleRedisLock implements ILock {private String name;private RedisTemplate redisTemplate;public SimpleRedisLock(String name, RedisTemplate redisTemplate) {this.name = name;this.redisTemplate = redisTemplate;}private static final String KEY_PREFIX = "lock:";@Override  public boolean tryLock(long timeoutSec) {Long ThreadId = BaseContext.getCurrent().getId();Boolean flag = redisTemplate.opsForValue().setIfAbsent(KEY_PREFIX+ name, ThreadId+"", timeoutSec, TimeUnit.SECONDS);
//        如果 flag 为 true，返回 true；如果 flag 为 false 或 null，返回 false。return BooleanUtil.isTrue(flag);}@Override  public void unlock() {redisTemplate.delete(KEY_PREFIX+ name);}  
}

一人一单测试代码
分布式锁（悲观锁）解决一人一单的问题，查后insert语句，乐观锁解决查后update语句

因为insert原来没有数据所以不可以用乐观锁，而update可以

		Long userId = UserHolder.getUser().getId();//确保锁的针对同一个用户SimpleRedisLock lock = new SimpleRedisLock("order:"+userId, redisTemplate);boolean isLock = lock.tryLock(1200); // 获取锁if (!isLock) {// 获取锁失败， 返回或者重试return Result.fail("不允许重复下单");//新增这块用的同步锁（分布式），而查询减票是乐观锁}try{// 获取代理对象（事务）IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();  // 拿到当前对象的代理对象，其实就是IVoucherOrderService这个接口的代理对象，返回的是Object，做个强转return proxy.createVoucherOrder(voucherId);  // 如果报错了是因为我们的接口中没有这个方法，那我们就在接口中创建一下这个方法就行}finally {lock.unlock();// 释放锁}

锁误删问题和解决方案

初始线程阻塞可能释放其他线程的锁
在这里插入图片描述
解决方法：释放锁的时候看是不是自己获取的锁
释放锁时观察是否是自己获取的锁

解决释放的锁不是自身原先获取的锁

在获取锁时存入线程标识（可以用UUID+ThreadId表示）
在释放锁时先获取锁中的线程序示，判断是否与当前线程标识一致
1.如果一致则释放锁
2.如果不一致则不释放锁

why use UUID?
ThreadId的规律是每个JVM创建线程就会自增赋值，这样的话可能JVM1和JVM2相同的线程id同时进入代码，单用ThreadId可能会导致误删的情况
所以这里使用UUID作为表示，由于这里UUID是static属性，所有对象公用一个，只能表示JVM的唯一性，而不能标识JVM中的线程，所以释放锁的时候比对ThreadId
static UUID标识JVM，ThreadID标识JVM里面的线程，保证线程唯一性

	private static final String KEY_PREFIX = "lock:";private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";@Overridepublic boolean tryLock(long timeoutSec) {// 获取线程标识String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁Boolean success = redisTemplate.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 = (String) redisTemplate.opsForValue().get(KEY_PREFIX + name);// 判断标识是否一致if (threadId.equals(id)) {// 释放锁redisTemplate.delete(KEY_PREFIX + name);}}

Redis Lua脚本

问题引出

由于我们判断和释放锁不是原子性操作
如果我们判断后发生了阻塞（JVM垃圾回收），锁超时释放，然后另一个线程获取
之后线程1恢复，因为已经判断，所以还是会删除锁，导致并发问题
在这里插入图片描述

解决方案

保证我们判断锁是否为自己的和释放锁的操作为原子的即可

Lua脚本类似于mysql中的事务保证了多个操作的原子性，redis中的事务不同于mysql事务达不到这种效果，所以采用Lua脚本来实现

为什么使用 Lua 脚本？

原子性：Lua 脚本在 Redis 中执行时是原子的，即脚本在执行期间不会被其他命令中断。
Lua 脚本在执行期间，Redis 不会处理其他命令，确保脚本的原子性。但长时间运行的脚本可能导致 Redis 阻塞，需谨慎使用。
减少网络开销：将多个操作合并为一个脚本，减少客户端与服务器之间的通信次数。
复杂操作：Lua 脚本支持条件判断、循环等复杂逻辑，适合处理需要多个步骤的操作。
Redis 会缓存加载的脚本，通过 SCRIPT LOAD 加载的脚本会一直保留，直到服务器重启或使用 SCRIPT FLUSH 清除。

使用脚本我们需要先编写一个脚本

在 Lua 脚本中，可以通过 redis.call() 或 redis.pcall() 调用 Redis 命令。
redis.call()：执行命令，出错时抛出异常。
redis.pcall()：执行命令，出错时返回错误信息。

EVAL "return redis.call('SET', KEYS[1], ARGV[1])" 1 mykey myvalue
KEYS[1] 是键参数，ARGV[1] 是值参数。

1 表示有 1 个键参数。

用法

EVAL：执行 Lua 脚本

EVAL "return 'Hello, Redis!'" 0

其中，“return ‘Hello, Redis!’” 是 Lua 脚本，0 表示没有键参数。

SCRIPT LOAD:加载脚本到 Redis，返回 SHA1 校验和。

SCRIPT LOAD "return 'Hello, Redis!'"

返回的 SHA1 值可用于后续的 EVALSHA 命令。

EVALSHA:通过 SHA1 值执行已加载的脚本。(此时已经缓存了)

EVALSHA <SHA1> 0

Lua脚本实现释放分布式锁

-- 这里的 KEYS[1] 就是传入的key，这里的 ARGV[1] 就是当前传递的参数值  
-- 获取当前的值，判断是否与当前传递的参数一致  
if (redis.call('GET', KEYS[1]) == ARGV[1]) then  -- 一致，则删除  return redis.call('DEL', KEYS[1])  
end  
-- 不一致，则返回0  
return 0

setnx实现的问题

setnx实现分布式锁问题
在这里插入图片描述
不可重入：方法A调用方法B，方法A和方法B都需要获取到该锁资源，若是不可冲入锁，会导致死锁，A执行不完不能释放锁，B拿不到锁导致A执行不完
不可重试：直接返回了false，应为可充实
超时释放：最好可以动态的超时释放
主从一致性：从节点变为主节点时候没有同步到锁的信息，然后其他线程就可以进入了

Redission

Redisson 是一个用于 Java 的 Redis 客户端，它不仅提供了对 Redis 数据库的简单 API 接口，还提供了许多高级功能，旨在简化分布式应用程序的开发。它又以下的特性：

丰富的数据结构：提供了多种高级数据结构，如映射、集合、列表等，兼容 Java 集合框架。
分布式执行：支持分布式任务处理，实现高并发的任务执行。
分布式锁：确保在分布式环境下对共享资源的安全访问。
对象映射：自动序列化和反序列化 Java 对象，简化数据存取。
反应式编程支持：适合高并发和低延迟的应用程序。
高可用性：支持 Redis Sentinel 和 Redis Cluster，确保稳定运行。

快速入门

导入依赖

<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") //redis://192.168.150.101:6379.setPassword("123456");// 创建RedissonClient对象return Redisson.create(config);}
}

使用Redisson

@Resource  
private RedissonClient redissonClient;  @Test  
void testRedisson() throws InterruptedException {  // 获取锁（可重入），指定锁的名称  RLock lock = redissonClient.getLock("anyLock");  // 尝试获取锁，参数分别是：获取锁的最大等待时间（单位是时间尝试），锁自动释放的时间，时间单位  boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS); // 无参的话就是不等待，30秒自动释放  // 判断释放锁获取成功  if (isLock) {  try {  System.out.println("执行业务");  } finally {  // 释放锁  lock.unlock();  }  }  
}

redission可重入锁原理

未完待续

秒杀优化(异步优化)

异步秒杀思路

先回顾一下
当用户发起请求，此时会请求nginx，nginx会访问到tomcat，而tomcat中的程序，会进行串行操作，分成如下几个步骤

1、查询优惠卷2、判断秒杀库存是否足够3、查询订单4、校验是否是一人一单5、扣减库存6、创建订单

在这六步操作中，又有很多操作是要去操作数据库的，而且还是一个线程串行执行，这样就会导致我们的程序执行的很慢，所以我们需要异步程序执行，那么如何加速呢？
在这里插入图片描述
优化方案：
我们将耗时比较短的逻辑判断放入到redis中，比如是否库存足够，是否一人一单，只要这种逻辑可以完成，就意味着我们是一定可以下单完成的，我们只需要进行快速的逻辑判断，根本就不用等下单逻辑走完，我们直接给用户返回成功
再在后台开一个线程，后台线程慢慢的去执行queue里边的消息

第一个难点是我们怎么在redis中去快速校验一人一单，还有库存判断
第二个难点是由于我们校验和tomct下单是两个线程，那么我们如何知道到底哪个单他最后是否成功，或者是下单完成，为了完成这件事我们在redis操作完之后，我们会将一些信息返回给前端，同时也会把这些信息丢到异步queue中去，后续操作中，可以通过这个id来查询我们tomcat中的下单逻辑是否完成了。

在这里插入图片描述

秒杀资格判断

redis中去做秒杀资格的判断
需要有两个数据1.库存2.用户是否购买过该优惠券
信息1库存可以直接存库存信息
信息2可以是value为set存储购买过该优惠券的用户id进行比对判断
在这里插入图片描述

整体思路：当用户下单之后，判断库存是否充足只需要导redis中去根据key找对应的value是否大于0即可，如果不充足，则直接结束，如果充足，继续在redis中判断用户是否可以下单，如果set集合中没有这条数据，说明他可以下单，如果set集合中没有这条记录，则将userId和优惠卷存入到redis中，并且返回0，整个过程需要保证是原子性的，我们可以使用lua来操作

当以上判断逻辑走完之后，我们可以判断当前redis中返回的结果是否是0 ，如果是0，则表示可以下单，则将之前说的信息存入到到queue中去，然后返回，然后再来个线程异步的下单，前端可以通过返回的订单id来判断是否下单成功。
在这里插入图片描述
优化代码

首先，在添加优惠券的同时，我们需要将该优惠券及其库存保存到redis中，方便我们之后在redis中快速判断优惠券库存是否充足。对添加优惠券方法做修改如下。

@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {// MP保存优惠券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());
}

redis判断采用lua脚本，代码如下。

-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2.订单key
local orderKey = 'seckill:order:' .. voucherId-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then-- 3.2.库存不足，返回1return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then-- 3.3.存在，说明是重复下单，返回2return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0

redis如何操作执行lua脚本？
1.写好lua脚本在同一目录
2.在需要执行的类中定义并读取lua脚本
3.在执行处使用redistemplate.execute执行并传参，参数是lua里面定义的
在这里插入图片描述

 //提前初始化脚本，避免每次去执行脚本时单独去创建脚本对象private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//加载lua脚本SECKILL_SCRIPT.setResultType(Long.class);//设置返回值}/*** 使用lua脚本完成扣减资格判断* @param voucherId* @return*/@Override
public Result seckillVoucher(Long voucherId) {//获取用户Long userId = UserHolder.getUser().getId();//生成订单idlong orderId = redisIdWorker.nextId("order");// 1.执行lua脚本  判断是否满足扣减资格（看缓存里面是否有 重复订单）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),//lua脚本里面没有代表key的参数  这里传入空参voucherId.toString(), userId.toString(), String.valueOf(orderId)  //根据lua脚本传入多个参数);int r = result.intValue();//将long类型转换为int  再去判断// 2.判断结果是否为0if (r != 0) {// 2.1.不为0 ，代表没有购买资格return Result.fail(r == 1 ? "库存不足" : "不能重复下单");   //r == 2 代表不能重复下单}//TODO 保存阻塞队列  待完成// 3.返回订单idreturn Result.ok(orderId);
}

Redis消息队列

抢购成功，将优惠券id和用户id封装后存入阻塞队列，然后开启线程池去阻塞队列里拿东西执行

BlockingQueue:阻塞队列对象，当一个线程从阻塞队列get值的时候，如果没有元素就会阻塞

1、创建阻塞队列，将 voucherOrder 对象放到队列当中

    /*** 初始化阻塞队列*/private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);// 保存阻塞队列  VoucherOrder voucherOrder = new VoucherOrder();// 2.3.订单idvoucherOrder.setId(orderId);// 2.4.用户idvoucherOrder.setUserId(userId);// 2.5.代金券idvoucherOrder.setVoucherId(voucherId);// 2.6.放入阻塞队列orderTasks.add(voucherOrder);

2.异步下单
视频

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//初始化返回值SECKILL_SCRIPT.setResultType(Long.class);}@PostConstruct //注解含义：在当前类初始化完毕后执行private void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//创建阻塞队列private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);//创建线程池（单线程线程池）private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//内部类，规定线程执行逻辑private class VoucherOrderHandler implements Runnable{@Overridepublic void run() {while (true){try {// 1.获取队列中的订单信息VoucherOrder voucherOrder = orderTasks.take();// 2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理订单异常",e);}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.获取用户Long userId = voucherOrder.getUserId();// 2.创建锁对象RLock lock = redissonClient.getLock("lock:order:" + userId);// 3.获取锁boolean isLock = lock.tryLock();// 4.若获取锁失败if (!isLock) {log.error("不允许重复下单");return;}// 获取锁成功 （理论上没有问题，lua脚本已经判断过了，这里再加锁只是兜底）try {//通过代理对象调用proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {boolean success = seckillVoucherService.update().setSql("stock = stock - 1")  //使用MP，设置sql语句.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();save(voucherOrder);}private IVoucherOrderService proxy;@Overridepublic Result seckillVoucher(Long voucherId) {// 获取用户Long userId = UserHolder.getUser().getId();// 1.执行lua脚本，判断用户是否用购买资格（库存不足与重复下单问题）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());// 2.判断结果是否为0int r = result.intValue();if(r!=0){// 2.1.不为0，代表没有购物资格return Result.fail(r==1?"库存不足":"不能重复下单");}// 2.2 为0，有购买资格，先创建订单，再将订单信息添加到阻塞队列VoucherOrder voucherOrder = new VoucherOrder();// 2.3 获取订单id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 2.4将订单信息存入阻塞队列，任务结束orderTasks.add(voucherOrder);//3.获取代理对象，方便后序线程使用，可以放在成员变量或者是voucherOrder里面proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回订单idreturn Result.ok(orderId);}}