《黑马点评》Redis高并发项目实战笔记(上)P1~P43

 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底层可重入锁解锁的逻辑:

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/540242.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

冲刺2024年思维100春季线上比赛:做做历年思维100真题(附答案)

今天是2024年4月13日&#xff0c;距离2024年春季思维100活动第一阶段的线上比赛4月20日还有7天。今年思维100活动的考试重点是什么呢&#xff1f;虽然主办方未公布&#xff0c;我们可以从历年的思维100真题中来分析和推测&#xff0c;把历年真题和背后的知识点吃透了&#xff0…

宠物领养|基于springboot的宠物领养系统设计与实现(附项目源码+论文)

基于springboot的宠物领养系统设计与实现 目录 一、摘要 二、相关技术 三、系统设计 3.1 整体功能设计图 3.2 功能具体细节设计 四、数据库设计 五、核心代码 六、论文参考 七、源码获取 一、摘要 如今社会上各行各业&#xff0c;都在用属于自己专用的软件来进行…

记录一下如何腾讯云服务器用客户端连MySQL

我一般喜欢用IDEA连数据库&#xff0c;别问我为啥&#xff08;就喜欢用一个软件解决所有问题&#xff09; 当然写SQL语句个人还是觉得sqlyog体验最佳&#xff01;

llama-factory SFT系列教程 (二),大模型在自定义数据集 lora 训练与部署

文章目录 简介支持的模型列表2. 添加自定义数据集3. lora 微调4. 大模型 lora 权重&#xff0c;部署问题 参考资料 简介 llama-factory SFT系列教程 (一)&#xff0c;大模型 API 部署与使用本文为 llama-factory SFT系列教程的第二篇&#xff1b; 支持的模型列表 模型名模型…

服务器配置环境步骤

1、创建虚拟环境 conda create --name 名字 pythonpython版本号2、进入虚拟环境 conda activate 名字3、确认自己要安装的torch版本和torchvision版本&#xff0c;进入https://pytorch.org/get-started/previous-versions/ 复制相应的命令&#xff0c;运行即可 注&#xff1a…

Visual Studio Code SSH 连接远程服务器

Visual Studio Code通过 SSH 连接远程服务器并实现免密登录&#xff0c;你可以按照以下步骤进行操作&#xff1a; 1. **安装插件**&#xff1a;首先&#xff0c;在 VS Code 中安装 "Remote - SSH" 插件。打开 VS Code&#xff0c;点击左侧的扩展图标&#xff0c;搜索…

C++如何排查并发编程死锁问题?

C如何排查并发编程死锁问题&#xff1f; 最近在Apache arrow里面写一个支持并行的算子&#xff1a;nested loop join&#xff0c;然后既然涉及到并行&#xff0c;这里就会遇到大家常说的死锁问题&#xff0c;假设你碰到了死锁问题&#xff0c;如何调试与定位呢&#xff1f; 那这…

最小均方(LMS)自适应滤波算法

1 LMS函数实现 % ----------------------------LMS(Least Mean Squre)算法------------------------------ % % parm: % xn 输入的信号序列 (列向量) % dn 所期望的响应序列 (列向量) % M 滤波器的阶数 (标量) % mu 收敛因子(步长) …

统信UOS(Linux)安装nvm node管理工具

整篇看完再操作&#xff0c;有坑&#xff01;&#xff01; 官网 nvm官网 按照官网方式安装&#xff0c;一直报 错 经过不断研究&#xff0c;正确步骤如下 1、下载安装包 可能因为网络安全不能访问github&#xff0c;我是链接热点下载的 wget https://github.com/nvm-sh/…

Java Maven项目推送到 Maven 中央仓库

准备阶段 namespace 域名认证 当需要在 sonatype 认证 com.xxx命名空间时&#xff0c;需要将 .xxx.com 配置域名解析。 记录类型&#xff1a;TXT 文本内容&#xff1a;验证的 key。 GPG 公私钥生成 GPG 下载地址&#xff1a;https://www.gnupg.org/download/index.html M…

RestTemplate—微服务远程调用—案例解析

简介&#xff1a;总结来说&#xff0c;微服务之间的调用方式有多种&#xff0c;选择哪种方式取决于具体的业务需求、技术栈和架构设计。RESTful API和HTTP客户端是常见的选择&#xff0c;而Feign和Ribbon等辅助库可以简化调用过程。RPC和消息队列适用于特定的场景&#xff0c;如…

单片机方案 发声毛绒小黄鸭

随着科技的不断进步&#xff0c;智能早教已经成为了新时代儿童教育的趋势。智能早教玩具&#xff0c;一款集互动陪伴、启蒙教育、情感培养于一身的高科技产品。它不仅能陪伴孩子成长&#xff0c;还能在游戏中启迪智慧&#xff0c;是家长和孩子的理想选择。 酷得电子方案开发特…

OV通配符证书:安全、便捷的网络认证新选择

OV通配符证书&#xff0c;即组织验证型通配符证书&#xff0c;其最大特点在于其通配符功能。这意味着&#xff0c;一个OV通配符证书可以覆盖同一主域名下的多个子域名&#xff0c;大大简化了证书管理和维护的复杂性。无论是大型企业还是个人网站&#xff0c;都可以通过OV通配符…

[面向对象] 单例模式与工厂模式

单例模式 是一种创建模式&#xff0c;保证一个类只有一个实例&#xff0c;且提供访问实例的全局节点。 工厂模式 面向对象其中的三大原则&#xff1a; 单一职责&#xff1a;一个类只有一个职责&#xff08;Game类负责什么时候创建英雄机&#xff0c;而不需要知道创建英雄机要…

多因子模型的数据处理

优质博文&#xff1a;IT-BLOG-CN 数据处理的基本目的是从多量的、可能是杂乱无章的、难以理解的数据中抽取并推导出有价值、有意义的数据。特别是金融数据&#xff0c;存在数据缺失&#xff0c;不完整以及极端异常值等问题&#xff0c;对于我们的分析和建模影响很多。 对于我…

Git分布式版本控制系统——Git常用命令(二)

五、Git常用命令————分支操作 同一个仓库可以有多个分支&#xff0c;各个分支相互独立&#xff0c;互不干扰 分支的相关命令&#xff0c;具体如下&#xff1a; git branch 查看分支 git branch [name] 创建分支&#x…

20240409在全志H3平台的Nano Pi NEO CORE开发板上运行Ubuntu Core16.04时跑通4G模块EC200A-CN【PPP模式】

20240409在全志H3平台的Nano Pi NEO CORE开发板上运行Ubuntu Core16.04时跑通4G模块EC200A-CN【PPP模式】 2024/4/9 14:25 【不建议使用ppp模式&#xff0c;功耗大&#xff0c;貌似更过分的&#xff01;网速还低&#xff01;】 【唯一的优点&#xff1a;ppp模式下是通过脚本配置…

降额的秘密——不要挑战datasheet!

原文来自微信公众号&#xff1a;工程师看海&#xff0c;与我联系&#xff1a;chunhou0820 看海原创视频教程&#xff1a;《运放秘籍》 大家好&#xff0c;我是工程师看海。 什么是降额设计&#xff1f;我们为什么要降额&#xff1f; 额指的是额定工作状态&#xff0c;降额就是…

CSS特效---HTML+CSS实现3D旋转卡片

1、演示 2、一切尽在代码中 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8" /><meta name"viewport" content"widthdevice-width, initial-scale1.0" /><title>Document</title&…

Ubuntu无网络标识的解决方法

1.出现的情况的特点 2.解决办法 2.1 进入root并输入密码 sudo su 2.2 更新NetworkManager的配置 得先有gedit或者vim&#xff0c;两个随意一个&#xff0c;这里用的gedit&#xff0c;没有就先弄gedit&#xff0c;有的话直接下一步 apt-get install gedit 或者vim apt-get ins…