P1 Redis企业实战课程介绍
P2 短信登录 导入黑马点评项目
首先在数据库连接下新建一个数据库hmdp,然后右键hmdp下的表,选择运行SQL文件,然后指定运行文件hmdp.sql即可(建议MySQL的版本在5.7及以上):
下面这个hm-dianping文件是项目源码。在IDEA中打开。
记得要修改数据库连接和Redis连接的密码:
运行程序后尝试访问:localhost:8081/shop-type/list 进行简单测试:
将nginx文件复制到一个没有中文路径的目录,然后点击nginx.exe运行:
在nginx所在目录打开CMD窗口,输入命令:start nginx.exe
访问:localhost:8080,选择用手机模式看,可以看到具体的页面:
P3 短信登录 基于session实现短信登录的流程
点击发送验证码可以看到验证码发送成功:
P4 短信登录 实现发送短信验证码功能
controller/UserController中写入如下代码:
@PostMapping("code")
public Result sendCode(@RequestParam("phone") String phone, HttpSession session) {
//发送短信验证码并保存验证码
return userService.sendCode(phone,session);
}
service/IUserService中写入如下代码:
public interface IUserService extends IService<User> {
Result sendCode(String phone, HttpSession session);
}
service/impl/UserServiceImpl中写入如下代码:
@Service
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements IUserService {
@Override
public Result sendCode(String phone, HttpSession session) {
//校验手机号
if(RegexUtils.isPhoneInvalid(phone)){
//不符合
return Result.fail("手机号格式错误");
}
//生成验证码
String code = RandomUtil.randomNumbers(6);
//保存验证码到session
session.setAttribute("code",code);
//发送验证码
log.debug("发送短信验证码成功,验证码:"+code);
return Result.ok();
}
}
P5 短信登录 实现短信验证码登录和注册功能
service/impl/UserServiceImpl的UserServiceImpl中写入如下代码:
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
String phone = loginForm.getPhone();
//校验手机
if(RegexUtils.isPhoneInvalid(phone)){
return Result.fail("手机号格式错误");
}
//校验验证码
Object cacheCode = session.getAttribute("code");
String code = loginForm.getCode();
if(cacheCode==null || !cacheCode.toString().equals(code)){
//不一致,报错
return Result.fail("验证码错误");
}
//一致根据手机号查用户
User user = query().eq("phone", phone).one();
//判断用户是否存在
if(user==null){
//不存在,创建用户并保存
user = createUserWithPhone(loginForm.getPhone());
}
//保存用户信息到session
session.setAttribute("user",user);
return null;
}
private User createUserWithPhone(String phone){
//1.创建用户
User user = new User();
user.setPhone(phone);
user.setNickName(USER_NICK_NAME_PREFIX+RandomUtil.randomString(10));
//2。保存用户
save(user);
return user;
}
前端点击发送验证码,后端直接把验证码摘抄后输入:
勾选协议然后确定登录,出现如下代码:
然后看到数据库后台记录已更新:
P6 短信登录 实现登录校验拦截器
preHandle前置拦截:
postHandle后置拦截:
afterCompletion视图渲染之后返回给用户之前:
在utils下面编写一个LoginInterceptor类,实现preHandle和afterCompletion这两个方法(这里User和UserDto的问题,我推荐的是统一使用UserDto,采用BeanUtils里的copy方法即可):
public class LoginInterceptor implements HandlerInterceptor {
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
//获取session
HttpSession session = request.getSession();
//获取用户
User user = (User) session.getAttribute("user");
//判断用户是否存在
if(user==null){
response.setStatus(401);
return false;
}
UserDTO userDTO = new UserDTO();
BeanUtils.copyProperties(user,userDTO);
//存在,保存用户信息的ThreadLocal
UserHolder.saveUser(userDTO);
//放行
return true;
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
//移除用户
UserHolder.removeUser();
}
}
在config下面创建一个MvcConfig类:
通过addInterceptors方法来添加拦截器,registry是拦截器的注册器。
用.excludePathPatterns来排除不需要拦截的路径。在这里code、login、bloghot、shop、shopType、upload和voucher等都不需要拦截。
@Configuration
public class MvcConfig implements WebMvcConfigurer {
@Override
public void addInterceptors(InterceptorRegistry registry){
registry.addInterceptor(new LoginInterceptor())
.excludePathPatterns(
"/user/code",
"/user/login",
"/upload/**",
"/blog/hot",
"/shop/**",
"/shop-type/**",
"/voucher/**"
);
}
}
输入手机号码点击获取验证码,写入返回后端的验证码,勾选协议之后,登录会直接返回首页,此时看我的个人主页没问题:
P7 短信登录 隐藏用户敏感信息
在P6已将User转为UserDTO返回给前端。
P8 短信登录 session共享的问题分析
多台Tomcat并不共享session存储空间,当请求切换不同Tomcat服务器时会导致数据丢失的问题。
session的替代方案应该满足:1.数据共享。2.内存存储。3.key、value结构。
P9 短信登录 Redis代替session的业务流程
想要保存用户的登录信息有2种方法:1.用String类型。2.用Hash类型。
String类型是以JSON字符串格式来保存,比较简单直观,但是占用内存比较多(因为有name和age这类的json格式):
Hash结构可以将对象中的每个字段独立存储,可以针对单个字段做CRUD,并且内存占用更少:
以随机的token作为key来存储用户的数据,token是用一个随机的字符串。
P10 短信登录 基于Redis实现短信登录
在UserServiceImpl中写入如下代码(调用StringRedisTemplate中的set方法进行数据插入,最好在key的前面加入业务前缀以示区分,形成区分):
@Resource
private StringRedisTemplate stringRedisTemplate;
在sendCode这个方法里将保存验证码的代码替换为下面:
//保存验证码到redis
stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY+phone,code,LOGIN_CODE_TTL, TimeUnit.MINUTES);
在login这个方法里进行如下2处修改:
首先是校验验证码:
//校验验证码
String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);
然后是添加把用户信息添加到Redis的逻辑:
//7.保存用户信息到redis----------------
//7.1 随机生成Token作为登录令牌
String token = UUID.randomUUID().toString(true);
//7.2 将User对象转为Hash存储
UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
Map<String, Object> userMap = BeanUtil.beanToMap(userDTO);
//7.3 存储
stringRedisTemplate.opsForHash().putAll("login:token:"+token,userMap);
//7.4设置token有效期
String tokenKey = LOGIN_USER_KEY+token;
stringRedisTemplate.expire(tokenKey,LOGIN_USER_TTL,TimeUnit.MINUTES);
return Result.ok(token);
在MvcConfig类上有@Configuration注解,说明是由Spring来负责依赖注入。
在MvcConfig类中要编写如下的代码:
@Configuration
public class MvcConfig implements WebMvcConfigurer {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public void addInterceptors(InterceptorRegistry registry){
registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))
.excludePathPatterns(
"/user/code",
"/user/login",
"/upload/**",
"/blog/hot",
"/shop/**",
"/shop-type/**",
"/voucher/**"
);
}
}
在utils下的LoginInterceptor中写入如下代码:
public class LoginInterceptor implements HandlerInterceptor {
@Resource
private StringRedisTemplate stringRedisTemplate;
public LoginInterceptor(StringRedisTemplate stringRedisTemplate){
this.stringRedisTemplate = stringRedisTemplate;
}
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
//TODO;1.获取请求头中的token
String token = request.getHeader("authorization");
if(StrUtil.isBlank(token)){
//不存在,拦截,返回401状态码
response.setStatus(401);
return false;
}
//TODO:2.基于TOKEN获取redis的用户
Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(LOGIN_USER_KEY + token);
//判断用户是否存在
if(userMap.isEmpty()){
//不存在,拦截,返回401状态码
response.setStatus(401);
return false;
}
//TODO:3.将查询到的Hash数据转化为UserDTO对象
UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
//TODO:4.存在,保存用户信息的ThreadLocal
UserHolder.saveUser(userDTO);
//TODO:5.刷新token有效期
stringRedisTemplate.expire(LOGIN_USER_KEY + token,RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);
//放行
return true;
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
//移除用户
UserHolder.removeUser();
}
}
测试:首先把Redis和数据库都启动。 原始的项目的Redis的服务器ID需要更改为自己的。点击发送验证码,redis中有记录,没问题:
但点击登录的时候会报一个无法将Long转String的错误。因为用的是stringRedisTemplate要求所有的字段都是string类型的。
需要对UserServiceImpl中如下的位置进行修改:
Map<String, Object> userMap = BeanUtil.beanToMap(userDTO,new HashMap<>(),
CopyOptions.create()
.setIgnoreNullValue(true)
.setFieldValueEditor((fieldName,fieldValue)->fieldValue.toString()));
效果如下:
P11 短信登录 解决状态登录刷新问题
现在只有在用户访问拦截器拦截的页面才会刷新页面,假如用户访问的是不需要拦截的页面则不会导致页面的刷新。
现在的解决思路是:新增一个拦截器,拦截一切路径。
复制LoginInterceptor变成一份新的RefreshTokenInterceptor,把下面几处地方改为return true即可:
LoginInterceptor的代码变成如下:
public class LoginInterceptor implements HandlerInterceptor {
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
//1.判断是否需要拦截(ThreadLocal中是否有用户)
if(UserHolder.getUser()==null){
//没有,需要拦截,设置状态码
response.setStatus(401);
//拦截
return false;
}
//放行
return true;
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
//移除用户
UserHolder.removeUser();
}
}
现在还需要在MvcConfig里面对拦截器进行更新配置,需要(用order)调整拦截器的执行顺序:
@Configuration
public class MvcConfig implements WebMvcConfigurer {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public void addInterceptors(InterceptorRegistry registry){
registry.addInterceptor(new LoginInterceptor())
.excludePathPatterns(
"/user/code",
"/user/login",
"/upload/**",
"/blog/hot",
"/shop/**",
"/shop-type/**",
"/voucher/**"
).order(1);
registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate))
.addPathPatterns("/**").order(0);
}
}
P12 什么是缓存
缓存就是数据交换的缓冲区,是存储数据的临时地方,一般读写性能较高。
缓存作用:降低后端负载;提高读写的效率,降低响应时间。
缓存成本:数据一致性成本(数据库里的数据如果发生变化,容易与缓存中的数据形成不一致)。代码维护成本高(搭建集群)。运营成本高。
P13 添加商户缓存
在ShopController类的queryShopById方法中:
@GetMapping("/{id}")
public Result queryShopById(@PathVariable("id") Long id) {
return Result.ok(shopService.queryById(id));
}
在IShopService接口中编写如下代码:
public interface IShopService extends IService<Shop> {
Object queryById(Long id);
}
在ShopServiceImpl类的queryById方法中编写具体代码:
@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Object queryById(Long id) {
String key = CACHE_SHOP_KEY + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isNotBlank(shopJson)){
//3.存在,直接返回
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
//4.不存在,根据id查询数据库
Shop shop = getById(id);
//5.不存在,返回错误
if(shop==null){
return Result.fail("店铺不存在!");
}
//6.存在,写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));
return Result.ok(shop);
}
}
核心是通过调用hutool工具包中的JSONUtil类来实现对象转JSON(方法:toJsonStr(对象))和JSON转对象(方法:toBean(json,Bean的类型))。
P14 缓存练习题分析
TODO:对分类进行缓存。
P15 缓存更新策略
主动更新:编写业务逻辑,在修改数据库的同时,更新缓存。
适用于高一致性的需求:主动更新,以超时剔除作为兜底方案。
主动更新策略:
1.由缓存的调用者,在更新数据库的同时更新缓存。(一般情况下使用该种方案)
2.缓存与数据库聚合为一个服务,由服务来维护一致性。调用者调用该服务,无需关心缓存的一致性问题。
3.调用者只操作缓存,由其它线程异步的将缓存数据持久化到数据库,保证最终一致。
对1进行分析:
1.选择删除缓存还是更新缓存?如果是更新缓存:每次更新数据库都会更新缓存,无效的写操作比较多。删除缓存:更新数据库时让缓存失效,查询时再更新缓存。
2.如何保证缓存与数据库的操作的同时成功或失败?
单体系统:将缓存与数据库操作放在一个事务。
分布式系统:利用TCC等分布式事务方案。
3.先操作缓存还是先操作数据库?
先删缓存,再操作(写)数据库:
先操作(写)数据库,再删除缓存(出现的概率比较低)
要求线程1来查询的时候缓存恰好失效了->在写入缓存的时候突然来了线程2,对数据库的数据进行了修改->此时线程1写回缓存的是旧数据。
P16 实现商铺缓存与数据库的双写一致
给查询商铺的缓存添加超时剔除和主动更新的策略。
修改ShopController中的业务逻辑,满足下面要求:
1.根据id查询商铺时,如果缓存未命中,则查询数据库,将数据库结果写入缓存,并设置超时时间。
2.根据id修改店铺时,先修改数据库,再删除缓存。
首先修改ShopServiceImpl的redis过期时间:
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);
修改ShopController中的updateShop方法:
@PutMapping
public Result updateShop(@RequestBody Shop shop) {
// 写入数据库
return Result.ok(shopService.update(shop));
}
向IShopService接口中添加update方法:
Object update(Shop shop);
向ShopServiceImpl类中添加update方法:
@Override
public Object update(Shop shop) {
Long id = shop.getId();
if(id == null){
return Result.fail("商铺id不存在");
}
updateById(shop);
stringRedisTemplate.delete(CACHE_SHOP_KEY + id);
return Result.ok();
}
首先删除缓存中的数据,然后看SQL语句是否执行,是否加上了TTL过期时间。
在PostMan中访问http://localhost:8081/shop,然后修改101茶餐厅为102茶餐厅:
注意要发送的是PUT请求,请求的内容如下:
{
"area": "大关",
"openHours": "10:00-22:00",
"sold": 4215,
"address": "金华路锦昌文华苑29号",
"comments": 3035,
"avgPrice": 80,
"score": 37,
"name": "102茶餐厅",
"typeId": 1,
"id": 1
}
然后去数据库看是否名称更新为102茶餐厅,然后看缓存中的数据是否被删除,用户刷新页面看到102茶餐厅,缓存中会有最新的数据。
P17 缓存穿透的解决思路
缓存穿透指的是客户端请求的数据在缓存中和数据库中都不存在,使得缓存永远不会生效,请求都会打到数据库。
2种解决方法:
1.缓存空对象。优点:实现简单,维护方便。缺点:额外的内存消耗。可能造成短期的不一致(可以设置TTL)。
2.布隆过滤。在客户端和Redis之间加个布隆过滤器(存在不一定存在,不存在一定不存在,有5%的错误率)。
优点:内存占用较少,没有多余key。缺点:实现复杂,存在误判可能。
P18 编码解决商铺查询的缓存穿透问题
下图是原始的:
下面是更改后的:
在ShopServiceImpl类里对queryById方法进行修改:
@Override
public Object queryById(Long id) {
String key = CACHE_SHOP_KEY + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isNotBlank(shopJson)){
//3.存在,直接返回
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
//上面是有值的情况,下面是无值的2种情况:A:空字符串。B:null。
if(shopJson != null){
return Result.fail("店铺信息不存在!");
}
//4.不存在,根据id查询数据库
Shop shop = getById(id);
//5.不存在,返回错误
if(shop==null){
stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
return Result.fail("店铺不存在!");
}
//6.存在,写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);
return Result.ok(shop);
}
测试:
localhost:8080/api/shop/1此时是命中数据。
localhost:8080/api/shop/0此时未命中数据。打开缓存可以看到缓存的是空,并且TTL是200秒。
总结缓存穿透:用户请求的数据在缓存中和数据库中都不存在,不断发起请求,会给数据库造成巨大压力。
缓存穿透:缓存null值和布隆过滤器。还可以增强id的复杂度,避免被猜测id规律。做好数据的基础格式校验。加强用户权限校验。做好热点参数的限流。
P19 缓存雪崩问题及解决思路
缓存雪崩:是指在同一时段大量的缓存key同时失效或者Redis服务宕机,导致大量请求打到数据库,带来巨大的压力。
解决方案:
1.(解决大量缓存key同时失效)给不同Key的TTL添加随机值。
2.(解决Redis宕机)利用Redis集群提高服务的可用性。
3.给缓存业务添加降级限流策略。
4.给业务添加多级缓存(浏览器可以有缓存,nginx可以有缓存,redis可以有缓存,数据库可以有缓存)。
P20 缓存击穿问题及解决方案
缓存击穿问题:也叫热点key问题,就是一个被高并发访问并且缓存重建业务较复杂的key突然消失了,无数的请求访问在瞬间给数据库带来巨大的冲击。
解决方案:
1.互斥锁。由获取互斥锁成功的线程来查询数据库重建缓存数据。缺点:未获得互斥锁的线程需要等待,性能略差。
2.逻辑过期。设置一个逻辑时间字段,查询缓存的时候检查逻辑时间看是否已过期。如果某个线程获取到互斥锁就开启新线程,由新线程查询数据库重建缓存数据。
其它线程在获取互斥锁失败后不会等待,而是直接返回过期的数据。只有当缓存重建完毕之后释放锁,新线程才会读到最新的数据。
互斥锁优点:
互斥锁没有额外的内存消耗:因为逻辑过期需要维护一个逻辑过期的字段,有额外内存消耗。
互斥锁可以保证强一致性,所有线程拿到的是最新数据。实现也很简单。
互斥锁缺点:
线程需要等待,性能受到影响。可能会有死锁的风险。
逻辑过期优点:
线程无需等待,性能较好。
逻辑过期缺点:
不保证一致性。有额外内存消耗。实现复杂。
P21 利用互斥锁解决缓存击穿问题
在ShopServiceImpl类中定义一个tryLock方法(在Redis中的setnx相当于setIfAbsent方法。)
public boolean tryLock(String key){
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
return BooleanUtil.isTrue(flag);
}
在ShopServiceImpl类中定义一个unLock方法用于解锁。
public void unLock(String key){
stringRedisTemplate.delete(key);
}
在ShopServiceImpl类中定义一个queryWithPassThrough方法。
public Shop queryWithPassThrough(Long id){
String key = CACHE_SHOP_KEY + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isNotBlank(shopJson)){
//3.存在,直接返回
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
return shop;
}
//上面是有值的情况,下面是无值的2种情况:A:空字符串。B:null。
if(shopJson != null){
return null;
}
//4.不存在,根据id查询数据库
Shop shop = getById(id);
//5.不存在,返回错误
if(shop==null){
stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
return null;
}
//6.存在,写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);
return shop;
}
在ShopServiceImpl类中定义一个queryWithMutex方法:
public Shop queryWithMutex(Long id){
String key = CACHE_SHOP_KEY + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isNotBlank(shopJson)){
//3.存在,直接返回
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
return shop;
}
//上面是有值的情况,下面是无值的2种情况:A:空字符串。B:null。
if(shopJson != null){
return null;
}
//4.实现缓存重建
//4.1 获取互斥锁
String lockKey = LOCK_SHOP_KEY+id;
Shop shop = null;
try {
boolean isLock = tryLock(lockKey);
//4.2 判断是否获取成功
if(!isLock){
//4.3 失败,则休眠并重试
Thread.sleep(50);
return queryWithMutex(id);
}
//4.4 获取互斥锁成功,根据id查询数据库
shop = getById(id);
//模拟重建的延时
Thread.sleep(200);
//5.数据库查询失败,返回错误
if(shop==null){
stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
return null;
}
//6.存在,写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),CACHE_SHOP_TTL, TimeUnit.MINUTES);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}finally {
//7.释放互斥锁
unLock(lockKey);
}
//8.返回
return shop;
}
在ShopServiceImpl类中修改queryById,调用queryWithMutex:
public Object queryById(Long id) {
//缓存穿透
//Shop shop = queryWithPassThrough(id);
//互斥锁解决缓存击穿
Shop shop = queryWithMutex(id);
return Result.ok(shop);
}
测试:
定义1000个线程,Ramp-Up时间为5。
请求地址:localhost:8081/shop/1。
设置完毕后点击绿色箭头运行,此时会提示是否保存测试文件,选择不保存(我测试选择保存会报错)。
可以在结果树这里看请求是否发送成功:
先删掉缓存,然后点击绿色箭头发送并发请求,可以发现所有线程请求成功,控制台对数据库的查询只有1次(没有出现多个线程争抢查询数据库的情况),测试成功。
P22 利用逻辑过期解决缓存击穿问题
如何添加逻辑过期字段?答:可以在utils包下定义RedisData类(可以让Shop继承RedisData类),也可以在RedisData中设置一个Shop类的data属性:
@Data
public class RedisData {
private LocalDateTime expireTime;
private Object data;
}
在ShopServiceImpl类中定义saveShop2Redis方法:
public void saveShop2Redis(Long id,Long expireSeconds){
//1.查询店铺数据
Shop shop = getById(id);
//2.封装逻辑过期时间
RedisData redisData = new RedisData();
redisData.setData(shop);
redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));
//3.写入Redis
stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY+id,JSONUtil.toJsonStr(redisData));
}
单元测试,在test包下的HmDianPingApplicationTests中创建testSaveShop类写入测试代码(这里要注意的是输入alt+insert之后选择Test Method要选择Junit 5来进行测试方法的编写):
@SpringBootTest
class HmDianPingApplicationTests {
@Resource
private ShopServiceImpl shopService;
@Test
void testSaveShop() {
shopService.saveShop2Redis(1L,10L);
}
}
可以看到redis中确实存入了数据:
在ShopServiceImpl中复制一份缓存穿透的代码,更改名称为queryWithLogicalExpire:
private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
public Shop queryWithLogicalExpire(Long id){
String key = CACHE_SHOP_KEY + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isBlank(shopJson)){
//3.不存在,返回空
return null;
}
//4.命中,需要先把json反序列化为对象
RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
JSONObject data = (JSONObject) redisData.getData();
Shop shop = JSONUtil.toBean(data, Shop.class);
//5.判断是否过期
//5.1 未过期直接返回店铺信息
LocalDateTime expireTime = redisData.getExpireTime();
if(expireTime.isAfter(LocalDateTime.now())){
return shop;
}
//5.2 已过期重建缓存
//6.缓存重建
//6.1.获取互斥锁
String lockKey = LOCK_SHOP_KEY + id;
boolean isLock = tryLock(lockKey);
//6.2.判断是否获取互斥锁成功
if(isLock){
//6.3.成功,开启独立线程,实现缓存重建
CACHE_REBUILD_EXECUTOR.submit(()->{
try {
saveShop2Redis(id,20L); //实际中应该设置为30分钟
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
unLock(lockKey);
}
});
}
//6.4.失败,返回过期的商铺信息
return shop;
}
测试:
先到数据库把102茶餐厅改为103茶餐厅(因为Redis之前插入了一条缓存为102茶餐厅,并且已经过期,此时数据库与缓存不一致),新的HTTP请求会将逻辑过期的数据删除,然后更新缓存。
线程数设置为100,Ramp-up时间设置为1
在查看结果树里面到中间某个HTTP请求会完成重建,响应数据会改变。
1.安全性问题:在高并发情况下是否会有很多线程来做重建。
2.一致性问题:在重建完成之前得到的是否是旧的数据。
P23 封装Redis工具类
在utils包下创建CacheClient类,先写入如下基础的代码:
@Slf4j
@Component
public class CacheClient {
private final StringRedisTemplate stringRedisTemplate;
public CacheClient(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
public void set(String key, Object value, Long time, TimeUnit unit){
stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(value),time,unit);
}
public void setWithLogicalExpire(String key, Object value,Long expire,TimeUnit unit){
//设置逻辑过期
RedisData redisData = new RedisData();
redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(expire)));
redisData.setData(value);
stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(redisData));
}
}
在CacheClient类中编写缓存穿透的共性方法queryWithPassThrough:
public <R,ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type,
Function<ID,R> dbFallBack,Long time,TimeUnit unit){
String key = keyPrefix + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isNotBlank(shopJson)){
//3.存在,直接返回
return JSONUtil.toBean(shopJson, type);
}
//上面是有值的情况,下面是无值的2种情况:A:空字符串。B:null。
if(shopJson != null){
return null;
}
//4.不存在,根据id查询数据库
R r = dbFallBack.apply(id);
//5.不存在,返回错误
if(r==null){
stringRedisTemplate.opsForValue().set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
return null;
}
//6.存在,写入Redis
this.set(key,r,time,unit);
return r;
}
编写完queryWithPassThrough之后可以到ShopServiceImpl中直接调用新的方法(记得引入CacheClient类):
@Resource
private CacheClient cacheClient;
@Override
public Object queryById(Long id) {
//调用工具类解决缓存击穿
Shop shop = cacheClient.queryWithPassThrough(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);
if(shop==null){
return Result.fail("店铺不存在!");
}
return Result.ok(shop);
}
进行测试:成功会对不存在的店铺空值进行缓存。
接下来拷贝queryWithLogicalExpire的代码到CacheClient类中进行改写:
private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
public <R,ID> R queryWithLogicalExpire(String keyPrefix,ID id,Class<R> type,Function<ID,R> dbFallBack,Long time,TimeUnit unit){
String key = keyPrefix + id;
//1.从Redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if(StrUtil.isBlank(shopJson)){
//3.不存在,返回空
return null;
}
//4.命中,需要先把json反序列化为对象
RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
JSONObject data = (JSONObject) redisData.getData();
R r = JSONUtil.toBean(data, type);
//5.判断是否过期
//5.1 未过期直接返回店铺信息
LocalDateTime expireTime = redisData.getExpireTime();
if(expireTime.isAfter(LocalDateTime.now())){
return r;
}
//5.2 已过期重建缓存
//6.缓存重建
//6.1.获取互斥锁
String lockKey = LOCK_SHOP_KEY + id;
boolean isLock = tryLock(lockKey);
//6.2.判断是否获取互斥锁成功
if(isLock){
//6.3.成功,开启独立线程,实现缓存重建
CACHE_REBUILD_EXECUTOR.submit(()->{
try {
//查询数据库
R r1 = dbFallBack.apply(id);
//写入redis
this.setWithLogicalExpire(key,r1,time,unit);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
unLock(lockKey);
}
});
}
//6.4.失败,返回过期的商铺信息
return r;
}
public boolean tryLock(String key){
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
return BooleanUtil.isTrue(flag);
}
public void unLock(String key){
stringRedisTemplate.delete(key);
}
改写test下的HmDianPingApplicationTests类:
@SpringBootTest
class HmDianPingApplicationTests {
@Resource
private CacheClient cacheClient;
@Resource
private ShopServiceImpl shopService;
@Test
void testSaveShop() throws InterruptedException {
Shop shop = shopService.getById(1L);
cacheClient.setWithLogicalExpire(CACHE_SHOP_KEY+1L,shop,10L,TimeUnit.SECONDS);
}
}
测试:首先运行HmDianPingApplicationTests类里的测试方法,10秒后逻辑过期,此时运行后台程序,修改数据库1号商铺的name字段,此时访问:localhost:8080/api/shop/1 会出现效果第1次访问为缓存旧值,然后发现缓存过期开始重建,第2次访问开始就是新值。数据库也只有1次重建。
P24 缓存总结
P25 优惠券秒杀 全局唯一ID
每个店铺都可以发布优惠券,当用户抢购时,就会生成订单并保存到tb_voucher_order这张表中,而订单表如果使用数据库自增ID会存在一些问题。
1.id的规律性太明显。
2.受单表数据量的限制(分表之后每张表都自增长,id会出现重复)。
全局ID生成器:是一种在分布式系统下用来生成全局唯一ID的工具。
要求全局唯一ID生成器满足如下几点:1.唯一性。2.高可用。3.高性能。4.递增性。5.安全性。
为了增加ID的安全性,我们可以不直接使用Redis自增的数值,而是拼接一些其它信息。
符号位永远为0代表整数。
31位的时间戳是以秒为单位,定义了一个起始时间,用当前时间减起始时间,预估可以使用69年。
32位的是序列号是Redis自增的值,支持每秒产生2^32个不同ID。
P26 优惠券秒杀 Redis实现全局唯一id
在utils包下定义一个RedisWorker类,是一个基于Redis的ID生成器。
如果只使用一个key来自增记录有一个坏处,最终key的自增数量会突破容量的上限,假如自增超过32位彼时便无法再存储新的数据,解决的方案是采用拼接日期。
@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 timeStamp = now.toEpochSecond(ZoneOffset.UTC) - 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;
}
}
在HmDianPingApplicationTests中写入如下的测试代码:
@Resource
private ShopServiceImpl shopService;
@Resource
private RedisIdWorker redisIdWorker;
private ExecutorService es = Executors.newFixedThreadPool(500);
@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("Result Time = " + (end-begin));
}
运行之后可以看到以十进制输出的所有编号:
可以在Redis中看到自增长的结果,1次是30000:
大概2秒可以生成3万条,速度还是可以的。
全局唯一ID生成策略:
1.UUID利用JDK自带的工具类即可生成,生成的是16进制的字符串,无单调递增的特性。
2.Redis自增(每天一个key,方便统计订单量。时间戳+计数器的格式。)
3.snowflake雪花算法(不依赖于Redis,性能更好,对于时钟依赖)
4.数据库自增
P27 优惠券秒杀 添加优惠券
每个店铺都可以发放优惠券,分为平价券和特价券。平价券可以任意抢购,特价券需要秒杀抢购。
tb_voucher:优惠券基本信息,优惠金额,使用规则等。
tb_seckill_voucher:优惠券的库存,开始抢购时间,结束抢购时间,只有特价优惠券才需要填写这些信息。
请求的信息如下可自行复制(注意beginTime和endTime需要修改):
{
"shopId":1,
"title":"100元代金券",
"subTitle":"周一至周五均可使用",
"rules":"全场通用\\n无需预约\\n可无限叠加\\不兑现、不找零\\n仅限堂食",
"payValue":8000,
"actualValue":10000,
"type":1,
"stock":100,
"beginTime":"2024-04-10T10:09:17",
"endTime":"2024-04-11T12:09:04"
}
注意要在请求头中带Authorization参数否则会报401(登录后进入“我的”页面,看网络包有Authorization的值):
以如下格式发送请求:
首先在tb_voucher表中可以看到新增的优惠券:
在tb_seckill_voucher表中也可以看到秒杀优惠券的具体信息:
在前端也能看到新增的100元代金券,注意优惠券的时间一定要进行更改,如果不在开始和结束时间区间内优惠券会处于下架状态是看不到的。
P28 优惠券秒杀 实现秒杀下单
首先要判断秒杀是否开始或结束,所以要先查询优惠券的信息,如果尚未开始或者已经结束无法下单。
要判断库存是否充足,如果不足则无法下单。
在VouchrOrderController类中:
@RestController
@RequestMapping("/voucher-order")
public class VoucherOrderController {
@Resource
private IVoucherService voucherService;
@PostMapping("seckill/{id}")
public Result seckillVoucher(@PathVariable("id") Long voucherId) {
return voucherService.seckillVoucher(voucherId);
}
}
在IVoucherOrderService中写入如下代码:
public interface IVoucherOrderService extends IService<VoucherOrder> {
Result seckillVoucher(Long voucherId);
}
在VoucherOrderServiceImpl中写入如下代码:
@Service
@Transactional
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resource
private ISeckillVoucherService seckillVoucherService;
@Resource
private RedisIdWorker redisIdWorker;
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券信息
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//2.判断秒杀是否开始
//2.1秒杀尚未开始返回异常
if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
return Result.fail("秒杀尚未开始");
}
//2.2秒杀已结束返回异常
if(voucher.getEndTime().isBefore(LocalDateTime.now())){
return Result.fail("秒杀已经结束");
}
//3.判断库存是否充足
if(voucher.getStock()<1){
//3.1库存不足返回异常
return Result.fail("库存不足!");
}
//3.2库存充足扣减库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId).update();
if(!success){
return Result.fail("库存不足!");
}
//4.创建订单,返回订单id
VoucherOrder voucherOrder = new VoucherOrder();
long orderId = redisIdWorker.nextId("order");//订单id
voucherOrder.setId(orderId);
Long userId = UserHolder.getUser().getId();//用户id
voucherOrder.setUserId(userId);
voucherOrder.setVoucherId(voucherId);//代金券id
save(voucherOrder);
return Result.ok(orderId);
}
}
测试:点击限时抢购之后会提示抢购成功。
P29 优惠券秒杀 库存超卖问题分析
Jmeter的配置如下:
注意Authorization要事先登录获取:
下面是结果:
发现tb_seckill_voucher中库存为-9,在tb_voucher_order中插入了109条数据,说明出现了超卖的问题。
正常逻辑:
非正常逻辑:
超卖问题是典型的多线程安全问题,针对这一问题的常见解决方案是加锁。
悲观锁:认为线程安全问题一定会发送,因此在操作数据之前要先获取锁,确保线程串行执行。像Synchronized、Lock都属于悲观锁。
乐观锁:认为线程安全问题不一定会发生,因此不加锁,只是在更新数据时去判断有没有其它线程对数据做了修改。
如果没有修改则认为是安全的,自己才更新数据。
如果已经被其它线程修改说明发生了安全问题,此时可以重试或异常。‘
乐观锁关键是判断之前查询得到的数据是否被修改过,常见的方法有2种:
1.版本号法:
2.CAS法(版本号法的简化版):查询的时候把库存查出来,更新的时候判断库存和之前查到的库存是否一致,如果一致则更新数据。
P30 优惠券秒杀 乐观锁解决超卖
只需加上下面这段代码即可:.eq("stock",voucher.getStock()) 。用于比较当前数据库的库存值和之前查询到的库存值是否相同,只有相同时才可以执行set语句。
//3.2库存充足扣减库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") //相当于set条件 set stock = stock - 1
.eq("voucher_id", voucherId) //相当于where条件 where id = ? and stock = ?
.eq("stock",voucher.getStock()).update();
但现在出现了异常值偏高的问题,正常的请求大约只占10%。
原理是因为:假如一次有30个线程涌入,查询到库存值为100,只有1个线程能把值改为99,其它29个线程比对库存值99发现和自己查询到的库存值100不同,所以都认为数据已经被修改过,所以都失败了。
乐观锁的问题,成功率太低。
现在只需要保证stock>0即可,只要存量大于0就可以任意扣减。
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") //相当于set条件 set stock = stock - 1
.eq("voucher_id", voucherId) //相当于where条件 where id = ? and stock = ?
.gt("stock",0).update();
乐观锁缺陷:
需要大量对数据库进行访问,容易导致数据库的崩溃。
总结:
P31 优惠券秒杀 实现一人一单功能
修改秒杀业务,要求同一个优惠券,一个用户只能下一单。
首先不建议把锁加在方法上,因为任何一个用户来了都要加这把锁,而且是同一把锁,方法之间变成串行执行,性能很差。
因此可以把锁加在用户id上,只有当id相同时才会对锁形成竞争关系。但是因为toString的内部是new了一个String字符串,每调一次toString都是生成一个全新的字符串对象,锁对象会变。
所以可以调用intern()方法,intern()方法会优先去字符串常量池里查找与目标字符串值相同的引用返回(只要字符串一样能保证返回的结果一样)。
但是因为事务是在函数执行结束之后由Spring进行提交,如果把锁加在createVoucherOrder内部其实有点小——因为如果解锁之后,其它线程可以进入,而此时事务尚未提交,仍然会导致安全性问题。
因此最终方案是把synchronized加在createVoucherOrder的方法外部,锁住的是用户id。
关于代理对象事务的问题:通常情况下,当一个使用了@Transactional注解的方法被调用时,Spring会从上下文中获取一个代理对象来管理事务。
但是如果加@Transactional方法是被同一个类中的另一个方法调用时,Spring不会使用代理对象,而是直接调用该方法,导致事务注解失效。
为避免这种情况,可以使用AopContext.currentProxy方法获取当前的代理对象,然后通过代理对象调用被@Transactional注解修饰的方法,确保事务生效。
在VoucherOrderServiceImpl中写入如下代码(注意:ctrl+alt+m可以把含有return的代码段进行提取):
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resource
private ISeckillVoucherService seckillVoucherService;
@Resource
private RedisIdWorker redisIdWorker;
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券信息
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//2.判断秒杀是否开始
//2.1秒杀尚未开始返回异常
if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
return Result.fail("秒杀尚未开始");
}
//2.2秒杀已结束返回异常
if(voucher.getEndTime().isBefore(LocalDateTime.now())){
return Result.fail("秒杀已经结束");
}
voucher = seckillVoucherService.getById(voucherId);
//3.判断库存是否充足
if(voucher.getStock()<1){
//3.1库存不足返回异常
return Result.fail("库存不足!");
}
Long userId = UserHolder.getUser().getId();
synchronized (userId.toString().intern()){
//获取代理对象
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
}
@Transactional
public Result createVoucherOrder(Long voucherId) {
//6.一人一单
Long userId = UserHolder.getUser().getId();
//6.1查询订单
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
//6.2判断是否存在
if(count>0){
//用户已经购买过了
return Result.fail("用户已经购买过一次!");
}
//3.2库存充足扣减库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") //相当于set条件 set stock = stock - 1
.eq("voucher_id", voucherId) //相当于where条件 where id = ? and stock = ?
.gt("stock",0).update();
if(!success){
return Result.fail("库存不足!");
}
//4.创建订单,返回订单id
VoucherOrder voucherOrder = new VoucherOrder();
long orderId = redisIdWorker.nextId("order");//订单id
voucherOrder.setId(orderId);
voucherOrder.setUserId(userId);
voucherOrder.setVoucherId(voucherId);//代金券id
save(voucherOrder);
return Result.ok(orderId);
}
}
在IVoucherOrderService接口中加入下面这个方法:
Result createVoucherOrder(Long voucherId);
在pom.xml中引入如下的依赖:
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
</dependency>
在启动类HmDianPingApplication上加如下注解:
@EnableAspectJAutoProxy(exposeProxy = true)
测试: 成功实现一名用户只能领取一张优惠券。
P32 优惠券秒杀 集群下的线程并发安全问题
本P主要是为了验证在集群下synchronized并不能保证线程的并发安全。
如下图可以设置项目启动的端口号,确保启动的项目之间端口号不同:
在nginx.conf中放开8082的这个配置:
向下面这个页面发送请求:
http://localhost:8080/api/voucher/list/1
可以看到请求会分别被8082和8081接收,是轮询的效果:
首先到tb_voucher_order把之前的订单删除,到tb_seckill_voucher中把stock重新改回100。
准备2个相同的秒杀请求:要注意请求的地址是:http://localhost:8080/api/voucher-order/seckill/13
我这里直接用Jemeter来进行测试,模拟高并发场景:
下面是效果:可以看到并发请求能够同时进入集群的每台结点。
正常情况:
在集群模式下,每一个节点都是一个全新的JVM,每个JVM都有自己的锁。锁监视器只能在当前JVM的范围内,监视线程实现互斥。
现在就要实现让多个JVM使用的是同一把锁。跨JVM、跨进程的锁。
P33 分布式锁 基本原理和不同实现方式对比
synchronized只能保证单个JVM内部的多个线程之间的互斥,而没法让集群下多个JVM进程间的线程互斥。
要让多个JVM进程能看到同一个锁监视器,而且同一时间只有一个线程能拿到锁监视器。
所以必须使用分布式锁,分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。
分布式锁要满足:多进程可见+互斥+高可用+高性能+安全性。
分布式锁可以通过MySQL或Redis或Zookeeper来实现。
MySQL:
1.互斥:是利用mysql本身的互斥锁机制。在执行写操作的时候,MySQL会自动分配一个互斥的锁。
2.可用性:好。3.性能:受限于MySQL性能。
4.安全性:事务机制,如果断开连接,会自动释放锁。
Redis:
1.互斥:利用setnx这样的互斥命令。往Redis里set数据只有不存在时才能set成功。
2.可用性:好,Redis支持主从和集群。3.性能:好。
4.安全性:如果没有执行删除key的操作,key不会自动释放。但可以利用锁的超时机制,到期自动释放。
Zookeeper:
1.利用节点的唯一性(节点不重复)和有序性(节点递增)实现互斥。利用有序性:id最小的节点获取锁成功;释放锁只需要删除id最小的节点。
2.可用性:好。3.性能:比Redis差,一般,强调强一致性,主从间同步需要时间。
4.安全性:好。因为是临时节点,断开连接会自动释放。
P34 分布式锁 Redis的分布式锁实现思路
假如获取锁后宕机,锁无法释放——>可以添加超时过期时间。
为了防止锁在SETEX和EXPIRE之间过期,可以直接用一条命令(原子操作)来实现设置过期时间(EX)和只有lock不存在时才能设置(NX)。
采用非阻塞式获取锁,如果成功返回true,失败返回false。
P35 分布式锁 实现Redis分布式锁版本1
在utils下面创建一个ILock接口:
public interface ILock {
//尝试获取锁
boolean tryLock(long timeoutSec);
//释放锁
void unlock();
}
在utils下面实现SimpleRedisLock类:
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:";
@Override
public 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);
}
@Override
public void unlock() {
//释放锁
stringRedisTemplate.delete(KEY_PREFIX+name);
}
}
更改VoucherOrderServiceImpl类中的seckillVoucher方法的代码:
@Resource
private RedisIdWorker redisIdWorker;
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Result seckillVoucher(Long voucherId) {
//1.查询优惠券信息
SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//2.判断秒杀是否开始
//2.1秒杀尚未开始返回异常
if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
return Result.fail("秒杀尚未开始");
}
//2.2秒杀已结束返回异常
if(voucher.getEndTime().isBefore(LocalDateTime.now())){
return Result.fail("秒杀已经结束");
}
voucher = seckillVoucherService.getById(voucherId);
//3.判断库存是否充足
if(voucher.getStock()<1){
//3.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();
}
}
经测试多台节点相同用户只能获取同一张优惠券成功:
P36 分布式锁 Redis分布式锁误删问题
假如某个线程(线程A)获取到锁之后,出现了业务阻塞,导致阻塞时间超过了锁自动释放的时间,锁因超时自动释放。此时其它线程(线程B)过来拿到了锁,开始执行业务。但线程A此时业务执行完毕,释放了锁,但释放的是线程B的锁。此时线程C过来看锁已被释放,趁虚而入拿到锁,此时线程B和线程C是并行执行。
要解决这个问题:线程在删除锁之前要先看锁是否是自己加的(获取锁的标示并判断是否一致)。
P37 分布式锁 解决Redis分布式锁误删问题
1.在获取锁时存入线程标示(可以用UUID表示)。
2.在释放锁时先获取锁中的线程标示,判断是否与当前线程标示一致(如果一致释放锁,如果不一致则不释放锁)。
首先要修改SimpleRedisLock里面的如下代码,主要是调用hutool工具包生成UUID(每次线程调用都会生成一个唯一的UUID),让Redis的前缀变成UUID+线程ID:
private static final String ID_PREFIX = UUID.fastUUID().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);
}
现在要修改的是SimpleRedisLock类里面的unlock方法,主要是比较当前线程的标示和Redis中锁的标示是否一致,只有标示一致才能释放锁:
@Override
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);
}
}
P38 分布式锁 分布式锁的原子性问题
现在假设出现了其它问题,比如线程1在判断完锁标示是否一致之后出现了阻塞(比如JVM垃圾回收FULL GC导致阻塞了过长时间),此时锁超时了,线程2趁虚而入获取了锁,此时线程1直接释放了线程2的锁,此时线程3趁虚而入继续给Redis加锁,此时会出现线程2和线程3并行执行。
根本的原因是:获取锁标示和释放锁的操作不是原子性的,现在要解决的问题就是将这两个操作变成原子性的。
P39 分布式锁 Lua脚本解决多条命令原子性问题
Redis提供Lua脚本功能,在一个脚本中编写多条Redis命令,确保多条命令执行时的原子性。
Lua是一种编程语言,它的基本语法可以参考网站:https://www.runoob.com/lua/lua-tutorial.html
执行脚本的方法:
执行一个写死的set命令:
在Lua语言里,数组的第一个元素下标是1。
P40 分布式锁 Java调用lua脚本改造分布式锁
繁琐版的Lua脚本内容如下:
-- 锁的key
local key = KEYS[1]
-- 当前线程标示
local threadId = ARGV[1]
--获取锁中的线程标示
local id = redis.call('get',key)
--比较线程标示与锁中的标示是否一致
if(id == threadId) then
--释放锁 del key
return redis.call('del',key)
end
return 0
简化版的Lua脚本内容如下:
--比较线程标示与锁中的标示是否一致
if(redis.call('get',KEYS[1]) == ARGV[1]) then
--释放锁 del key
return redis.call('del',KEYS[1])
end
return 0
在resources下创建unlock.lua,会提示下载一个plugins点击install,然后只需要下载一个EmmyLua即可,实测如果下载了多个Lua相关的插件会产生冲突,最终导致IDEA打不开,这真是血泪的教训!
在SimpleRedisLock中写入如下的代码,因为我们希望的是在一开始就将Lua的脚本加载好,而不是等到要调用释放锁的时候再去加载Lua脚本,所以采用静态变量和静态代码块,这些部分在类初始化的时候就会被加载:
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
static {
UNLOCK_SCRIPT = new DefaultRedisScript<>();
UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
UNLOCK_SCRIPT.setResultType(Long.class);
}
在SimpleRedisLock类的unlock方法中写入如下的代码:
@Override
public void unlock() {
stringRedisTemplate.execute(UNLOCK_SCRIPT,
Collections.singletonList(KEY_PREFIX+name),
ID_PREFIX + Thread.currentThread().getId());
}
在程序1和程序2的下面这个位置打上断点:
在测试API中测试访问如下的URL:
http://localhost:8080/api/voucher-order/seckill/14
分别测试秒杀优惠券1和2:
在Redis中能看到程序1获取锁成功,然后直接把lock锁删掉,模拟超时释放的情况:
然后让程序2往下走一步,可以看到程序2获取到了锁
然后可以直接放行程序1,会看到结果是程序2加的锁没有被删除。
最后放行程序2,会看到程序2加的锁被删除。
总结:
基于Redis的分布式锁的实现思路:
1.利用set nx ex获取锁,并设置过期时间,保存线程标示。
2.释放锁时先判断线程标示是否与自己一致,一致则删除锁。
特性:
1.利用set nx满足互斥性。
2.利用set nx保障故障时锁依然能够释放,避免死锁,提高安全性。
3.利用Redis集群保障高可用和高并发的特性。
P41 分布式锁 Redisson功能介绍
目前基于setnx实现的分布式锁存在以下几个问题:
1.不可重入:同一线程无法多次获取同一把锁。
2.不可重试:获取锁只尝试一次就返回false,没有重试机制。
3.超时释放:锁超时释放虽然可以避免死锁,但如果是业务执行耗时较长,也会导致锁释放存在安全隐患。
4.主从一致性:如果Redis提供了主从集群,主从同步存在延迟,当主节点宕机时,如果从节点还未同步主节点中的锁数据,则会出现锁信息的不一致。
Redisson是一个在Redis的基础上实现的Java驻内存数据网格。它不仅提供了一系列的分布式的Java常用对象,还提供了许多分布式服务,其中包含了各种分布式锁的实现。
P42 分布式锁 Redisson快速入门
第1步,先引入依赖:
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
第2步,在config包下创建RedissonConfig类,写入如下代码:
@Configuration
public class RedissonConfig{
@Bean
public RedissonClient redissonClient(){
//配置
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("");
//创建RedissonClient对象
return Redisson.create(config);
}
}
第3步,引入RedissonClient,调用getLock获取锁对象,然后用tryLock获取锁。
第4步,启动服务
发送下面的请求:
在执行释放锁的语句前,可以看到Redis中有锁的记录:
用jmeter来测试,可以发现没有出现并发安全问题:
P43 分布式锁 Redisson的可重入锁原理
ReentrantLock可重入锁的原理:获取锁的时候在判断这个锁已经被占有的情况下,会检查占有锁的是否是当前线程,如果是当前线程,也会获取锁成功。会有一个计数器记录重入的次数。
会通过下面的结构来记录某个线程重入了几次锁。
每释放一次锁采用的策略是把重入次数减1。
加锁和释放锁是成对出现的,因此当方法执行到最外层结束时,重入的次数一定会减为0。
1.是否存在锁
2.存在锁,判断是否是自己的。
是,锁计数+1。
不是,获取锁失败。
3.不存在锁
获取锁,添加线程标示。
Redisson底层可重入锁加锁的逻辑:
Redisson底层可重入锁解锁的逻辑: