js做网站跳转,做赚钱问卷调查的网站,阿里云wordpress主机,影视网站怎么做原创实战篇Redis
开篇导读
亲爱的小伙伴们大家好#xff0c;马上咱们就开始实战篇的内容了#xff0c;相信通过本章的学习#xff0c;小伙伴们就能理解各种redis的使用啦#xff0c;接下来咱们来一起看看实战篇我们要学习一些什么样的内容
短信登录
这一块我们会使用redis共…实战篇Redis
开篇导读
亲爱的小伙伴们大家好马上咱们就开始实战篇的内容了相信通过本章的学习小伙伴们就能理解各种redis的使用啦接下来咱们来一起看看实战篇我们要学习一些什么样的内容
短信登录
这一块我们会使用redis共享session来实现
商户查询缓存
通过本章节我们会理解缓存击穿缓存穿透缓存雪崩等问题让小伙伴的对于这些概念的理解不仅仅是停留在概念上更是能在代码中看到对应的内容
优惠卷秒杀
通过本章节我们可以学会Redis的计数器功能 结合Lua完成高性能的redis操作同时学会Redis分布式锁的原理包括Redis的三种消息队列
附近的商户
我们利用Redis的GEOHash来完成对于地理坐标的操作
UV统计
主要是使用Redis来完成统计功能
用户签到
使用Redis的BitMap数据统计功能
好友关注
基于Set集合的关注、取消关注共同关注等等功能这一块知识咱们之前就讲过这次我们在项目中来使用一下
打人探店
基于List来完成点赞列表的操作同时基于SortedSet来完成点赞的排行榜功能
以上这些内容咱们统统都会给小伙伴们讲解清楚让大家充分理解如何使用Redis 1、短信登录
1.1、导入讯飞点评项目
1.1.1 、导入SQL 1.1.2、有关当前模型
手机或者app端发起请求请求我们的nginx服务器nginx基于七层模型走的事HTTP协议可以实现基于Lua直接绕开tomcat访问redis也可以作为静态资源服务器轻松扛下上万并发 负载均衡到下游tomcat服务器打散流量我们都知道一台4核8G的tomcat在优化和处理简单业务的加持下大不了就处理1000左右的并发 经过nginx的负载均衡分流后利用集群支撑起整个项目同时nginx在部署了前端项目后更是可以做到动静分离进一步降低tomcat服务的压力这些功能都得靠nginx起作用所以nginx是整个项目中重要的一环。
在tomcat支撑起并发流量后我们如果让tomcat直接去访问Mysql根据经验Mysql企业级服务器只要上点并发一般是16或32 核心cpu32 或64G内存像企业级mysql加上固态硬盘能够支撑的并发大概就是4000起~7000左右上万并发 瞬间就会让Mysql服务器的cpu硬盘全部打满容易崩溃所以我们在高并发场景下会选择使用mysql集群同时为了进一步降低Mysql的压力同时增加访问的性能我们也会加入Redis同时使用Redis集群使得Redis对外提供更好的服务。 1.1.3、导入后端项目
在资料中提供了一个项目源码 1.1.4、导入前端工程 1.1.5 运行前端项目 1.2 、基于Session实现登录流程
发送验证码
用户在提交手机号后会校验手机号是否合法如果不合法则要求用户重新输入手机号
如果手机号合法后台此时生成对应的验证码同时将验证码进行保存然后再通过短信的方式将验证码发送给用户
短信验证码登录、注册
用户将验证码和手机号进行输入后台从session中拿到当前验证码然后和用户输入的验证码进行校验如果不一致则无法通过校验如果一致则后台根据手机号查询用户如果用户不存在则为用户创建账号信息保存到数据库无论是否存在都会将用户信息保存到session中方便后续获得当前登录信息
校验登录状态:
用户在请求时候会从cookie中携带者JsessionId到后台后台通过JsessionId从session中拿到用户信息如果没有session信息则进行拦截如果有session信息则将用户信息保存到threadLocal中并且放行 1.3 、实现发送短信验证码功能
页面流程 具体代码如下
贴心小提示
具体逻辑上文已经分析我们仅仅只需要按照提示的逻辑写出代码即可。
发送验证码 Overridepublic Result sendCode(String phone, HttpSession session) {// 1.校验手机号if (RegexUtils.isPhoneInvalid(phone)) {// 2.如果不符合返回错误信息return Result.fail(手机号格式错误);}// 3.符合生成验证码String code RandomUtil.randomNumbers(6);// 4.保存验证码到 sessionsession.setAttribute(code,code);// 5.发送验证码log.debug(发送短信验证码成功验证码{}, code);// 返回okreturn Result.ok();}登录 Overridepublic Result login(LoginFormDTO loginForm, HttpSession session) {// 1.校验手机号String phone loginForm.getPhone();if (RegexUtils.isPhoneInvalid(phone)) {// 2.如果不符合返回错误信息return Result.fail(手机号格式错误);}// 3.校验验证码Object cacheCode session.getAttribute(code);String code loginForm.getCode();if(cacheCode null || !cacheCode.toString().equals(code)){//3.不一致报错return Result.fail(验证码错误);}//一致根据手机号查询用户User user query().eq(phone, phone).one();//5.判断用户是否存在if(user null){//不存在则创建user createUserWithPhone(phone);}//7.保存用户信息到session中session.setAttribute(user,user);return Result.ok();}1.4、实现登录拦截功能
温馨小贴士tomcat的运行原理 当用户发起请求时会访问我们像tomcat注册的端口任何程序想要运行都需要有一个线程对当前端口号进行监听tomcat也不例外当监听线程知道用户想要和tomcat连接连接时那会由监听线程创建socket连接socket都是成对出现的用户通过socket像互相传递数据当tomcat端的socket接受到数据后此时监听线程会从tomcat的线程池中取出一个线程执行用户请求在我们的服务部署到tomcat后线程会找到用户想要访问的工程然后用这个线程转发到工程中的controllerservicedao中并且访问对应的DB在用户执行完请求后再统一返回再找到tomcat端的socket再将数据写回到用户端的socket完成请求和响应
通过以上讲解我们可以得知 每个用户其实对应都是去找tomcat线程池中的一个线程来完成工作的 使用完成后再进行回收既然每个请求都是独立的所以在每个用户去访问我们的工程时我们可以使用threadlocal来做到线程隔离每个线程操作自己的一份数据
温馨小贴士关于threadlocal
如果小伙伴们看过threadLocal的源码你会发现在threadLocal中无论是他的put方法和他的get方法 都是先从获得当前用户的线程然后从线程中取出线程的成员变量map只要线程不一样map就不一样所以可以通过这种方式来做到线程隔离 拦截器代码
public class LoginInterceptor implements HandlerInterceptor {Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {//1.获取sessionHttpSession session request.getSession();//2.获取session中的用户Object user session.getAttribute(user);//3.判断用户是否存在if(user null){//4.不存在拦截返回401状态码response.setStatus(401);return false;}//5.存在保存用户信息到ThreadlocalUserHolder.saveUser((User)user);//6.放行return true;}
}让拦截器生效
Configuration
public class MvcConfig implements WebMvcConfigurer {Resourceprivate StringRedisTemplate stringRedisTemplate;Overridepublic void addInterceptors(InterceptorRegistry registry) {// 登录拦截器registry.addInterceptor(new LoginInterceptor()).excludePathPatterns(/shop/**,/voucher/**,/shop-type/**,/upload/**,/blog/hot,/user/code,/user/login).order(1);// token刷新的拦截器registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns(/**).order(0);}
}1.5、隐藏用户敏感信息
我们通过浏览器观察到此时用户的全部信息都在这样极为不靠谱所以我们应当在返回用户信息之前将用户的敏感信息进行隐藏采用的核心思路就是书写一个UserDto对象这个UserDto对象就没有敏感信息了我们在返回前将有用户敏感信息的User对象转化成没有敏感信息的UserDto对象那么就能够避免这个尴尬的问题了
在登录方法处修改
//7.保存用户信息到session中
session.setAttribute(user, BeanUtils.copyProperties(user,UserDTO.class));在拦截器处
//5.存在保存用户信息到Threadlocal
UserHolder.saveUser((UserDTO) user);在UserHolder处将user对象换成UserDTO
public class UserHolder {private static final ThreadLocalUserDTO tl new ThreadLocal();public static void saveUser(UserDTO user){tl.set(user);}public static UserDTO getUser(){return tl.get();}public static void removeUser(){tl.remove();}
}1.6、session共享问题
核心思路分析
每个tomcat中都有一份属于自己的session,假设用户第一次访问第一台tomcat并且把自己的信息存放到第一台服务器的session中但是第二次这个用户访问到了第二台tomcat那么在第二台服务器上肯定没有第一台服务器存放的session所以此时 整个登录拦截功能就会出现问题我们能如何解决这个问题呢早期的方案是session拷贝就是说虽然每个tomcat上都有不同的session但是每当任意一台服务器的session修改时都会同步给其他的Tomcat服务器的session这样的话就可以实现session的共享了
但是这种方案具有两个大问题
1、每台服务器中都有完整的一份session数据服务器压力过大。
2、session拷贝数据时可能会出现延迟
所以咱们后来采用的方案都是基于redis来完成我们把session换成redisredis数据本身就是共享的就可以避免session共享的问题了 1.7 Redis代替session的业务流程
1.7.1、设计key的结构
首先我们要思考一下利用redis来存储数据那么到底使用哪种结构呢由于存入的数据比较简单我们可以考虑使用String或者是使用哈希如下图如果使用String同学们注意他的value用多占用一点空间如果使用哈希则他的value中只会存储他数据本身如果不是特别在意内存其实使用String就可以啦。 1.7.2、设计key的具体细节
所以我们可以使用String结构就是一个简单的keyvalue键值对的方式但是关于key的处理session他是每个用户都有自己的session但是redis的key是共享的咱们就不能使用code了
在设计这个key的时候我们之前讲过需要满足两点
1、key要具有唯一性
2、key要方便携带
如果我们采用phone手机号这个的数据来存储当然是可以的但是如果把这样的敏感数据存储到redis中并且从页面中带过来毕竟不太合适所以我们在后台生成一个随机串token然后让前端带来这个token就能完成我们的整体逻辑了
1.7.3、整体访问流程
当注册完成后用户去登录会去校验用户提交的手机号和验证码是否一致如果一致则根据手机号查询用户信息不存在则新建最后将用户数据保存到redis并且生成token作为redis的key当我们校验用户是否登录时会去携带着token进行访问从redis中取出token对应的value判断是否存在这个数据如果没有则拦截如果存在则将其保存到threadLocal中并且放行。 1.8 基于Redis实现短信登录
这里具体逻辑就不分析了之前咱们已经重点分析过这个逻辑啦。
UserServiceImpl代码
Override
public Result login(LoginFormDTO loginForm, HttpSession session) {// 1.校验手机号String phone loginForm.getPhone();if (RegexUtils.isPhoneInvalid(phone)) {// 2.如果不符合返回错误信息return Result.fail(手机号格式错误);}// 3.从redis获取验证码并校验String cacheCode stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY phone);String code loginForm.getCode();if (cacheCode null || !cacheCode.equals(code)) {// 不一致报错return Result.fail(验证码错误);}// 4.一致根据手机号查询用户 select * from tb_user where phone ?User user query().eq(phone, phone).one();// 5.判断用户是否存在if (user null) {// 6.不存在创建新用户并保存user createUserWithPhone(phone);}// 7.保存用户信息到 redis中// 7.1.随机生成token作为登录令牌String token UUID.randomUUID().toString(true);// 7.2.将User对象转为HashMap存储UserDTO userDTO BeanUtil.copyProperties(user, UserDTO.class);MapString, Object userMap BeanUtil.beanToMap(userDTO, new HashMap(),CopyOptions.create().setIgnoreNullValue(true).setFieldValueEditor((fieldName, fieldValue) - fieldValue.toString()));// 7.3.存储String tokenKey LOGIN_USER_KEY token;stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);// 7.4.设置token有效期stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);// 8.返回tokenreturn Result.ok(token);
}1.9 解决状态登录刷新问题
1.9.1 初始方案思路总结
在这个方案中他确实可以使用对应路径的拦截同时刷新登录token令牌的存活时间但是现在这个拦截器他只是拦截需要被拦截的路径假设当前用户访问了一些不需要拦截的路径那么这个拦截器就不会生效所以此时令牌刷新的动作实际上就不会执行所以这个方案他是存在问题的 1.9.2 优化方案
既然之前的拦截器无法对不需要拦截的路径生效那么我们可以添加一个拦截器在第一个拦截器中拦截所有的路径把第二个拦截器做的事情放入到第一个拦截器中同时刷新令牌因为第一个拦截器有了threadLocal的数据所以此时第二个拦截器只需要判断拦截器中的user对象是否存在即可完成整体刷新功能。 1.9.3 代码
RefreshTokenInterceptor
public class RefreshTokenInterceptor implements HandlerInterceptor {private StringRedisTemplate stringRedisTemplate;public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate stringRedisTemplate;}Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {// 1.获取请求头中的tokenString token request.getHeader(authorization);if (StrUtil.isBlank(token)) {return true;}// 2.基于TOKEN获取redis中的用户String key LOGIN_USER_KEY token;MapObject, Object userMap stringRedisTemplate.opsForHash().entries(key);// 3.判断用户是否存在if (userMap.isEmpty()) {return true;}// 5.将查询到的hash数据转为UserDTOUserDTO userDTO BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);// 6.存在保存用户信息到 ThreadLocalUserHolder.saveUser(userDTO);// 7.刷新token有效期stringRedisTemplate.expire(key, LOGIN_USER_TTL, TimeUnit.MINUTES);// 8.放行return true;}Overridepublic void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {// 移除用户UserHolder.removeUser();}
}
LoginInterceptor
public class LoginInterceptor implements HandlerInterceptor {Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {// 1.判断是否需要拦截ThreadLocal中是否有用户if (UserHolder.getUser() null) {// 没有需要拦截设置状态码response.setStatus(401);// 拦截return false;}// 有用户则放行return true;}
}2、商户查询缓存
2.1 什么是缓存?
前言:什么是缓存?
就像自行车,越野车的避震器
举个例子:越野车,山地自行车,都拥有避震器,防止车体加速后因惯性,在酷似U字母的地形上飞跃,硬着陆导致的损害,像个弹簧一样;
同样,实际开发中,系统也需要避震器,防止过高的数据访问猛冲系统,导致其操作线程无法及时处理信息而瘫痪;
这在实际开发中对企业讲,对产品口碑,用户评价都是致命的;所以企业非常重视缓存技术;
缓存(Cache),就是数据交换的缓冲区,俗称的缓存就是缓冲区内的数据,一般从数据库中获取,存储于本地代码(例如:
例1:Static final ConcurrentHashMapK,V map new ConcurrentHashMap(); 本地用于高并发例2:static final CacheK,V USER_CACHE CacheBuilder.newBuilder().build(); 用于redis等缓存例3:Static final MapK,V map new HashMap(); 本地缓存由于其被Static修饰,所以随着类的加载而被加载到内存之中,作为本地缓存,由于其又被final修饰,所以其引用(例3:map)和对象(例3:new HashMap())之间的关系是固定的,不能改变,因此不用担心赋值()导致缓存失效;
2.1.1 为什么要使用缓存
一句话:因为速度快,好用
缓存数据存储于代码中,而代码运行在内存中,内存的读写性能远高于磁盘,缓存可以大大降低用户访问并发量带来的服务器读写压力
实际开发过程中,企业的数据量,少则几十万,多则几千万,这么大数据量,如果没有缓存来作为避震器,系统是几乎撑不住的,所以企业会大量运用到缓存技术;
但是缓存也会增加代码复杂度和运营的成本: 2.1.2 如何使用缓存
实际开发中,会构筑多级缓存来使系统运行速度进一步提升,例如:本地缓存与redis中的缓存并发使用
浏览器缓存主要是存在于浏览器端的缓存
**应用层缓存**可以分为tomcat本地缓存比如之前提到的map或者是使用redis作为缓存
**数据库缓存**在数据库中有一片空间是 buffer pool增改查数据都会先加载到mysql的缓存中
**CPU缓存**当代计算机最大的问题是 cpu性能提升了但内存读写速度没有跟上所以为了适应当下的情况增加了cpu的L1L2L3级的缓存 2.2 添加商户缓存
在我们查询商户信息时我们是直接操作从数据库中去进行查询的大致逻辑是这样直接查询数据库那肯定慢咯所以我们需要增加缓存
GetMapping(/{id})
public Result queryShopById(PathVariable(id) Long id) {//这里是直接查询数据库return shopService.queryById(id);
}2.2.1 、缓存模型和思路
标准的操作方式就是查询数据库之前先查询缓存如果缓存数据存在则直接从缓存中返回如果缓存数据不存在再查询数据库然后将数据存入redis。 2.1.2、代码如下
代码思路如果缓存有则直接返回如果缓存不存在则查询数据库然后存入redis。 2.3 缓存更新策略
缓存更新是redis为了节约内存而设计出来的一个东西主要是因为内存数据宝贵当我们向redis插入太多数据此时就可能会导致缓存中的数据过多所以redis会对部分数据进行更新或者把他叫为淘汰更合适。
**内存淘汰**redis自动进行当redis内存达到咱们设定的max-memery的时候会自动触发淘汰机制淘汰掉一些不重要的数据(可以自己设置策略方式)
**超时剔除**当我们给redis设置了过期时间ttl之后redis会将超时的数据进行删除方便咱们继续使用缓存
**主动更新**我们可以手动调用方法把缓存删掉通常用于解决缓存和数据库不一致问题 2.3.1 、数据库缓存不一致解决方案
由于我们的缓存的数据源来自于数据库,而数据库的数据是会发生变化的,因此,如果当数据库中数据发生变化,而缓存却没有同步,此时就会有一致性问题存在,其后果是:
用户使用缓存中的过时数据,就会产生类似多线程数据安全问题,从而影响业务,产品口碑等;怎么解决呢有如下几种方案
Cache Aside Pattern 人工编码方式缓存调用者在更新完数据库后再去更新缓存也称之为双写方案
Read/Write Through Pattern : 由系统本身完成数据库与缓存的问题交由系统本身去处理
Write Behind Caching Pattern 调用者只操作缓存其他线程去异步处理数据库实现最终一致 2.3.2 、数据库和缓存不一致采用什么方案
综合考虑使用方案一但是方案一调用者如何处理呢这里有几个问题
操作缓存和数据库时有三个问题需要考虑
如果采用第一个方案那么假设我们每次操作数据库后都操作缓存但是中间如果没有人查询那么这个更新动作实际上只有最后一次生效中间的更新动作意义并不大我们可以把缓存删除等待再次查询时将缓存中的数据加载出来 删除缓存还是更新缓存 更新缓存每次更新数据库都更新缓存无效写操作较多删除缓存更新数据库时让缓存失效查询时再更新缓存 如何保证缓存与数据库的操作的同时成功或失败 单体系统将缓存与数据库操作放在一个事务分布式系统利用TCC等分布式事务方案
应该具体操作缓存还是操作数据库我们应当是先操作数据库再删除缓存原因在于如果你选择第一种方案在两个线程并发来访问时假设线程1先来他先把缓存删了此时线程2过来他查询缓存数据并不存在此时他写入缓存当他写入缓存后线程1再执行更新动作时实际上写入的就是旧的数据新的数据被旧数据覆盖了。
先操作缓存还是先操作数据库 先删除缓存再操作数据库先操作数据库再删除缓存 2.4 实现商铺和缓存与数据库双写一致
核心思路如下
修改ShopController中的业务逻辑满足下面的需求
根据id查询店铺时如果缓存未命中则查询数据库将数据库结果写入缓存并设置超时时间
根据id修改店铺时先修改数据库再删除缓存
修改重点代码1修改ShopServiceImpl的queryById方法
设置redis缓存时添加过期时间 修改重点代码2
代码分析通过之前的淘汰我们确定了采用删除策略来解决双写问题当我们修改了数据之后然后把缓存中的数据进行删除查询时发现缓存中没有数据则会从mysql中加载最新的数据从而避免数据库和缓存不一致的问题 2.5 缓存穿透问题的解决思路
缓存穿透 缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在这样缓存永远不会生效这些请求都会打到数据库。
常见的解决方案有两种
缓存空对象 优点实现简单维护方便缺点 额外的内存消耗可能造成短期的不一致 布隆过滤 优点内存占用较少没有多余key缺点 实现复杂存在误判可能
**缓存空对象思路分析**当我们客户端访问不存在的数据时先请求redis但是此时redis中没有数据此时会访问到数据库但是数据库中也没有数据这个数据穿透了缓存直击数据库我们都知道数据库能够承载的并发不如redis这么高如果大量的请求同时过来访问这种不存在的数据这些请求就都会访问到数据库简单的解决方案就是哪怕这个数据在数据库中也不存在我们也把这个数据存入到redis中去这样下次用户过来访问这个不存在的数据那么在redis中也能找到这个数据就不会进入到缓存了
**布隆过滤**布隆过滤器其实采用的是哈希思想来解决这个问题通过一个庞大的二进制数组走哈希思想去判断当前这个要查询的这个数据是否存在如果布隆过滤器判断存在则放行这个请求会去访问redis哪怕此时redis中的数据过期了但是数据库中一定存在这个数据在数据库中查询出来这个数据后再将其放入到redis中
假设布隆过滤器判断这个数据不存在则直接返回
这种方式优点在于节约内存空间存在误判误判原因在于布隆过滤器走的是哈希思想只要哈希思想就可能存在哈希冲突 2.6 编码解决商品查询的缓存穿透问题
核心思路如下
在原来的逻辑中我们如果发现这个数据在mysql中不存在直接就返回404了这样是会存在缓存穿透问题的
现在的逻辑中如果这个数据不存在我们不会返回404 还是会把这个数据写入到Redis中并且将value设置为空欧当再次发起查询时我们如果发现命中之后判断这个value是否是null如果是null则是之前写入的数据证明是缓存穿透数据如果不是则直接返回数据。 小总结
缓存穿透产生的原因是什么
用户请求的数据在缓存中和数据库中都不存在不断发起这样的请求给数据库带来巨大压力
缓存穿透的解决方案有哪些
缓存null值布隆过滤增强id的复杂度避免被猜测id规律做好数据的基础格式校验加强用户权限校验做好热点参数的限流
2.7 缓存雪崩问题及解决思路
缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机导致大量请求到达数据库带来巨大压力。
解决方案
给不同的Key的TTL添加随机值利用Redis集群提高服务的可用性给缓存业务添加降级限流策略给业务添加多级缓存 2.8 缓存击穿问题及解决思路
缓存击穿问题也叫热点Key问题就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了无数的请求访问会在瞬间给数据库带来巨大的冲击。
常见的解决方案有两种
互斥锁逻辑过期
逻辑分析假设线程1在查询缓存之后本来应该去查询数据库然后把这个数据重新加载到缓存的此时只要线程1走完这个逻辑其他线程就都能从缓存中加载这些数据了但是假设在线程1没有走完的时候后续的线程2线程3线程4同时过来访问当前这个方法 那么这些线程都不能从缓存中查询到数据那么他们就会同一时刻来访问查询缓存都没查到接着同一时间去访问数据库同时的去执行数据库代码对数据库访问压力过大 解决方案一、使用锁来解决
因为锁能实现互斥性。假设线程过来只能一个人一个人的来访问数据库从而避免对于数据库访问压力过大但这也会影响查询的性能因为此时会让查询的性能从并行变成了串行我们可以采用tryLock方法 double check来解决这样的问题。
假设现在线程1过来访问他查询缓存没有命中但是此时他获得到了锁的资源那么线程1就会一个人去执行逻辑假设现在线程2过来线程2在执行过程中并没有获得到锁那么线程2就可以进行到休眠直到线程1把锁释放后线程2获得到锁然后再来执行逻辑此时就能够从缓存中拿到数据了。 解决方案二、逻辑过期方案
方案分析我们之所以会出现这个缓存击穿问题主要原因是在于我们对key设置了过期时间假设我们不设置过期时间其实就不会有缓存击穿的问题但是不设置过期时间这样数据不就一直占用我们内存了吗我们可以采用逻辑过期方案。
我们把过期时间设置在 redis的value中注意这个过期时间并不会直接作用于redis而是我们后续通过逻辑去处理。假设线程1去查询缓存然后从value中判断出来当前的数据已经过期了此时线程1去获得互斥锁那么其他线程会进行阻塞获得了锁的线程他会开启一个 线程去进行 以前的重构数据的逻辑直到新开的线程完成这个逻辑后才释放锁 而线程1直接进行返回假设现在线程3过来访问由于线程线程2持有着锁所以线程3无法获得锁线程3也直接返回数据只有等到新开的线程2把重建数据构建完后其他线程才能走返回正确的数据。
这种方案巧妙在于异步的构建缓存缺点在于在构建完缓存之前返回的都是脏数据。 进行对比
**互斥锁方案**由于保证了互斥性所以数据一致且实现简单因为仅仅只需要加一把锁而已也没其他的事情需要操心所以没有额外的内存消耗缺点在于有锁就有死锁问题的发生且只能串行执行性能肯定受到影响
逻辑过期方案 线程读取过程中不需要等待性能好有一个额外的线程持有锁去进行重构数据但是在重构数据完成前其他的线程只能返回之前的数据且实现起来麻烦 2.9 利用互斥锁解决缓存击穿问题
核心思路相较于原来从缓存中查询不到数据后直接查询数据库而言现在的方案是 进行查询之后如果从缓存没有查询到数据则进行互斥锁的获取获取互斥锁后判断是否获得到了锁如果没有获得到则休眠过一会再进行尝试直到获取到锁为止才能进行查询
如果获取到了锁的线程再去进行查询查询后将数据写入redis再释放锁返回数据利用互斥锁就能保证只有一个线程去执行操作数据库的逻辑防止缓存击穿 操作锁的代码
核心思路就是利用redis的setnx方法来表示获取锁该方法含义是redis中如果没有这个key则插入成功返回1在stringRedisTemplate中返回true 如果有这个key则插入失败则返回0在stringRedisTemplate返回false我们可以通过true或者是false来表示是否有线程成功插入key成功插入的key的线程我们认为他就是获得到锁的线程。 private boolean tryLock(String key) {Boolean flag stringRedisTemplate.opsForValue().setIfAbsent(key, 1, 10, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);
}private void unlock(String key) {stringRedisTemplate.delete(key);
}操作代码 public Shop queryWithMutex(Long id) {String key CACHE_SHOP_KEY id;// 1、从redis中查询商铺缓存String shopJson stringRedisTemplate.opsForValue().get(key);// 2、判断是否存在if (StrUtil.isNotBlank(shopJson)) {// 存在,直接返回return JSONUtil.toBean(shopJson, Shop.class);}//判断命中的值是否是空值if (shopJson ! null) {//返回一个错误信息return null;}// 4.实现缓存重构//4.1 获取互斥锁String lockKey lock:shop: 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);// 5.不存在返回错误if(shop null){//将空值写入redisstringRedisTemplate.opsForValue().set(key,,CACHE_NULL_TTL,TimeUnit.MINUTES);//返回错误信息return null;}//6.写入redisstringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),CACHE_NULL_TTL,TimeUnit.MINUTES);}catch (Exception e){throw new RuntimeException(e);}finally {//7.释放互斥锁unlock(lockKey);}return shop;}3.0 、利用逻辑过期解决缓存击穿问题
需求修改根据id查询商铺的业务基于逻辑过期方式来解决缓存击穿问题
思路分析当用户开始查询redis时判断是否命中如果没有命中则直接返回空数据不查询数据库而一旦命中后将value取出判断value中的过期时间是否满足如果没有过期则直接返回redis中的数据如果过期则在开启独立线程后直接返回之前的数据独立线程去重构数据重构完成后释放互斥锁。 如果封装数据因为现在redis中存储的数据的value需要带上过期时间此时要么你去修改原来的实体类要么你
步骤一、
新建一个实体类我们采用第二个方案这个方案对原来代码没有侵入性。
Data
public class RedisData {private LocalDateTime expireTime;private Object data;
}步骤二、
在ShopServiceImpl 新增此方法利用单元测试进行缓存预热 在测试类中 步骤三正式代码
ShopServiceImpl
private static final ExecutorService CACHE_REBUILD_EXECUTOR Executors.newFixedThreadPool(10);
public Shop queryWithLogicalExpire( Long id ) {String key CACHE_SHOP_KEY id;// 1.从redis查询商铺缓存String json stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isBlank(json)) {// 3.存在直接返回return null;}// 4.命中需要先把json反序列化为对象RedisData redisData JSONUtil.toBean(json, RedisData.class);Shop shop JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);LocalDateTime expireTime redisData.getExpireTime();// 5.判断是否过期if(expireTime.isAfter(LocalDateTime.now())) {// 5.1.未过期直接返回店铺信息return shop;}// 5.2.已过期需要缓存重建// 6.缓存重建// 6.1.获取互斥锁String lockKey LOCK_SHOP_KEY id;boolean isLock tryLock(lockKey);// 6.2.判断是否获取锁成功if (isLock){CACHE_REBUILD_EXECUTOR.submit( ()-{try{//重建缓存this.saveShop2Redis(id,20L);}catch (Exception e){throw new RuntimeException(e);}finally {unlock(lockKey);}});}// 6.4.返回过期的商铺信息return shop;
}3.1、封装Redis工具类
基于StringRedisTemplate封装一个缓存工具类满足下列需求
方法1将任意Java对象序列化为json并存储在string类型的key中并且可以设置TTL过期时间方法2将任意Java对象序列化为json并存储在string类型的key中并且可以设置逻辑过期时间用于处理缓
存击穿问题
方法3根据指定的key查询缓存并反序列化为指定类型利用缓存空值的方式解决缓存穿透问题方法4根据指定的key查询缓存并反序列化为指定类型需要利用逻辑过期解决缓存击穿问题
将逻辑进行封装
Slf4j
Component
public class CacheClient {private final StringRedisTemplate stringRedisTemplate;private static final ExecutorService CACHE_REBUILD_EXECUTOR Executors.newFixedThreadPool(10);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 time, TimeUnit unit) {// 设置逻辑过期RedisData redisData new RedisData();redisData.setData(value);redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));// 写入RedisstringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));}public R,ID R queryWithPassThrough(String keyPrefix, ID id, ClassR type, FunctionID, R dbFallback, Long time, TimeUnit unit){String key keyPrefix id;// 1.从redis查询商铺缓存String json stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isNotBlank(json)) {// 3.存在直接返回return JSONUtil.toBean(json, type);}// 判断命中的是否是空值if (json ! null) {// 返回一个错误信息return null;}// 4.不存在根据id查询数据库R r dbFallback.apply(id);// 5.不存在返回错误if (r null) {// 将空值写入redisstringRedisTemplate.opsForValue().set(key, , CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回错误信息return null;}// 6.存在写入redisthis.set(key, r, time, unit);return r;}public R, ID R queryWithLogicalExpire(String keyPrefix, ID id, ClassR type, FunctionID, R dbFallback, Long time, TimeUnit unit) {String key keyPrefix id;// 1.从redis查询商铺缓存String json stringRedisTemplate.opsForValue().get(key);// 2.判断是否存在if (StrUtil.isBlank(json)) {// 3.存在直接返回return null;}// 4.命中需要先把json反序列化为对象RedisData redisData JSONUtil.toBean(json, RedisData.class);R r JSONUtil.toBean((JSONObject) redisData.getData(), type);LocalDateTime expireTime redisData.getExpireTime();// 5.判断是否过期if(expireTime.isAfter(LocalDateTime.now())) {// 5.1.未过期直接返回店铺信息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 newR dbFallback.apply(id);// 重建缓存this.setWithLogicalExpire(key, newR, time, unit);} catch (Exception e) {throw new RuntimeException(e);}finally {// 释放锁unlock(lockKey);}});}// 6.4.返回过期的商铺信息return r;}public R, ID R queryWithMutex(String keyPrefix, ID id, ClassR type, FunctionID, 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);}// 判断命中的是否是空值if (shopJson ! null) {// 返回一个错误信息return null;}// 4.实现缓存重建// 4.1.获取互斥锁String lockKey LOCK_SHOP_KEY id;R r null;try {boolean isLock tryLock(lockKey);// 4.2.判断是否获取成功if (!isLock) {// 4.3.获取锁失败休眠并重试Thread.sleep(50);return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);}// 4.4.获取锁成功根据id查询数据库r dbFallback.apply(id);// 5.不存在返回错误if (r null) {// 将空值写入redisstringRedisTemplate.opsForValue().set(key, , CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回错误信息return null;}// 6.存在写入redisthis.set(key, r, time, unit);} catch (InterruptedException e) {throw new RuntimeException(e);}finally {// 7.释放锁unlock(lockKey);}// 8.返回return r;}private boolean tryLock(String key) {Boolean flag stringRedisTemplate.opsForValue().setIfAbsent(key, 1, 10, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);}private void unlock(String key) {stringRedisTemplate.delete(key);}
}在ShopServiceImpl 中
Resource
private CacheClient cacheClient;Overridepublic Result queryById(Long id) {// 解决缓存穿透Shop shop cacheClient.queryWithPassThrough(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);// 互斥锁解决缓存击穿// Shop shop cacheClient// .queryWithMutex(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);// 逻辑过期解决缓存击穿// Shop shop cacheClient// .queryWithLogicalExpire(CACHE_SHOP_KEY, id, Shop.class, this::getById, 20L, TimeUnit.SECONDS);if (shop null) {return Result.fail(店铺不存在);}// 7.返回return Result.ok(shop);}3、优惠卷秒杀
3.1 -全局唯一ID
每个店铺都可以发布优惠券 当用户抢购时就会生成订单并保存到tb_voucher_order这张表中而订单表如果使用数据库自增ID就存在一些问题
id的规律性太明显受单表数据量的限制
场景分析如果我们的id具有太明显的规则用户或者说商业对手很容易猜测出来我们的一些敏感信息比如商城在一天时间内卖出了多少单这明显不合适。
场景分析二随着我们商城规模越来越大mysql的单表的容量不宜超过500W数据量过大之后我们要进行拆库拆表但拆分表了之后他们从逻辑上讲他们是同一张表所以他们的id是不能一样的 于是乎我们需要保证id的唯一性。
全局ID生成器是一种在分布式系统下用来生成全局唯一ID的工具一般要满足下列特性 为了增加ID的安全性我们可以不直接使用Redis自增的数值而是拼接一些其它信息
ID的组成部分符号位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
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));
}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 在扣减库存时改为
boolean success seckillVoucherService.update().setSql(stock stock -1) //set stock stock -1.eq(voucher_id, voucherId).eq(stock,voucher.getStock()).update(); //where id and stock ?以上逻辑的核心含义是只要我扣减库存时的库存和之前我查询到的库存是一样的就意味着没有人在中间修改过库存那么此时就是安全的但是以上这种方式通过测试发现会有很多失败的情况失败的原因在于在使用乐观锁过程中假设100个线程同时都拿到了100的库存然后大家一起去进行扣减但是100个人中只有1个人能扣减成功其他的人在处理时他们在扣减时库存已经被修改过了所以此时其他线程都会失败
修改代码方案二、
之前的方式要修改前后都保持一致但是这样我们分析过成功的概率太低所以我们的乐观锁需要变一下改成stock大于0 即可
boolean success seckillVoucherService.update().setSql(stock stock -1).eq(voucher_id, voucherId).update().gt(stock,0); //where id ? and stock 0知识小扩展
针对cas中的自旋压力过大我们可以使用Longaddr这个类去解决
Java8 提供的一个对AtomicLong改进后的一个类LongAdder
大量线程并发更新一个原子性的时候天然的问题就是自旋会导致并发性问题当然这也比我们直接使用syn来的好
所以利用这么一个类LongAdder来进行优化
如果获取某个值则会对cell和base的值进行递增最后返回一个完整的值 3.6 优惠券秒杀-一人一单
需求修改秒杀业务要求同一个优惠券一个用户只能下一单
现在的问题在于
优惠卷是为了引流但是目前的情况是一个人可以无限制的抢这个优惠卷所以我们应当增加一层逻辑让一个用户只能下一个单而不是让一个用户下多个单
具体操作逻辑如下比如时间是否充足如果时间充足则进一步判断库存是否足够然后再根据优惠卷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.的方式调用的事务想要生效还得利用代理来生效所以这个地方我们需要获得原始的事务对象 来操作事务 3.7 集群环境下的并发问题
通过加锁可以解决在单机情况下的一人一单安全问题但是在集群模式下就不行了。
1、我们将服务启动两份端口分别为8081和8082 2、然后修改nginx的conf目录下的nginx.conf文件配置反向代理和负载均衡 具体操作(略)
有关锁失效原因分析
由于现在我们部署了多个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走到删除锁逻辑时如果没有卡过自动释放锁的时间点则判断当前这把锁是属于自己的于是删除这把锁。 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的拿锁比锁删锁实际上并不是原子性的我们要防止刚才的情况发生 4.7 Lua脚本解决多条命令原子性问题
Redis提供了Lua脚本功能在一个脚本中编写多条Redis命令确保多条命令执行时的原子性。Lua是一种编程语言它的基本语法大家可以参考网站https://www.runoob.com/lua/lua-tutorial.html这里重点介绍Redis提供的调用函数我们可以使用lua去操作redis又能保证他的原子性这样就可以实现拿锁比锁删锁是一个原子性动作了作为Java程序员这一块并不作一个简单要求并不需要大家过于精通只需要知道他有什么作用即可。
这里重点介绍Redis提供的调用函数语法如下
redis.call(命令名称, key, 其它参数, ...)例如我们要执行set name jack则脚本是这样
# 执行 set name jack
redis.call(set, name, jack)例如我们要先执行set name Rose再执行get name则脚本如下
# 先执行 set name jack
redis.call(set, name, Rose)
# 再执行 get name
local name redis.call(get, name)
# 返回
return name写好脚本以后需要用Redis命令来调用脚本调用脚本的常见命令如下 例如我们要执行 redis.call(‘set’, ‘name’, ‘jack’) 这个脚本语法如下 如果脚本中的key、value不想写死可以作为参数传递。key类型参数会放入KEYS数组其它参数会放入ARGV数组在脚本中可以从KEYS和ARGV数组获取这些参数 接下来我们来回一下我们释放锁的逻辑
释放锁的业务流程是这样的
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 04.8 利用Java代码调用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常用对象还提供了许多分布式服务其中就包含了各种分布式锁的实现。
Redission提供了分布式锁的多种多样的功能 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();}}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有线程加锁失败则会再次去进行重试. 6、秒杀优化
6.1 秒杀优化-异步秒杀思路
我们来回顾一下下单流程
当用户发起请求此时会请求nginxnginx会访问到tomcat而tomcat中的程序会进行串行操作分成如下几个步骤
1、查询优惠卷
2、判断秒杀库存是否足够
3、查询订单
4、校验是否是一人一单
5、扣减库存
6、创建订单
在这六步操作中又有很多操作是要去操作数据库的而且还是一个线程串行执行 这样就会导致我们的程序执行的很慢所以我们需要异步程序执行那么如何加速呢
在这里笔者想给大家分享一下课程内没有的思路看看有没有小伙伴这么想比如我们可以不可以使用异步编排来做或者说我开启N多线程N多个线程一个线程执行查询优惠卷一个执行判断扣减库存一个去创建订单等等然后再统一做返回这种做法和课程中有哪种好呢答案是课程中的好因为如果你采用我刚说的方式如果访问的人很多那么线程池中的线程可能一下子就被消耗完了而且你使用上述方案最大的特点在于你觉得时效性会非常重要但是你想想是吗并不是比如我只要确定他能做这件事然后我后边慢慢做就可以了我并不需要他一口气做完这件事所以我们应当采用的是课程中类似消息队列的方式来完成我们的需求而不是使用线程池或者是异步编排的方式来完成这个需求 优化方案我们将耗时比较短的逻辑判断放入到redis中比如是否库存足够比如是否一人一单这样的操作只要这种逻辑可以完成就意味着我们是一定可以下单完成的我们只需要进行快速的逻辑判断根本就不用等下单逻辑走完我们直接给用户返回成功 再在后台开一个线程后台线程慢慢的去执行queue里边的消息这样程序不就超级快了吗而且也不用担心线程池消耗殆尽的问题因为这里我们的程序中并没有手动使用任何线程池当然这里边有两个难点
第一个难点是我们怎么在redis中去快速校验一人一单还有库存判断
第二个难点是由于我们校验和tomct下单是两个线程那么我们如何知道到底哪个单他最后是否成功或者是下单完成为了完成这件事我们在redis操作完之后我们会将一些信息返回给前端同时也会把这些信息丢到异步queue中去后续操作中可以通过这个id来查询我们tomcat中的下单逻辑是否完成了。 我们现在来看看整体思路当用户下单之后判断库存是否充足只需要导redis中去根据key找对应的value是否大于0即可如果不充足则直接结束如果充足继续在redis中判断用户是否可以下单如果set集合中没有这条数据说明他可以下单如果set集合中没有这条记录则将userId和优惠卷存入到redis中并且返回0整个过程需要保证是原子性的我们可以使用lua来操作
当以上判断逻辑走完之后我们可以判断当前redis中返回的结果是否是0 如果是0则表示可以下单则将之前说的信息存入到到queue中去然后返回然后再来个线程异步的下单前端可以通过返回的订单id来判断是否下单成功。 6.2 秒杀优化-Redis完成秒杀资格判断
需求 新增秒杀优惠券的同时将优惠券信息保存到Redis中 基于Lua脚本判断秒杀库存、一人一单决定用户是否抢购成功 如果抢购成功将优惠券id和用户id封装后存入阻塞队列 开启线程任务不断从阻塞队列中获取信息实现异步下单功能
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中//SECKILL_STOCK_KEY 这个变量定义在RedisConstans中//private static final String SECKILL_STOCK_KEY seckill:stock:stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY voucher.getId(), voucher.getStock().toString());
}完整lua表达式
-- 1.参数列表
-- 1.1.优惠券id
local voucherId ARGV[1]
-- 1.2.用户id
local userId ARGV[2]
-- 1.3.订单id
local orderId ARGV[3]-- 2.数据key
-- 2.1.库存key
local stockKey seckill:stock: .. voucherId
-- 2.2.订单key
local orderKey seckill:order: .. voucherId-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call(get, stockKey)) 0) then-- 3.2.库存不足返回1return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call(sismember, orderKey, userId) 1) then-- 3.3.存在说明是重复下单返回2return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call(incrby, stockKey, -1)
-- 3.5.下单保存用户sadd orderKey userId
redis.call(sadd, orderKey, userId)
-- 3.6.发送消息到队列中 XADD stream.orders * k1 v1 k2 v2 ...
redis.call(xadd, stream.orders, *, userId, userId, voucherId, voucherId, id, orderId)
return 0当以上lua表达式执行完毕后剩下的就是根据步骤3,4来执行我们接下来的任务了
VoucherOrderServiceImpl
Override
public Result seckillVoucher(Long voucherId) {//获取用户Long userId UserHolder.getUser().getId();long orderId redisIdWorker.nextId(order);// 1.执行lua脚本Long result stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));int r result.intValue();// 2.判断结果是否为0if (r ! 0) {// 2.1.不为0 代表没有购买资格return Result.fail(r 1 ? 库存不足 : 不能重复下单);}//TODO 保存阻塞队列// 3.返回订单idreturn Result.ok(orderId);
}6.3 秒杀优化-基于阻塞队列实现秒杀优化
VoucherOrderServiceImpl
修改下单动作现在我们去下单时是通过lua表达式去原子执行判断逻辑如果判断我出来不为0 则要么是库存不足要么是重复下单返回错误信息如果是0则把下单的逻辑保存到队列中去然后异步执行
//异步处理线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR Executors.newSingleThreadExecutor();//在类初始化之后执行因为当这个类初始化好了之后随时都是有可能要执行的
PostConstruct
private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
// 用于线程池处理的任务
// 当初始化完毕后就会去从对列中去拿信息private class VoucherOrderHandler implements Runnable{Overridepublic void run() {while (true){try {// 1.获取队列中的订单信息VoucherOrder voucherOrder orderTasks.take();// 2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error(处理订单异常, e);}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {//1.获取用户Long userId voucherOrder.getUserId();// 2.创建锁对象RLock redisLock redissonClient.getLock(lock:order: userId);// 3.尝试获取锁boolean isLock redisLock.lock();// 4.判断是否获得锁成功if (!isLock) {// 获取锁失败直接返回失败或者重试log.error(不允许重复下单);return;}try {//注意由于是spring的事务是放在threadLocal中此时的是多线程事务会失效proxy.createVoucherOrder(voucherOrder);} finally {// 释放锁redisLock.unlock();}}//aprivate BlockingQueueVoucherOrder orderTasks new ArrayBlockingQueue(1024 * 1024);Overridepublic Result seckillVoucher(Long voucherId) {Long userId UserHolder.getUser().getId();long orderId redisIdWorker.nextId(order);// 1.执行lua脚本Long result stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));int r result.intValue();// 2.判断结果是否为0if (r ! 0) {// 2.1.不为0 代表没有购买资格return Result.fail(r 1 ? 库存不足 : 不能重复下单);}VoucherOrder voucherOrder new VoucherOrder();// 2.3.订单idlong orderId redisIdWorker.nextId(order);voucherOrder.setId(orderId);// 2.4.用户idvoucherOrder.setUserId(userId);// 2.5.代金券idvoucherOrder.setVoucherId(voucherId);// 2.6.放入阻塞队列orderTasks.add(voucherOrder);//3.获取代理对象proxy (IVoucherOrderService)AopContext.currentProxy();//4.返回订单idreturn Result.ok(orderId);}Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {Long userId voucherOrder.getUserId();// 5.1.查询订单int count query().eq(user_id, userId).eq(voucher_id, voucherOrder.getVoucherId()).count();// 5.2.判断是否存在if (count 0) {// 用户已经购买过了log.error(用户已经购买过了);return ;}// 6.扣减库存boolean success seckillVoucherService.update().setSql(stock stock - 1) // set stock stock - 1.eq(voucher_id, voucherOrder.getVoucherId()).gt(stock, 0) // where id ? and stock 0.update();if (!success) {// 扣减失败log.error(库存不足);return ;}save(voucherOrder);}
小总结
秒杀业务的优化思路是什么
先利用Redis完成库存余量、一人一单判断完成抢单业务再将下单业务放入阻塞队列利用独立线程异步下单基于阻塞队列的异步秒杀存在哪些问题 内存限制问题数据安全问题
7、Redis消息队列
7.1 Redis消息队列-认识消息队列
什么是消息队列字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色
消息队列存储和管理消息也被称为消息代理Message Broker生产者发送消息到消息队列消费者从消息队列获取消息并处理消息 使用队列的好处在于 **解耦**所谓解耦举一个生活中的例子就是快递员(生产者)把快递放到快递柜里边(Message Queue)去我们(消费者)从快递柜里边去拿东西这就是一个异步如果耦合那么这个快递员相当于直接把快递交给你这事固然好但是万一你不在家那么快递员就会一直等你这就浪费了快递员的时间所以这种思想在我们日常开发中是非常有必要的。
这种场景在我们秒杀中就变成了我们下单之后利用redis去进行校验下单条件再通过队列把消息发送出去然后再启动一个线程去消费这个消息完成解耦同时也加快我们的响应速度。
这里我们可以使用一些现成的mq比如kafkarabbitmq等等但是呢如果没有安装mq我们也可以直接使用redis提供的mq方案降低我们的部署和学习成本。
7.2 Redis消息队列-基于List实现消息队列
基于List结构模拟消息队列
消息队列Message Queue字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表很容易模拟出队列效果。
队列是入口和出口不在一边因此我们可以利用LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。 不过要注意的是当队列中没有消息时RPOP或LPOP操作会返回null并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。 基于List的消息队列有哪些优缺点 优点
利用Redis存储不受限于JVM内存上限基于Redis的持久化机制数据安全性有保证可以满足消息有序性
缺点
无法避免消息丢失只支持单消费者
7.3 Redis消息队列-基于PubSub的消息队列
PubSub发布订阅是Redis2.0版本引入的消息传递模型。顾名思义消费者可以订阅一个或多个channel生产者向对应channel发送消息后所有订阅者都能收到相关消息。
SUBSCRIBE channel [channel] 订阅一个或多个频道 PUBLISH channel msg 向一个频道发送消息 PSUBSCRIBE pattern[pattern] 订阅与pattern格式匹配的所有频道 基于PubSub的消息队列有哪些优缺点 优点
采用发布订阅模型支持多生产、多消费
缺点
不支持数据持久化无法避免消息丢失消息堆积有上限超出时数据丢失
7.4 Redis消息队列-基于Stream的消息队列
Stream 是 Redis 5.0 引入的一种新数据类型可以实现一个功能非常完善的消息队列。
发送消息的命令 例如 读取消息的方式之一XREAD 例如使用XREAD读取第一个消息 XREAD阻塞方式读取最新的消息 在业务开发中我们可以循环的调用XREAD阻塞方式来查询最新消息从而实现持续监听队列的效果伪代码如下 注意当我们指定起始ID为$时代表读取最新的消息如果我们处理一条消息的过程中又有超过1条以上的消息到达队列则下次获取时也只能获取到最新的一条会出现漏读消息的问题
STREAM类型消息队列的XREAD命令特点
消息可回溯一个消息可以被多个消费者读取可以阻塞读取有消息漏读的风险
7.5 Redis消息队列-基于Stream的消息队列-消费者组
消费者组Consumer Group将多个消费者划分到一个组中监听同一个队列。具备下列特点 创建消费者组 key队列名称 groupName消费者组名称 ID起始ID标示$代表队列中最后一个消息0则代表队列中第一个消息 MKSTREAM队列不存在时自动创建队列 其它常见命令
删除指定的消费者组
XGROUP DESTORY key groupName给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername从消费者组读取消息
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]group消费组名称consumer消费者名称如果消费者不存在会自动创建一个消费者count本次查询的最大数量BLOCK milliseconds当没有消息时最长等待时间NOACK无需手动ACK获取到消息后自动确认STREAMS key指定队列名称ID获取消息的起始ID
“”从下一个未消费的消息开始 其它根据指定id从pending-list中获取已消费但未确认的消息例如0是从pending-list中的第一个消息开始
消费者监听消息的基本思路
STREAM类型消息队列的XREADGROUP命令特点
消息可回溯可以多消费者争抢消息加快消费速度可以阻塞读取没有消息漏读的风险有消息确认机制保证消息至少被消费一次
最后我们来个小对比 7.6 基于Redis的Stream结构作为消息队列实现异步秒杀下单
需求
创建一个Stream类型的消息队列名为stream.orders修改之前的秒杀下单Lua脚本在认定有抢购资格后直接向stream.orders中添加消息内容包含voucherId、userId、orderId项目启动时开启一个线程任务尝试获取stream.orders中的消息完成下单\
修改lua表达式,新增3.6 VoucherOrderServiceImpl
private class VoucherOrderHandler implements Runnable {Overridepublic void run() {while (true) {try {// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 ListMapRecordString, Object, Object list stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(stream.orders, ReadOffset.lastConsumed()));// 2.判断订单信息是否为空if (list null || list.isEmpty()) {// 如果为null说明没有消息继续下一次循环continue;}// 解析数据MapRecordString, Object, Object record list.get(0);MapObject, Object value record.getValue();VoucherOrder voucherOrder BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge(s1, g1, record.getId());} catch (Exception e) {log.error(处理订单异常, e);//处理异常消息handlePendingList();}}}private void handlePendingList() {while (true) {try {// 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0ListMapRecordString, Object, Object list stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1),StreamOffset.create(stream.orders, ReadOffset.from(0)));// 2.判断订单信息是否为空if (list null || list.isEmpty()) {// 如果为null说明没有异常消息结束循环break;}// 解析数据MapRecordString, Object, Object record list.get(0);MapObject, Object value record.getValue();VoucherOrder voucherOrder BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge(s1, g1, record.getId());} catch (Exception e) {log.error(处理pendding订单异常, e);try{Thread.sleep(20);}catch(Exception e){e.printStackTrace();}}}}
}
8、达人探店
8.1、达人探店-发布探店笔记
发布探店笔记
探店笔记类似点评网站的评价往往是图文结合。对应的表有两个 tb_blog探店笔记表包含笔记中的标题、文字、图片等 tb_blog_comments其他用户对探店笔记的评价
具体发布流程 上传接口
Slf4j
RestController
RequestMapping(upload)
public class UploadController {PostMapping(blog)public Result uploadImage(RequestParam(file) MultipartFile image) {try {// 获取原始文件名称String originalFilename image.getOriginalFilename();// 生成新文件名String fileName createNewFileName(originalFilename);// 保存文件image.transferTo(new File(SystemConstants.IMAGE_UPLOAD_DIR, fileName));// 返回结果log.debug(文件上传成功{}, fileName);return Result.ok(fileName);} catch (IOException e) {throw new RuntimeException(文件上传失败, e);}}}注意同学们在操作时需要修改SystemConstants.IMAGE_UPLOAD_DIR 自己图片所在的地址在实际开发中图片一般会放在nginx上或者是云存储上。
BlogController
RestController
RequestMapping(/blog)
public class BlogController {Resourceprivate IBlogService blogService;PostMappingpublic Result saveBlog(RequestBody Blog blog) {//获取登录用户UserDTO user UserHolder.getUser();blog.setUpdateTime(user.getId());//保存探店博文blogService.saveBlog(blog);//返回idreturn Result.ok(blog.getId());}
}8.2 达人探店-查看探店笔记
实现查看发布探店笔记的接口 实现代码
BlogServiceImpl
Override
public Result queryBlogById(Long id) {// 1.查询blogBlog blog getById(id);if (blog null) {return Result.fail(笔记不存在);}// 2.查询blog有关的用户queryBlogUser(blog);return Result.ok(blog);
}8.3 达人探店-点赞功能
初始代码
GetMapping(/likes/{id})
public Result queryBlogLikes(PathVariable(id) Long id) {//修改点赞数量blogService.update().setSql(liked liked 1 ).eq(id,id).update();return Result.ok();
}问题分析这种方式会导致一个用户无限点赞明显是不合理的
造成这个问题的原因是我们现在的逻辑发起请求只是给数据库1所以才会出现这个问题 完善点赞功能
需求
同一个用户只能点赞一次再次点击则取消点赞如果当前用户已经点赞则点赞按钮高亮显示前端已实现判断字段Blog类的isLike属性
实现步骤
给Blog类中添加一个isLike字段标示是否被当前用户点赞修改点赞功能利用Redis的set集合判断是否点赞过未点赞过则点赞数1已点赞过则点赞数-1修改根据id查询Blog的业务判断当前登录用户是否点赞过赋值给isLike字段修改分页查询Blog业务判断当前登录用户是否点赞过赋值给isLike字段
为什么采用set集合
因为我们的数据是不能重复的当用户操作过之后无论他怎么操作都是
具体步骤
1、在Blog 添加一个字段
TableField(exist false)
private Boolean isLike;2、修改代码 Overridepublic Result likeBlog(Long id){// 1.获取登录用户Long userId UserHolder.getUser().getId();// 2.判断当前登录用户是否已经点赞String key BLOG_LIKED_KEY id;Boolean isMember stringRedisTemplate.opsForSet().isMember(key, userId.toString());if(BooleanUtil.isFalse(isMember)){//3.如果未点赞可以点赞//3.1 数据库点赞数1boolean isSuccess update().setSql(liked liked 1).eq(id, id).update();//3.2 保存用户到Redis的set集合if(isSuccess){stringRedisTemplate.opsForSet().add(key,userId.toString());}}else{//4.如果已点赞取消点赞//4.1 数据库点赞数-1boolean isSuccess update().setSql(liked liked - 1).eq(id, id).update();//4.2 把用户从Redis的set集合移除if(isSuccess){stringRedisTemplate.opsForSet().remove(key,userId.toString());}}8.4 达人探店-点赞排行榜
在探店笔记的详情页面应该把给该笔记点赞的人显示出来比如最早点赞的TOP5形成点赞排行榜
之前的点赞是放到set集合但是set集合是不能排序的所以这个时候咱们可以采用一个可以排序的set集合就是咱们的sortedSet 我们接下来来对比一下这些集合的区别是什么
所有点赞的人需要是唯一的所以我们应当使用set或者是sortedSet
其次我们需要排序就可以直接锁定使用sortedSet啦 修改代码
BlogServiceImpl
点赞逻辑代码 Overridepublic Result likeBlog(Long id) {// 1.获取登录用户Long userId UserHolder.getUser().getId();// 2.判断当前登录用户是否已经点赞String key BLOG_LIKED_KEY id;Double score stringRedisTemplate.opsForZSet().score(key, userId.toString());if (score null) {// 3.如果未点赞可以点赞// 3.1.数据库点赞数 1boolean isSuccess update().setSql(liked liked 1).eq(id, id).update();// 3.2.保存用户到Redis的set集合 zadd key value scoreif (isSuccess) {stringRedisTemplate.opsForZSet().add(key, userId.toString(), System.currentTimeMillis());}} else {// 4.如果已点赞取消点赞// 4.1.数据库点赞数 -1boolean isSuccess update().setSql(liked liked - 1).eq(id, id).update();// 4.2.把用户从Redis的set集合移除if (isSuccess) {stringRedisTemplate.opsForZSet().remove(key, userId.toString());}}return Result.ok();}private void isBlogLiked(Blog blog) {// 1.获取登录用户UserDTO user UserHolder.getUser();if (user null) {// 用户未登录无需查询是否点赞return;}Long userId user.getId();// 2.判断当前登录用户是否已经点赞String key blog:liked: blog.getId();Double score stringRedisTemplate.opsForZSet().score(key, userId.toString());blog.setIsLike(score ! null);}点赞列表查询列表
BlogController
GetMapping(/likes/{id})
public Result queryBlogLikes(PathVariable(id) Long id) {return blogService.queryBlogLikes(id);
}BlogService
Override
public Result queryBlogLikes(Long id) {String key BLOG_LIKED_KEY id;// 1.查询top5的点赞用户 zrange key 0 4SetString top5 stringRedisTemplate.opsForZSet().range(key, 0, 4);if (top5 null || top5.isEmpty()) {return Result.ok(Collections.emptyList());}// 2.解析出其中的用户idListLong ids top5.stream().map(Long::valueOf).collect(Collectors.toList());String idStr StrUtil.join(,, ids);// 3.根据用户id查询用户 WHERE id IN ( 5 , 1 ) ORDER BY FIELD(id, 5, 1)ListUserDTO userDTOS userService.query().in(id, ids).last(ORDER BY FIELD(id, idStr )).list().stream().map(user - BeanUtil.copyProperties(user, UserDTO.class)).collect(Collectors.toList());// 4.返回return Result.ok(userDTOS);
}9、好友关注
9.1 好友关注-关注和取消关注
针对用户的操作可以对用户进行关注和取消关注功能。 实现思路
需求基于该表数据结构实现两个接口
关注和取关接口判断是否关注的接口
关注是User之间的关系是博主与粉丝的关系数据库中有一张tb_follow表来标示 注意: 这里需要把主键修改为自增长简化开发。
FollowController
//关注
PutMapping(/{id}/{isFollow})
public Result follow(PathVariable(id) Long followUserId, PathVariable(isFollow) Boolean isFollow) {return followService.follow(followUserId, isFollow);
}
//取消关注
GetMapping(/or/not/{id})
public Result isFollow(PathVariable(id) Long followUserId) {return followService.isFollow(followUserId);
}FollowService
取消关注service
Override
public Result isFollow(Long followUserId) {// 1.获取登录用户Long userId UserHolder.getUser().getId();// 2.查询是否关注 select count(*) from tb_follow where user_id ? and follow_user_id ?Integer count query().eq(user_id, userId).eq(follow_user_id, followUserId).count();// 3.判断return Result.ok(count 0);}关注serviceOverridepublic Result follow(Long followUserId, Boolean isFollow) {// 1.获取登录用户Long userId UserHolder.getUser().getId();String key follows: userId;// 1.判断到底是关注还是取关if (isFollow) {// 2.关注新增数据Follow follow new Follow();follow.setUserId(userId);follow.setFollowUserId(followUserId);boolean isSuccess save(follow);} else {// 3.取关删除 delete from tb_follow where user_id ? and follow_user_id ?remove(new QueryWrapperFollow().eq(user_id, userId).eq(follow_user_id, followUserId));}return Result.ok();}9.2 好友关注-共同关注
想要去看共同关注的好友需要首先进入到这个页面这个页面会发起两个请求
1、去查询用户的详情
2、去查询用户的笔记
以上两个功能和共同关注没有什么关系大家可以自行将笔记中的代码拷贝到idea中就可以实现这两个功能了我们的重点在于共同关注功能。 // UserController 根据id查询用户
GetMapping(/{id})
public Result queryUserById(PathVariable(id) Long userId){// 查询详情User user userService.getById(userId);if (user null) {return Result.ok();}UserDTO userDTO BeanUtil.copyProperties(user, UserDTO.class);// 返回return Result.ok(userDTO);
}// BlogController 根据id查询博主的探店笔记
GetMapping(/of/user)
public Result queryBlogByUserId(RequestParam(value current, defaultValue 1) Integer current,RequestParam(id) Long id) {// 根据用户查询PageBlog page blogService.query().eq(user_id, id).page(new Page(current, SystemConstants.MAX_PAGE_SIZE));// 获取当前页数据ListBlog records page.getRecords();return Result.ok(records);
}接下来我们来看看共同关注如何实现
需求利用Redis中恰当的数据结构实现共同关注功能。在博主个人页面展示出当前用户与博主的共同关注呢。
当然是使用我们之前学习过的set集合咯在set集合中有交集并集补集的api我们可以把两人的关注的人分别放入到一个set集合中然后再通过api去查看这两个set集合中的交集数据。 我们先来改造当前的关注列表
改造原因是因为我们需要在用户关注了某位用户后需要将数据放入到set集合中方便后续进行共同关注同时当取消关注时也需要从set集合中进行删除
FollowServiceImpl
Override
public Result follow(Long followUserId, Boolean isFollow) {// 1.获取登录用户Long userId UserHolder.getUser().getId();String key follows: userId;// 1.判断到底是关注还是取关if (isFollow) {// 2.关注新增数据Follow follow new Follow();follow.setUserId(userId);follow.setFollowUserId(followUserId);boolean isSuccess save(follow);if (isSuccess) {// 把关注用户的id放入redis的set集合 sadd userId followerUserIdstringRedisTemplate.opsForSet().add(key, followUserId.toString());}} else {// 3.取关删除 delete from tb_follow where user_id ? and follow_user_id ?boolean isSuccess remove(new QueryWrapperFollow().eq(user_id, userId).eq(follow_user_id, followUserId));if (isSuccess) {// 把关注用户的id从Redis集合中移除stringRedisTemplate.opsForSet().remove(key, followUserId.toString());}}return Result.ok();
}具体的关注代码
FollowServiceImpl
Override
public Result followCommons(Long id) {// 1.获取当前用户Long userId UserHolder.getUser().getId();String key follows: userId;// 2.求交集String key2 follows: id;SetString intersect stringRedisTemplate.opsForSet().intersect(key, key2);if (intersect null || intersect.isEmpty()) {// 无交集return Result.ok(Collections.emptyList());}// 3.解析id集合ListLong ids intersect.stream().map(Long::valueOf).collect(Collectors.toList());// 4.查询用户ListUserDTO users userService.listByIds(ids).stream().map(user - BeanUtil.copyProperties(user, UserDTO.class)).collect(Collectors.toList());return Result.ok(users);
}9.3 好友关注-Feed流实现方案
当我们关注了用户后这个用户发了动态那么我们应该把这些数据推送给用户这个需求其实我们又把他叫做Feed流关注推送也叫做Feed流直译为投喂。为用户持续的提供“沉浸式”的体验通过无限下拉刷新获取新的信息。
对于传统的模式的内容解锁我们是需要用户去通过搜索引擎或者是其他的方式去解锁想要看的内容 对于新型的Feed流的的效果不需要我们用户再去推送信息而是系统分析用户到底想要什么然后直接把内容推送给用户从而使用户能够更加的节约时间不用主动去寻找。 Feed流的实现有两种模式
Feed流产品有两种常见模式 Timeline不做内容筛选简单的按照内容发布时间排序常用于好友或关注。例如朋友圈
优点信息全面不会有缺失。并且实现也相对简单缺点信息噪音较多用户不一定感兴趣内容获取效率低
智能排序利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户
优点投喂用户感兴趣信息用户粘度很高容易沉迷缺点如果算法不精准可能起到反作用 本例中的个人页面是基于关注的好友来做Feed流因此采用Timeline的模式。该模式的实现方案有三种
我们本次针对好友的操作采用的就是Timeline的方式只需要拿到我们关注用户的信息然后按照时间排序即可
因此采用Timeline的模式。该模式的实现方案有三种
拉模式推模式推拉结合
拉模式也叫做读扩散
该模式的核心含义就是当张三和李四和王五发了消息后都会保存在自己的邮箱中假设赵六要读取信息那么他会从读取他自己的收件箱此时系统会从他关注的人群中把他关注人的信息全部都进行拉取然后在进行排序
优点比较节约空间因为赵六在读信息时并没有重复读取而且读取完之后可以把他的收件箱进行清楚。
缺点比较延迟当用户读取数据时才去关注的人里边去读取数据假设用户关注了大量的用户那么此时就会拉取海量的内容对服务器压力巨大。 推模式也叫做写扩散。
推模式是没有写邮箱的当张三写了一个内容此时会主动的把张三写的内容发送到他的粉丝收件箱中去假设此时李四再来读取就不用再去临时拉取了
优点时效快不用临时拉取
缺点内存压力大假设一个大V写信息很多人关注他 就会写很多分数据到粉丝那边去 推拉结合模式也叫做读写混合兼具推和拉两种模式的优点。
推拉模式是一个折中的方案站在发件人这一段如果是个普通的人那么我们采用写扩散的方式直接把数据写入到他的粉丝中去因为普通的人他的粉丝关注量比较小所以这样做没有压力如果是大V那么他是直接将数据先写入到一份到发件箱里边去然后再直接写一份到活跃粉丝收件箱里边去现在站在收件人这端来看如果是活跃粉丝那么大V和普通的人发的都会直接写入到自己收件箱里边来而如果是普通的粉丝由于他们上线不是很频繁所以等他们上线时再从发件箱里边去拉信息。 9.4 好友关注-推送到粉丝收件箱
需求
修改新增探店笔记的业务在保存blog到数据库的同时推送到粉丝的收件箱收件箱满足可以根据时间戳排序必须用Redis的数据结构实现查询收件箱数据时可以实现分页查询
Feed流中的数据会不断更新所以数据的角标也在变化因此不能采用传统的分页模式。
传统了分页在feed流是不适用的因为我们的数据会随时发生变化
假设在t1 时刻我们去读取第一页此时page 1 size 5 那么我们拿到的就是10~6 这几条记录假设现在t2时候又发布了一条记录此时t3 时刻我们来读取第二页读取第二页传入的参数是page2 size5 那么此时读取到的第二页实际上是从6 开始然后是6~2 那么我们就读取到了重复的数据所以feed流的分页不能采用原始方案来做。 Feed流的滚动分页
我们需要记录每次操作的最后一条然后从这个位置开始去读取数据
举个例子我们从t1时刻开始拿第一页数据拿到了10~6然后记录下当前最后一次拿取的记录就是6t2时刻发布了新的记录此时这个11放到最顶上但是不会影响我们之前记录的6此时t3时刻来拿第二页第二页这个时候拿数据还是从6后一点的5去拿就拿到了5-1的记录。我们这个地方可以采用sortedSet来做可以进行范围查询并且还可以记录当前获取数据时间戳最小值就可以实现滚动分页了 核心的意思就是我们在保存完探店笔记后获得到当前笔记的粉丝然后把数据推送到粉丝的redis中去。
Override
public Result saveBlog(Blog blog) {// 1.获取登录用户UserDTO user UserHolder.getUser();blog.setUserId(user.getId());// 2.保存探店笔记boolean isSuccess save(blog);if(!isSuccess){return Result.fail(新增笔记失败!);}// 3.查询笔记作者的所有粉丝 select * from tb_follow where follow_user_id ?ListFollow follows followService.query().eq(follow_user_id, user.getId()).list();// 4.推送笔记id给所有粉丝for (Follow follow : follows) {// 4.1.获取粉丝idLong userId follow.getUserId();// 4.2.推送String key FEED_KEY userId;stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());}// 5.返回idreturn Result.ok(blog.getId());
}9.5好友关注-实现分页查询收邮箱
需求在个人主页的“关注”卡片中查询并展示推送的Blog信息
具体操作如下
1、每次查询完成后我们要分析出查询出数据的最小时间戳这个值会作为下一次查询的条件
2、我们需要找到与上一次查询相同的查询个数作为偏移量下次查询时跳过这些查询过的数据拿到我们需要的数据
综上我们的请求参数中就需要携带 lastId上一次查询的最小时间戳 和偏移量这两个参数。
这两个参数第一次会由前端来指定以后的查询就根据后台结果作为条件再次传递到后台。 一、定义出来具体的返回值实体类
Data
public class ScrollResult {private List? list;private Long minTime;private Integer offset;
}BlogController
注意RequestParam 表示接受url地址栏传参的注解当方法上参数的名称和url地址栏不相同时可以通过RequestParam 来进行指定
GetMapping(/of/follow)
public Result queryBlogOfFollow(RequestParam(lastId) Long max, RequestParam(value offset, defaultValue 0) Integer offset){return blogService.queryBlogOfFollow(max, offset);
}BlogServiceImpl
Override
public Result queryBlogOfFollow(Long max, Integer offset) {// 1.获取当前用户Long userId UserHolder.getUser().getId();// 2.查询收件箱 ZREVRANGEBYSCORE key Max Min LIMIT offset countString key FEED_KEY userId;SetZSetOperations.TypedTupleString typedTuples stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, 0, max, offset, 2);// 3.非空判断if (typedTuples null || typedTuples.isEmpty()) {return Result.ok();}// 4.解析数据blogId、minTime时间戳、offsetListLong ids new ArrayList(typedTuples.size());long minTime 0; // 2int os 1; // 2for (ZSetOperations.TypedTupleString tuple : typedTuples) { // 5 4 4 2 2// 4.1.获取idids.add(Long.valueOf(tuple.getValue()));// 4.2.获取分数(时间戳long time tuple.getScore().longValue();if(time minTime){os;}else{minTime time;os 1;}}os minTime max ? os : os offset;// 5.根据id查询blogString idStr StrUtil.join(,, ids);ListBlog blogs query().in(id, ids).last(ORDER BY FIELD(id, idStr )).list();for (Blog blog : blogs) {// 5.1.查询blog有关的用户queryBlogUser(blog);// 5.2.查询blog是否被点赞isBlogLiked(blog);}// 6.封装并返回ScrollResult r new ScrollResult();r.setList(blogs);r.setOffset(os);r.setMinTime(minTime);return Result.ok(r);
}10、附近商户
10.1、附近商户-GEO数据结构的基本用法
GEO就是Geolocation的简写形式代表地理坐标。Redis在3.2版本中加入了对GEO的支持允许存储地理坐标信息帮助我们根据经纬度来检索数据。常见的命令有
GEOADD添加一个地理空间信息包含经度longitude、纬度latitude、值memberGEODIST计算指定的两个点之间的距离并返回GEOHASH将指定member的坐标转为hash字符串形式并返回GEOPOS返回指定member的坐标GEORADIUS指定圆心、半径找到该圆内包含的所有member并按照与圆心之间的距离排序后返回。6.以后已废弃GEOSEARCH在指定范围内搜索member并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。6.2.新功能GEOSEARCHSTORE与GEOSEARCH功能一致不过可以把结果存储到一个指定的key。 6.2.新功能
10.2、 附近商户-导入店铺数据到GEO
具体场景说明 当我们点击美食之后会出现一系列的商家商家中可以按照多种排序方式我们此时关注的是距离这个地方就需要使用到我们的GEO向后台传入当前app收集的地址(我们此处是写死的) 以当前坐标作为圆心同时绑定相同的店家类型type以及分页信息把这几个条件传入后台后台查询出对应的数据再返回。 我们要做的事情是将数据库表中的数据导入到redis中去redis中的GEOGEO在redis中就一个menber和一个经纬度我们把x和y轴传入到redis做的经纬度位置去但我们不能把所有的数据都放入到menber中去毕竟作为redis是一个内存级数据库如果存海量数据redis还是力不从心所以我们在这个地方存储他的id即可。
但是这个时候还有一个问题就是在redis中并没有存储type所以我们无法根据type来对数据进行筛选所以我们可以按照商户类型做分组类型相同的商户作为同一组以typeId为key存入同一个GEO集合中即可
代码
HmDianPingApplicationTests
Test
void loadShopData() {// 1.查询店铺信息ListShop list shopService.list();// 2.把店铺分组按照typeId分组typeId一致的放到一个集合MapLong, ListShop map list.stream().collect(Collectors.groupingBy(Shop::getTypeId));// 3.分批完成写入Redisfor (Map.EntryLong, ListShop entry : map.entrySet()) {// 3.1.获取类型idLong typeId entry.getKey();String key SHOP_GEO_KEY typeId;// 3.2.获取同类型的店铺的集合ListShop value entry.getValue();ListRedisGeoCommands.GeoLocationString locations new ArrayList(value.size());// 3.3.写入redis GEOADD key 经度 纬度 memberfor (Shop shop : value) {// stringRedisTemplate.opsForGeo().add(key, new Point(shop.getX(), shop.getY()), shop.getId().toString());locations.add(new RedisGeoCommands.GeoLocation(shop.getId().toString(),new Point(shop.getX(), shop.getY())));}stringRedisTemplate.opsForGeo().add(key, locations);}
}10.3 附近商户-实现附近商户功能
SpringDataRedis的2.3.9版本并不支持Redis 6.2提供的GEOSEARCH命令因此我们需要提示其版本修改自己的POM
第一步导入pom
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactIdexclusionsexclusionartifactIdspring-data-redis/artifactIdgroupIdorg.springframework.data/groupId/exclusionexclusionartifactIdlettuce-core/artifactIdgroupIdio.lettuce/groupId/exclusion/exclusions
/dependency
dependencygroupIdorg.springframework.data/groupIdartifactIdspring-data-redis/artifactIdversion2.6.2/version
/dependency
dependencygroupIdio.lettuce/groupIdartifactIdlettuce-core/artifactIdversion6.1.6.RELEASE/version
/dependency第二步
ShopController
GetMapping(/of/type)
public Result queryShopByType(RequestParam(typeId) Integer typeId,RequestParam(value current, defaultValue 1) Integer current,RequestParam(value x, required false) Double x,RequestParam(value y, required false) Double y
) {return shopService.queryShopByType(typeId, current, x, y);
}ShopServiceImpl
Overridepublic Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {// 1.判断是否需要根据坐标查询if (x null || y null) {// 不需要坐标查询按数据库查询PageShop page query().eq(type_id, typeId).page(new Page(current, SystemConstants.DEFAULT_PAGE_SIZE));// 返回数据return Result.ok(page.getRecords());}// 2.计算分页参数int from (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;int end current * SystemConstants.DEFAULT_PAGE_SIZE;// 3.查询redis、按照距离排序、分页。结果shopId、distanceString key SHOP_GEO_KEY typeId;GeoResultsRedisGeoCommands.GeoLocationString results stringRedisTemplate.opsForGeo() // GEOSEARCH key BYLONLAT x y BYRADIUS 10 WITHDISTANCE.search(key,GeoReference.fromCoordinate(x, y),new Distance(5000),RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end));// 4.解析出idif (results null) {return Result.ok(Collections.emptyList());}ListGeoResultRedisGeoCommands.GeoLocationString list results.getContent();if (list.size() from) {// 没有下一页了结束return Result.ok(Collections.emptyList());}// 4.1.截取 from ~ end的部分ListLong ids new ArrayList(list.size());MapString, Distance distanceMap new HashMap(list.size());list.stream().skip(from).forEach(result - {// 4.2.获取店铺idString shopIdStr result.getContent().getName();ids.add(Long.valueOf(shopIdStr));// 4.3.获取距离Distance distance result.getDistance();distanceMap.put(shopIdStr, distance);});// 5.根据id查询ShopString idStr StrUtil.join(,, ids);ListShop shops query().in(id, ids).last(ORDER BY FIELD(id, idStr )).list();for (Shop shop : shops) {shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());}// 6.返回return Result.ok(shops);}11、用户签到
11.1、用户签到-BitMap功能演示
我们针对签到功能完全可以通过mysql来完成比如说以下这张表 用户一次签到就是一条记录假如有1000万用户平均每人每年签到次数为10次则这张表一年的数据量为 1亿条
每签到一次需要使用8 8 1 1 3 1共22 字节的内存一个月则最多需要600多字节
我们如何能够简化一点呢其实可以考虑小时候一个挺常见的方案就是小时候咱们准备一张小小的卡片你只要签到就打上一个勾我最后判断你是否签到其实只需要到小卡片上看一看就知道了
我们可以采用类似这样的方案来实现我们的签到需求。
我们按月来统计用户签到信息签到记录为1未签到则记录为0.
把每一个bit位对应当月的每一天形成了映射关系。用0和1标示业务状态这种思路就称为位图BitMap。这样我们就用极小的空间来实现了大量数据的表示
Redis中是利用string类型数据结构实现BitMap因此最大上限是512M转换为bit则是 2^32个bit位。 BitMap的操作命令有
SETBIT向指定位置offset存入一个0或1GETBIT 获取指定位置offset的bit值BITCOUNT 统计BitMap中值为1的bit位的数量BITFIELD 操作查询、修改、自增BitMap中bit数组中的指定位置offset的值BITFIELD_RO 获取BitMap中bit数组并以十进制形式返回BITOP 将多个BitMap的结果做位运算与 、或、异或BITPOS 查找bit数组中指定范围内第一个0或1出现的位置
11.2 、用户签到-实现签到功能
需求实现签到接口将当前用户当天签到信息保存到Redis中
思路我们可以把年和月作为bitMap的key然后保存到一个bitMap中每次签到就到对应的位上把数字从0变成1只要对应是1就表明说明这一天已经签到了反之则没有签到。
我们通过接口文档发现此接口并没有传递任何的参数没有参数怎么确实是哪一天签到呢这个很容易可以通过后台代码直接获取即可然后到对应的地址上去修改bitMap。 代码
UserController PostMapping(/sign)public Result sign(){return userService.sign();}UserServiceImpl
Override
public Result sign() {// 1.获取当前登录用户Long userId UserHolder.getUser().getId();// 2.获取日期LocalDateTime now LocalDateTime.now();// 3.拼接keyString keySuffix now.format(DateTimeFormatter.ofPattern(:yyyyMM));String key USER_SIGN_KEY userId keySuffix;// 4.获取今天是本月的第几天int dayOfMonth now.getDayOfMonth();// 5.写入Redis SETBIT key offset 1stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true);return Result.ok();
}11.3 用户签到-签到统计
**问题1**什么叫做连续签到天数 从最后一次签到开始向前统计直到遇到第一次未签到为止计算总的签到次数就是连续签到天数。 Java逻辑代码获得当前这个月的最后一次签到数据定义一个计数器然后不停的向前统计直到获得第一个非0的数字即可每得到一个非0的数字计数器1直到遍历完所有的数据就可以获得当前月的签到总天数了
**问题2**如何得到本月到今天为止的所有签到数据
BITFIELD key GET u[dayOfMonth] 0
假设今天是10号那么我们就可以从当前月的第一天开始获得到当前这一天的位数是10号那么就是10位去拿这段时间的数据就能拿到所有的数据了那么这10天里边签到了多少次呢统计有多少个1即可。
问题3如何从后向前遍历每个bit位
注意bitMap返回的数据是10进制哪假如说返回一个数字8那么我哪儿知道到底哪些是0哪些是1呢我们只需要让得到的10进制数字和1做与运算就可以了因为1只有遇见1 才是1其他数字都是0 我们把签到结果和1进行与操作每与一次就把签到结果向右移动一位依次内推我们就能完成逐个遍历的效果了。
需求实现下面接口统计当前用户截止当前时间在本月的连续签到天数
有用户有时间我们就可以组织出对应的key此时就能找到这个用户截止这天的所有签到记录再根据这套算法就能统计出来他连续签到的次数了 代码
UserController
GetMapping(/sign/count)
public Result signCount(){return userService.signCount();
}UserServiceImpl
Override
public Result signCount() {// 1.获取当前登录用户Long userId UserHolder.getUser().getId();// 2.获取日期LocalDateTime now LocalDateTime.now();// 3.拼接keyString keySuffix now.format(DateTimeFormatter.ofPattern(:yyyyMM));String key USER_SIGN_KEY userId keySuffix;// 4.获取今天是本月的第几天int dayOfMonth now.getDayOfMonth();// 5.获取本月截止今天为止的所有的签到记录返回的是一个十进制的数字 BITFIELD sign:5:202203 GET u14 0ListLong result stringRedisTemplate.opsForValue().bitField(key,BitFieldSubCommands.create().get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0));if (result null || result.isEmpty()) {// 没有任何签到结果return Result.ok(0);}Long num result.get(0);if (num null || num 0) {return Result.ok(0);}// 6.循环遍历int count 0;while (true) {// 6.1.让这个数字与1做与运算得到数字的最后一个bit位 // 判断这个bit位是否为0if ((num 1) 0) {// 如果为0说明未签到结束break;}else {// 如果不为0说明已签到计数器1count;}// 把数字右移一位抛弃最后一个bit位继续下一个bit位num 1;}return Result.ok(count);
}11.4 额外加餐-关于使用bitmap来解决缓存穿透的方案
回顾缓存穿透
发起了一个数据库不存在的redis里边也不存在的数据通常你可以把他看成一个攻击
解决方案 判断id0 如果数据库是空那么就可以直接往redis里边把这个空数据缓存起来
第一种解决方案遇到的问题是如果用户访问的是id不存在的数据则此时就无法生效
第二种解决方案遇到的问题是如果是不同的id那就可以防止下次过来直击数据
所以我们如何解决呢
我们可以将数据库的数据所对应的id写入到一个list集合中当用户过来访问的时候我们直接去判断list中是否包含当前的要查询的数据如果说用户要查询的id数据并不在list集合中则直接返回如果list中包含对应查询的id数据则说明不是一次缓存穿透数据则直接放行。 现在的问题是这个主键其实并没有那么短而是很长的一个 主键
哪怕你单独去提取这个主键但是在11年左右淘宝的商品总量就已经超过10亿个
所以如果采用以上方案这个list也会很大所以我们可以使用bitmap来减少list的存储空间
我们可以把list数据抽象成一个非常大的bitmap我们不再使用list而是将db中的id数据利用哈希思想比如
id % bitmap.size 算出当前这个id对应应该落在bitmap的哪个索引上然后将这个值从0变成1然后当用户来查询数据时此时已经没有了list让用户用他查询的id去用相同的哈希算法 算出来当前这个id应当落在bitmap的哪一位然后判断这一位是0还是1如果是0则表明这一位上的数据一定不存在 采用这种方式来处理需要重点考虑一个事情就是误差率所谓的误差率就是指当发生哈希冲突的时候产生的误差。 12、UV统计
12.1 、UV统计-HyperLogLog
首先我们搞懂两个概念
UV全称Unique Visitor也叫独立访客量是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站只记录1次。PV全称Page View也叫页面访问量或点击量用户每访问网站的一个页面记录1次PV用户多次打开页面则记录多次PV。往往用来衡量网站的流量。
通常来说UV会比PV大很多所以衡量同一个网站的访问量我们需要综合考虑很多因素所以我们只是单纯的把这两个值作为一个参考值
UV统计在服务端做会比较麻烦因为要判断该用户是否已经统计过了需要将统计过的用户信息保存。但是如果每个访问的用户都保存到Redis中数据量会非常恐怖那怎么处理呢
Hyperloglog(HLL)是从Loglog算法派生的概率算法用于确定非常大的集合的基数而不需要存储其所有值。相关算法原理大家可以参考https://juejin.cn/post/6844903785744056333#heading-0 Redis中的HLL是基于string结构实现的单个HLL的内存永远小于16kb内存占用低的令人发指作为代价其测量结果是概率性的有小于0.81的误差。不过对于UV统计来说这完全可以忽略。 12.2 UV统计-测试百万数据的统计
测试思路我们直接利用单元测试向HyperLogLog中添加100万条数据看看内存占用和统计效果如何 经过测试我们会发生他的误差是在允许范围内并且内存占用极小
