Redis学习笔记（实战篇3）

一、分布式锁-redission1. 存在的问题(1) 不可重入// 方法A加了分布式锁 public void methodA() { lock(); // 线程拿到锁 methodB(); // 方法B也加了同一个分布式锁 unlock(); } // 方法B也加了同一个分布式锁 public void methodB() { lock(); // 同一个线程再次申请锁被拒绝 → 死锁 // ...业务 unlock(); }线程执行methodA拿到锁后调用methodB时再次尝试拿同一个锁这时候分布式锁会认为 “锁已经被别人占了”导致线程自己阻塞自己形成死锁。(2) 不可重试现在的setnx实现线程尝试拿锁一次如果失败返回false就直接结束了没有 “再试一次” 的机制。但实际业务里比如秒杀、订单创建锁竞争往往是短暂的线程应该可以重试几次提高拿到锁的成功率。(3) 超时释放我们给锁加了过期时间比如 30 秒本来是为了防止 “服务挂了锁不释放” 导致死锁但带来了新问题如果业务执行时间超过了锁的过期时间锁会自动释放这时候其他线程就能拿到锁操作同一个资源导致数据不一致。(4) 主从一致性① Redis 主从集群的原理主节点Master负责写操作加锁、解锁从节点Slave负责读操作主节点的数据会异步同步到从节点如果主节点挂了集群会把一个从节点升级为新的主节点② 问题场景线程 A 向主节点加锁成功主节点还没把这个锁数据同步到从节点主节点突然宕机了集群选举一个从节点成为新主节点但这个新主节点没有刚才的锁数据认为锁不存在线程 B 来拿锁直接成功这时候就出现了两个线程同时持有同一个锁的情况锁失效并发安全问题爆发。2. Redission快速入门(1) 配置Redisson客户端Configuration public class RedissonConfig { Bean public RedissonClient redissonClient(){ // 配置 Config config new Config(); config.useSingleServer().setAddress(redis://192.168.150.101:6379) .setPassword(123321); // 创建RedissonClient对象 return Redisson.create(config); } }(2) 如何使用Redission的分布式锁Resource private RedissionClient redissonClient; Test void testRedisson() throws Exception{ //获取锁(可重入)指定锁的名称 RLock lock redissonClient.getLock(anyLock); //尝试获取锁参数分别是获取锁的最大等待时间(期间会重试)锁自动释放时间时间单位 boolean isLock lock.tryLock(1,10,TimeUnit.SECONDS); //判断获取锁成功 if(isLock){ try{ System.out.println(执行业务); }finally{ //释放锁 lock.unlock(); } } }(3) VoucherOrderServiceImplResource 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. redission可重入锁原理(1) 3 个核心参数参数含义作用KEYS[1]锁的大 key锁名称代表这把锁的整体用来判断锁是否存在ARGV[1]锁的过期时间毫秒防止锁死锁即使客户端宕机也会自动释放ARGV[2]锁的小 key持有者标识格式客户端ID : 线程ID用来判断锁是否属于当前线程(2) 脚本的核心逻辑-- 步骤1锁不存在 → 直接加锁 if (redis.call(exists, KEYS[1]) 0) then redis.call(hset, KEYS[1], ARGV[2], 1); -- 新建hash锁小key对应值1第一次持有 redis.call(pexpire, KEYS[1], ARGV[1]); -- 给锁设置过期时间 return nil; -- 返回nil 加锁成功 end; -- 步骤2锁存在但属于当前线程 → 可重入 if (redis.call(hexists, KEYS[1], ARGV[2]) 1) then redis.call(hincrby, KEYS[1], ARGV[2], 1); -- 重入次数1 redis.call(pexpire, KEYS[1], ARGV[1]); -- 刷新过期时间 return nil; -- 返回nil 重入成功 end; -- 步骤3锁存在且不属于当前线程 → 抢锁失败 return redis.call(pttl, KEYS[1]); -- 返回锁的剩余过期时间4. redission锁重试和WatchDog机制(1) Lua 抢锁逻辑条件操作返回值含义锁不存在插入锁Hash 结构设置过期时间null抢锁成功锁存在且属于当前线程重入次数 1刷新过期时间null可重入成功锁存在且不属于当前线程无操作锁的剩余过期时间ttl抢锁失败(2)lock()核心抢锁流程long threadId Thread.currentThread().getId(); Long ttl tryAcquire(-1, leaseTime, unit, threadId); // lock acquired if (ttl null) { return; }说明① Long ttl tryAcquire(-1, leaseTime, unit, threadId);-1是waitTime的默认值表示无限等待直到抢到锁leaseTime锁的过期时间无参lock()时默认-1带参lock(10, TimeUnit.SECONDS)时为 10unit时间单位如TimeUnit.MILLISECONDSthreadId当前线程 ID。② 返回值ttlnull→ 抢锁 / 可重入成功非 null 数字 → 锁被其他线程持有返回锁的剩余过期时间比如返回 20000 代表锁还有 20 秒过期。(3) WatchDog看门狗续约机制RFutureLong ttlRemainingFuture tryLockInnerAsync(waitTime, commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); ttlRemainingFuture.onComplete((ttlRemaining, e) - { if (e ! null) { return; } // lock acquired if (ttlRemaining null) { scheduleExpirationRenewal(threadId); } }); return ttlRemainingFuture;说明①tryLockInnerAsync(...)的第二个参数commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()含义获取 Redisson 配置的「看门狗默认超时时间」默认值是30 秒30000 毫秒作用把锁的初始过期时间设为 30 秒替代用户传入的leaseTime。② if (ttlRemaining null) { scheduleExpirationRenewal(threadId); }逻辑只有抢锁成功ttlRemainingnull才调用scheduleExpirationRenewal(threadId)scheduleExpirationRenewal核心作用是「启动看门狗续约线程」是连接抢锁和续约的关键方法。5. redission锁的MutiLock原理(1) 存在的问题我们去写命令写在主机上 主机会将数据同步给从机但是假设在主机还没有来得及把数据写入到从机去的时候此时主机宕机哨兵会发现主机宕机并且选举一个slave变成master而此时新的master中实际上并没有锁信息此时锁信息就已经丢掉了。(2) 解决方案为了解决这个问题redission提出来了MutiLock锁使用这把锁就不使用主从了每个节点的地位都是一样的 这把锁加锁的逻辑需要写入到每一个主丛节点上只有所有的服务器都写入成功此时才是加锁成功。假设现在某个节点挂了那么它去获得锁的时候只要有一个节点拿不到都不能算是加锁成功就保证了加锁的可靠性。二、秒杀优化1. 异步秒杀思路(1) 当用户下单之后判断库存是否充足只需要到redis中去根据key找对应的value是否大于0即可。如果不充足则直接结束如果充足继续在redis中判断用户是否可以下单如果set集合中没有这条数据说明它可以下单如果set集合中没有这条记录则将userId和优惠卷存入到redis中并且返回0整个过程需要保证是原子性的我们可以使用lua来操作。(2) 校验通过后无需等待完整下单流程完成直接给用户返回 “下单受理成功”附带订单 ID同时将下单任务丢入异步队列后台单独线程消费异步队列中的任务慢慢执行完整的数据库下单逻辑创建订单、扣减库存等前端通过返回的订单 ID查询异步下单的最终结果成功 / 失败。2. Redis完成秒杀资格判断(1) 需求新增秒杀优惠券的同时将优惠券信息保存到Redis中基于Lua脚本判断秒杀库存、一人一单决定用户是否抢购成功如果抢购成功将优惠券id和用户id封装后存入阻塞队列开启线程任务不断从阻塞队列中获取信息实现异步下单功能(2) 代码实现完整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.库存不足返回1 return 1 end -- 3.2.判断用户是否下单 SISMEMBER orderKey userId if(redis.call(sismember, orderKey, userId) 1) then -- 3.3.存在说明是重复下单返回2 return 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说明① Lua 里的..是什么..是Lua 语言的字符串拼接运算符作用和 Java 里用拼接字符串比如a b完全一样只是语法不同。② if(tonumber(redis.call(get, stockKey)) 0) then return 1 endtonumber()是 Lua 的内置函数作用是把字符串类型的数字转成数值类型VoucherOrderServiceImplOverride public Result seckillVoucher(Long voucherId) { //获取用户 Long userId UserHolder.getUser().getId(); long orderId redisIdWorker.nextId(order); // 1.执行lua脚本 Long result stringRedisTemplate.execute( SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(orderId) ); int r result.intValue(); // 2.判断结果是否为0 if (r ! 0) { // 2.1.不为0 代表没有购买资格 return Result.fail(r 1 ? 库存不足 : 不能重复下单); } //TODO 保存阻塞队列 // 3.返回订单id return Result.ok(orderId); }3. 基于阻塞队列实现秒杀优化//异步处理线程池 private static final ExecutorService SECKILL_ORDER_EXECUTOR Executors.newSingleThreadExecutor(); //在类初始化之后执行因为当这个类初始化好了之后随时都是有可能要执行的 PostConstruct private void init() { SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler()); } // 用于线程池处理的任务 // 当初始化完毕后就会去从对列中去拿信息 private class VoucherOrderHandler implements Runnable{ Override public 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 redisLock redissonClient.getLock(lock:order: userId); // 3.尝试获取锁 boolean isLock redisLock.lock(); // 4.判断是否获得锁成功 if (!isLock) { // 获取锁失败直接返回失败或者重试 log.error(不允许重复下单); return; } try { //注意由于是spring的事务是放在threadLocal中此时的是多线程事务会失效 proxy.createVoucherOrder(voucherOrder); } finally { // 释放锁 redisLock.unlock(); } } //a private BlockingQueueVoucherOrder orderTasks new ArrayBlockingQueue(1024 * 1024); Override public Result seckillVoucher(Long voucherId) { Long userId UserHolder.getUser().getId(); long orderId redisIdWorker.nextId(order); // 1.执行lua脚本 Long result stringRedisTemplate.execute( SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(orderId) ); int r result.intValue(); // 2.判断结果是否为0 if (r ! 0) { // 2.1.不为0 代表没有购买资格 return Result.fail(r 1 ? 库存不足 : 不能重复下单); } VoucherOrder voucherOrder new VoucherOrder(); // 2.3.订单id long orderId redisIdWorker.nextId(order); voucherOrder.setId(orderId); // 2.4.用户id voucherOrder.setUserId(userId); // 2.5.代金券id voucherOrder.setVoucherId(voucherId); // 2.6.放入阻塞队列 orderTasks.add(voucherOrder); //3.获取代理对象 proxy (IVoucherOrderService)AopContext.currentProxy(); //4.返回订单id return Result.ok(orderId); } Transactional public void createVoucherOrder(VoucherOrder voucherOrder) { Long userId voucherOrder.getUserId(); // 5.1.查询订单 int count query().eq(user_id, userId).eq(voucher_id, voucherOrder.getVoucherId()).count(); // 5.2.判断是否存在 if (count 0) { // 用户已经购买过了 log.error(用户已经购买过了); return ; } // 6.扣减库存 boolean success seckillVoucherService.update() .setSql(stock stock - 1) // set stock stock - 1 .eq(voucher_id, voucherOrder.getVoucherId()).gt(stock, 0) // where id ? and stock 0 .update(); if (!success) { // 扣减失败 log.error(库存不足); return ; } save(voucherOrder); }说明① private static final ExecutorService SECKILL_ORDER_EXECUTOR Executors.newSingleThreadExecutor();这是 Java 的线程池Executors.newSingleThreadExecutor()会创建一个「只有 1 个工作线程」的线程池 —— 所有任务都由这 1 个线程按顺序处理。② PostConstruct private void init() { SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler()); }PostConstructSpring 注解作用是「当前类被 Spring 初始化完成后项目启动时立即执行这个方法」③ private class VoucherOrderHandler implements Runnable { ... }这是一个内部线程任务类实现了Runnable接口Java 中 “可被线程执行的任务” 都要实现这个接口里面的run()方法是线程要执行的核心逻辑。三、Postman测试方法1. 发送验证码不需要登录(1) 请求方式POST(2) 直连后端http://localhost:8082/user/code?phone[phone](3) 预期响应{ success: true }2. 登录获取 token(1) 请求方式POST(2) 直连后端http://localhost:8082/user/login(3) Body选择 raw格式 JSON{ phone: [phone], code: 518208 }注意code 是第一步发送验证码后在控制台日志里看到的 6 位数字(4) 预期响应1c3e03a8a5734dd1a8a5285c0d49a63c这个字符串就是 token复制保存下来3. 测试登录状态验证 token 是否有效(1) 请求方式GET(2) 直连后端http://localhost:8082/user/me(3) Headersauthorization: 1c3e03a8a5734dd1a8a5285c0d49a63c(4) 预期响应{ success: true, data: { id: 1, nickName: user_xxx, icon: ... } }4. 秒杀下单需要登录(1) 请求方式POST(2) 直连后端http://localhost:8082/voucher-order/seckill/10(3) Headersauthorization: 1c3e03a8a5734dd1a8a5285c0d49a63c(4) Body不需要(5) 可能响应{ success: true, data: 1234567890 }四、JMeter使用方法1. 添加线程组(1) 右键 “测试计划” → 添加 → 线程(用户) → 线程组(2) 配置线程数(用户数)并发用户数如 100Ramp-Up时间(秒)启动时间如 10循环次数每个线程执行次数如 12. 添加 HTTP 请求(1) 右键 “线程组” → 添加 → 取样器 → HTTP 请求(2) 配置名称秒杀下单自定义服务器名称或IPlocalhost或后端服务器IP端口号8081或你的后端端口方法POST路径/voucher-order/seckill/10你的秒杀接口路径3. 添加 HTTP 信息头管理器(1) 右键 “线程组” → 添加 → 配置元件 → HTTP 信息头管理器(2) 配置名称authorization值你的登录token例如1c3e03a8a5734dd1a8a5285c0d49a63c4. 添加监听器查看结果右键 “线程组” → 添加 → 监听器 → 选择察看结果树查看每个请求的详细响应调试用汇总报告查看统计信息吞吐量、错误率等聚合报告更详细的性能报告5. 运行测试点击顶部工具栏的绿色“启动”按钮