网站设计就业前景,保定市做网站的电话,top域名,网站建设取得实效面试重点#xff1a;本篇包含悲观锁#xff0c;乐观锁#xff0c;多线程以及分布式锁的知识 目录
3.优惠卷秒杀
3.1 -全局唯一ID
3.2 -Redis实现全局唯一Id
3.3 添加优惠卷
3.4 实现秒杀下单
3.5 库存超卖问题分析
3.6 乐观锁解决超卖问题
3.7 优惠券秒杀-一人一单 … 面试重点本篇包含悲观锁乐观锁多线程以及分布式锁的知识 目录
3.优惠卷秒杀
3.1 -全局唯一ID
3.2 -Redis实现全局唯一Id
3.3 添加优惠卷
3.4 实现秒杀下单
3.5 库存超卖问题分析
3.6 乐观锁解决超卖问题
3.7 优惠券秒杀-一人一单
3.8 集群环境下的并发问题
4、分布式锁
4.1 、基本原理和实现方式对比
4.2 、Redis分布式锁的实现核心思路
4.3 实现分布式锁初级版本
4.4 Redis分布式锁误删情况说明
4.5 解决Redis分布式锁误删问题
4.6 分布式锁的原子性问题
4.7 Lua脚本解决多条命令原子性问题
4.8 调用Lua脚本改造分布式锁
5、分布式锁-redission
5.1 redission功能
5.2 Redission入门
5.3 redission可重入锁原理
5.4 redission锁重试和WatchDog机制
5.5 redission锁的MutiLock原理 3.优惠卷秒杀
3.1 -全局唯一ID
每个店铺都可以发布优惠券 当用户抢购时就会生成订单并保存到tb_voucher_order这张表中而订单表如果使用数据库自增ID就存在一些问题 id的规律性太明显 受单表数据量的限制
场景分析如果我们的id具有太明显的规则用户或者说商业对手很容易猜测出来我们的一些敏感信息比如商城在一天时间内卖出了多少单这明显不合适。
场景分析二随着我们商城规模越来越大mysql的单表的容量不宜超过500W数据量过大之后我们要进行拆库拆表但拆分表了之后他们从逻辑上讲他们是同一张表所以他们的id是不能一样的 于是乎我们需要保证id的唯一性。
全局ID生成器是一种在分布式系统下用来生成全局唯一ID的工具一般要满足下列特性 为了增加ID的安全性我们可以不直接使用Redis自增的数值而是拼接一些其它信息 成部分符号位1bit永远为0
时间戳31bit以秒为单位可以使用69年
序列号32bit秒内的计数器支持每秒产生2^32个不同ID
3.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;}
}
测试类
知识小贴士countdownlatch
countdownlatch名为信号枪主要的作用是同步协调在多线程的等待于唤醒问题
我们如果没有CountDownLatch 那么由于程序是异步的当异步程序没有执行完时主线程就已经执行完了然后我们期望的是分线程全部走完之后主线程再走所以我们此时需要使用到CountDownLatch
两个最重要的方法
1、countDown
2、await
await 方法 是阻塞方法我们担心分线程没有执行完时main线程就先执行所以使用await可以让main线程阻塞那么什么时候main线程不再阻塞呢当CountDownLatch 内部维护的 变量变为0时就不再阻塞直接放行那么什么时候CountDownLatch 维护的变量变为0 呢我们只需要调用一次countDown 内部变量就减少1我们让分线程和变量绑定 执行完一个分线程就减少一个变量当分线程全部走完CountDownLatch 维护的变量就是0此时await就不再阻塞统计出来的时间也就是所有分线程执行完后的时间。
Test
void testIdWorker() throws InterruptedException {CountDownLatch latch new CountDownLatch(300);Runnable task () - {for (int i 0; i 100; i) {long id redisIdWorker.nextId(order);System.out.println(id id);}latch.countDown();};long begin System.currentTimeMillis();for (int i 0; i 300; i) {es.submit(task);}latch.await();long end System.currentTimeMillis();System.out.println(time (end - begin));
} Test导的是 import org.junit.jupiter.api.Test包 在执行测试类时需要打开虚拟机在虚拟机上关闭防火墙然后启动redis才可运行 systemctl stop firewalld
cd /usr/local/src/redis-6.2.6
redis-server redis.conf 3.3 添加优惠卷
每个店铺都可以发布优惠券分为平价券和特价券。平价券可以任意购买而特价券需要秒杀抢购 tb_voucher优惠券的基本信息优惠金额、使用规则等 tb_seckill_voucher优惠券的库存、开始抢购时间结束抢购时间。特价优惠券才需要填写这些信息
平价卷由于优惠力度并不是很大所以是可以任意领取
而代金券由于优惠力度大所以像第二种卷就得限制数量从表结构上也能看出特价卷除了具有优惠卷的基本信息以外还具有库存抢购时间结束时间等等字段
新增普通卷代码 VoucherController
PostMapping
public Result addVoucher(RequestBody Voucher voucher) {voucherService.save(voucher);return Result.ok(voucher.getId());
}
新增秒杀卷代码
VoucherController
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());
} 3.4 实现秒杀下单 下单核心思路当我们点击抢购时会触发右侧的请求我们只需要编写对应的controller即可 秒杀下单应该思考的内容
下单时需要判断两点 秒杀是否开始或结束如果尚未开始或已经结束则无法下单 库存是否充足不足则无法下单
下单核心逻辑分析
当用户开始进行下单我们应当去查询优惠卷信息查询到优惠卷信息判断是否满足秒杀条件
比如时间是否充足如果时间充足则进一步判断库存是否足够如果两者都满足则扣减库存创建订单然后返回订单id如果有一个条件不满足则直接结束。 VoucherOrderServiceImpl
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(库存不足);}//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);save(voucherOrder);return Result.ok(orderId);} 3.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中的var1var2 是预估值如果预估值 内存值则代表中间没有被人修改过此时就将新值去替换 内存值
其中do while 是为了在操作失败时再次进行自旋操作即把之前的逻辑再操作一次。
int var5;
do {var5 this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 var4));return var5;
课程中的使用方式
课程中的使用方式是没有像cas一样带自旋的操作也没有对version的版本号1 他的操作逻辑是在操作时对版本号进行1 操作然后要求version 如果是1 的情况下才能操作那么第一个线程在操作后数据库中的version变成了2但是他自己满足version1 所以没有问题此时线程2执行线程2 最后也需要加上条件version 1 但是现在由于线程1已经操作过了所以线程2操作时就不满足version1 的条件了所以线程2无法执行成功 3.6 乐观锁解决超卖问题
修改代码方案一、
VoucherOrderServiceImpl 在扣减库存时改为 以上逻辑的核心含义是只要我扣减库存时的库存和之前我查询到的库存是一样的就意味着没有人在中间修改过库存那么此时就是安全的但是以上这种方式通过测试发现会有很多失败的情况失败的原因在于在使用乐观锁过程中假设100个线程同时都拿到了100的库存然后大家一起去进行扣减但是100个人中只有1个人能扣减成功其他的人在处理时他们在扣减时库存已经被修改过了所以此时其他线程都会失败 修改代码方案二、
之前的方式要修改前后都保持一致但是这样我们分析过成功的概率太低所以我们的乐观锁需要变一下改成stock大于0 即可 知识小扩展
针对cas中的自旋压力过大我们可以使用Longaddr这个类去解决
Java8 提供的一个对AtomicLong改进后的一个类LongAdder
大量线程并发更新一个原子性的时候天然的问题就是自旋会导致并发性问题当然这也比我们直接使用syn来的好
所以利用这么一个类LongAdder来进行优化
如果获取某个值则会对cell和base的值进行递增最后返回一个完整的值 3.7 优惠券秒杀-一人一单
需求修改秒杀业务要求同一个优惠券一个用户只能下一单
现在的问题在于
优惠卷是为了引流但是目前的情况是一个人可以无限制的抢这个优惠卷所以我们应当增加一层逻辑让一个用户只能下一个单而不是让一个用户下多个单
具体操作逻辑如下比如时间是否充足如果时间充足则进一步判断库存是否足够然后再根据优惠卷id和用户id查询是否已经下过这个订单如果下过这个订单则不再下单否则进行下单 VoucherOrderServiceImpl
初步代码增加一人一单逻辑 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(库存不足);}// 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);save(voucherOrder);return Result.ok(orderId);}
存在问题现在的问题还是和之前一样并发过来查询数据库都不存在订单所以我们还是需要加锁但是乐观锁比较适合更新数据而现在是插入数据所以我们需要使用悲观锁操作
注意在这里提到了非常多的问题我们需要慢慢的来思考首先我们的初始方案是封装了一个createVoucherOrder方法同时为了确保他线程安全在方法上添加了一把synchronized 锁
Transactional
public synchronized 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);save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);
}
但是这样添加锁锁的粒度太粗了在使用锁过程中控制锁粒度 是一个非常重要的事情因为如果锁的粒度太大会导致每个线程进来都会锁住所以我们需要去控制锁的粒度以下这段代码需要修改为 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);save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);}
}
但是以上代码还是存在问题问题的原因在于当前方法被spring的事务控制如果你在方法内部加锁可能会导致当前方法事务还没有提交但是锁已经释放也会导致问题所以我们选择将当前方法整体包裹起来确保事务不会出现问题如下
在seckillVoucher 方法中添加以下逻辑这样就能保证事务的特性同时也控制了锁的粒度 但是以上做法依然有问题因为你调用的方法其实是this.的方式调用的事务想要生效还得利用代理来生效所以这个地方我们需要获得原始的事务对象 来操作事务 最终代码
VoucherOrderServiceImpl
package com.hmdp.service.impl;import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;
import java.time.LocalDateTime;/*** p* 服务实现类* /p** author chan* since 2025-5-2*/
Service
public class VoucherOrderServiceImpl extends ServiceImplVoucherOrderMapper, VoucherOrder implements IVoucherOrderService {Resourceprivate ISeckillVoucherService seckillVoucherService;Resourceprivate RedisIdWorker redisIdWorker;Overridepublic 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()){IVoucherOrderService proxy (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);}}Transactionalpublic 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);save(voucherOrder);// 7.返回订单idreturn Result.ok(orderId);}
}pom.xml dependencygroupIdorg.aspectj/groupIdartifactIdaspectjrt/artifactId/dependency
启动类加注解
EnableAspectJAutoProxy(exposeProxy true) 3.8 集群环境下的并发问题
有关锁失效原因分析
由于现在我们部署了多个tomcat每个tomcat都有一个属于自己的jvm那么假设在服务器A的tomcat内部有两个线程这两个线程由于使用的是同一份代码那么他们的锁对象是同一个是可以实现互斥的但是如果现在是服务器B的tomcat内部又有两个线程但是他们的锁对象写的虽然和服务器A一样但是锁对象却不是同一个所以线程3和线程4可以实现互斥但是却无法和线程1和线程2实现互斥这就是 集群环境下syn锁失效的原因在这种情况下我们就需要使用分布式锁来解决这个问题。 4、分布式锁
4.1 、基本原理和实现方式对比
分布式锁满足分布式系统或集群模式下多进程可见并且互斥的锁。
分布式锁的核心思想就是让大家都使用同一把锁只要大家使用的是同一把锁那么我们就能锁住线程不让线程进行让程序串行执行这就是分布式锁的核心思路 那么分布式锁应该满足一些什么样的条件呢 可见性多个线程都能看到相同的结果注意这个地方说的可见性并不是并发编程中指的内存可见性只是说多个进程之间都能感知到变化的意思
互斥互斥是分布式锁的最基本的条件使得程序串行执行
高可用程序不易崩溃时时刻刻都保证较高的可用性
高性能由于加锁本身就让性能降低所有对于分布式锁本身需要他就较高的加锁性能和释放锁性能
安全性安全也是程序中必不可少的一环 常见的分布式锁有三种
Mysqlmysql本身就带有锁机制但是由于mysql性能本身一般所以采用分布式锁的情况下其实使用mysql作为分布式锁比较少见
Redisredis作为分布式锁是非常常见的一种使用方式现在企业级开发中基本都使用redis或者zookeeper作为分布式锁利用setnx这个方法如果插入key成功则表示获得到了锁如果有人插入成功其他人插入失败则表示无法获得到锁利用这套逻辑来实现分布式锁
Zookeeperzookeeper也是企业级开发中较好的一个实现分布式锁的方案由于本套视频并不讲解zookeeper的原理和分布式锁的实现所以不过多阐述 4.2 、Redis分布式锁的实现核心思路
实现分布式锁时需要实现的两个基本方法 获取锁 互斥确保只能有一个线程获取锁 非阻塞尝试一次成功返回true失败返回false 释放锁 手动释放 超时释放获取锁时添加一个超时时间
核心思路
我们利用redis 的setNx 方法当有多个线程进入时我们就利用该方法第一个线程进入时redis 中就有这个key 了返回了1如果结果是1则表示他抢到了锁那么他去执行业务然后再删除锁退出锁逻辑没有抢到锁的等待一定时间后重试即可 4.3 实现分布式锁初级版本 加锁逻辑
锁的基本接口 SimpleRedisLock
利用setnx方法进行加锁同时增加过期时间防止死锁此方法可以保证加锁和增加过期时间具有原子性
private static final String KEY_PREFIXlock:
Override
public boolean tryLock(long timeoutSec) {// 获取线程标示String threadId Thread.currentThread().getId()// 获取锁Boolean success stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX name, threadId , timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(success);
} 释放锁逻辑
SimpleRedisLock
释放锁防止删除别人的锁
public void unlock() {//通过del删除锁stringRedisTemplate.delete(KEY_PREFIX name);
}
修改业务代码 Overridepublic 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(1200);//加锁失败if (!isLock) {return Result.fail(不允许重复下单);}try {//获取代理对象(事务)IVoucherOrderService proxy (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//释放锁lock.unlock();}} 4.4 Redis分布式锁误删情况说明
逻辑说明
持有锁的线程在锁的内部出现了阻塞导致他的锁自动释放这时其他线程线程2来尝试获得锁就拿到了这把锁然后线程2在持有锁执行过程中线程1反应过来继续执行而线程1执行过程中走到了删除锁逻辑此时就会把本应该属于线程2的锁进行删除这就是误删别人锁的情况说明
解决方案解决方案就是在每个线程释放锁的时候去判断一下当前这把锁是否属于自己如果属于自己则不进行锁的删除假设还是上边的情况线程1卡顿锁自动释放线程2进入到锁的内部执行逻辑此时线程1反应过来然后删除锁但是线程1一看当前这把锁不是属于自己于是不进行删除锁逻辑当线程2走到删除锁逻辑时如果没有卡过自动释放锁的时间点则判断当前这把锁是属于自己的于是删除这把锁。 解决方案 此时这个锁已经变成了线程二的锁线程一完成业务后无法释放不属于自己的锁此时线程二完成之前锁一直没有释放 具体代码如下:
package com.hmdp.utils;import cn.hutool.core.lang.UUID;
import org.springframework.data.redis.core.StringRedisTemplate;import java.util.concurrent.TimeUnit;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) {//获取当前创建所的线程标识(value)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 (id.equals(threadId)) {//释放锁stringRedisTemplate.delete(KEY_PREFIX name);}}
}此处的UUID是用来区别不同的JVM的而线程ID是用来区别同一个JVM中的不同线程的 4.5 解决Redis分布式锁误删问题
需求修改之前的分布式锁实现满足在获取锁时存入线程标示可以用UUID表示 在释放锁时先获取锁中的线程标示判断是否与当前线程标示一致 如果一致则释放锁 如果不一致则不释放锁
核心逻辑在存入锁时放入自己线程的标识在删除锁时判断当前这把锁的标识是不是自己存入的如果是则进行删除如果不是则不进行删除。 具体代码如下加锁
private static final String ID_PREFIX UUID.randomUUID().toString(true) -;
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);
}
释放锁
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);}
}
有关代码实操说明
在我们修改完此处代码后我们重启工程然后启动两个线程第一个线程持有锁后手动释放锁第二个线程 此时进入到锁内部再放行第一个线程此时第一个线程由于锁的value值并非是自己所以不能释放锁也就无法删除别人的锁此时第二个线程能够正确释放锁通过这个案例初步说明我们解决了锁误删的问题。 4.6 分布式锁的原子性问题
更为极端的误删逻辑说明
线程1现在持有锁之后在执行业务逻辑过程中他正准备删除锁而且已经走到了条件判断的过程中比如他已经拿到了当前这把锁确实是属于他自己的正准备删除锁但是此时他的锁到期了那么此时线程2进来但是线程1他会接着往后执行当他卡顿结束后他直接就会执行删除锁那行代码相当于条件判断并没有起到作用这就是删锁时的原子性问题之所以有这个问题是因为线程1的拿锁比锁删锁实际上并不是原子性的。 锁名称一样但是锁的线程标识不一样。线程1判断后发现锁的线程标识和当前线程一样于是根据锁名释放锁但是业务阻塞导致自己的锁超时释放此时线程二开始执行线程2拿到同样名称的锁开始执行业务此时线程1的阻塞解决后立刻根据锁名称把线程2的锁误删了可以删的原因是因为已经判断过判断一致后未能来的及释放锁就遭遇了阻塞所以在阻塞立刻解决就会滞后地执行释放锁的行为 4.7 Lua脚本解决多条命令原子性问题
Redis提供了Lua脚本功能在一个脚本中编写多条Redis命令确保多条命令执行时的原子性。Lua是一种编程语言它的基本语法大家可以参考网站Lua 教程 | 菜鸟教程这里重点介绍Redis提供的调用函数我们可以使用lua去操作redis又能保证他的原子性这样就可以实现拿锁比锁删锁是一个原子性动作了作为Java程序员这一块并不作一个简单要求并不需要大家过于精通只需要知道他有什么作用即可。
这里重点介绍Redis提供的调用函数语法如下 写好脚本以后需要用Redis命令来调用脚本调用脚本的常见命令如下 接下来我们来回一下我们释放锁的逻辑
释放锁的业务流程是这样的
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 4.8 调用Lua脚本改造分布式锁
lua脚本本身并不需要大家花费太多时间去研究只需要知道如何调用大致是什么意思即可所以在笔记中并不会详细的去解释这些lua表达式的含义。
我们的RedisTemplate中可以利用execute方法去执行lua脚本参数对应关系就如下图股 Java代码
private static final DefaultRedisScriptLong 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同样可以保证不会删除别人的锁同时还能保证原子性。 5、分布式锁-redission
5.1 redission功能
基于setnx实现的分布式锁存在下面的问题
重入问题重入问题是指 获得锁的线程可以再次进入到相同的锁的代码块中可重入锁的意义在于防止死锁比如HashTable这样的代码中他的方法都是使用synchronized修饰的假如他在一个方法内调用另一个方法那么此时如果是不可重入的不就死锁了吗所以可重入锁他的主要意义是防止死锁我们的synchronized和Lock锁都是可重入的。
不可重试是指目前的分布式只能尝试一次我们认为合理的情况是当线程在获得锁失败后他应该能再次尝试获得锁。
超时释放我们在加锁时增加了过期时间这样的我们可以防止死锁但是如果卡顿的时间超长虽然我们采用了lua表达式防止删锁的时候误删别人的锁但是毕竟没有锁住有安全隐患
主从一致性 如果Redis提供了主从集群当我们向集群写数据时主机需要异步的将数据同步给从机而万一在同步过去之前主机宕机了就会出现死锁问题。 Redission Redisson是一个在Redis的基础上实现的Java驻内存数据网格In-Memory Data Grid。它不仅提供了一系列的分布式的Java常用对象还提供了许多分布式服务其中就包含了各种分布式锁的实现。 5.2 Redission入门
引入依赖
dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.13.6/version
/dependency
配置Redisson客户端
Configuration
public class RedissonConfig {Beanpublic RedissonClient redissonClient(){// 配置Config config new Config();config.useSingleServer().setAddress(redis://192.168.150.101:6379).setPassword(123321);// 创建RedissonClient对象return Redisson.create(config);}
}如何使用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();}}}
在 VoucherOrderServiceImpl
注入RedissonClient
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();}} 注意下面皆为Redisson原理讲解部分 5.3 redission可重入锁原理
在Lock锁中他是借助于底层的一个voaltile的一个state变量来记录重入的状态的比如当前没有人持有这把锁那么state0假如有人持有这把锁那么state1如果持有这把锁的人再次持有这把锁那么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]); 5.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) 这句话相当于对以上抢锁进行了监听也就是说当上边抢锁完毕后此方法会被调用具体调用的逻辑就是去后台开启一个线程进行续约逻辑也就是看门狗线程
RFutureLong 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;}RFutureBoolean future renewExpirationAsync(threadId);future.onComplete((res, e) - {if (e ! null) {log.error(Cant update lock getName() expiration, e);return;}if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
} 5.5 redission锁的MutiLock原理
为了提高redis的可用性我们会搭建集群或者主从现在以主从为例
此时我们去写命令写在主机上 主机会将数据同步给从机但是假设在主机还没有来得及把数据写入到从机去的时候此时主机宕机哨兵会发现主机宕机并且选举一个slave变成master而此时新的master中实际上并没有锁信息此时锁信息就已经丢掉了。 为了解决这个问题redission提出来了MutiLock锁使用这把锁咱们就不使用主从了每个节点的地位都是一样的 这把锁加锁的逻辑需要写入到每一个主丛节点上只有所有的服务器都写入成功此时才是加锁成功假设现在某个节点挂了那么他去获得锁的时候只要有一个节点拿不到都不能算是加锁成功就保证了加锁的可靠性。 那么MutiLock 加锁原理是什么呢笔者画了一幅图来说明
当我们去设置了多个锁时redission会将多个锁添加到一个集合中然后用while循环去不停去尝试拿锁但是会有一个总共的加锁时间这个时间是用需要加锁的个数 * 1500ms 假设有3个锁那么时间就是4500ms假设在这4500ms内所有的锁都加锁成功 那么此时才算是加锁成功如果在4500ms有线程加锁失败则会再次去进行重试. 大功告成
