黑马点评实战篇总结_黑马点评多少个模块

一共分为11个部分

目录

第一部分短信登录

第二部分缓存

第三部分优惠卷秒杀

第四分布式锁

第五秒杀优化

第六消息队列

第七达人探店

第八好友关注

第九附近商户

第十用户连续签到实现

第十一UV统计

---

第一部分短信登录

MybatisPlus的基本实现流程

首先随便用一个案例来了解一开基本实现流程

POJO实体类

Mapper

Service

ServiceImpl

POJO实体类

记得加上我们的TableName注解和Data注解

 

Mapper

我们弄的Mapper要继承我们的BaseMapper<>,BaseMapper里面放的要是我们的POJO实体类类型

Service

我们要继承IService类，IService<>里面要放的也是我们的POJO实体类类型

ServiceImpl

我们的实现类

继承ServiceImpl<我们的Mapper，我们的实体类类型>

连接GroupService接口

Controller

验证码功能

首先是验证码功能

我们首先看看工具类里面的方法，它是基于这个来弄的

然后使用我们的String类型的一个叫matches（）的方法来进行匹配，regex参数是我们的正则匹配表达式

里面有三个方法，都是基于一个自己写的mismatch方法写的，然后我们第二个参数是我们的常量类定义的正则匹配表达式

记得把某些用户前缀定义成我们的常量类

登录校验拦截器

连接HandlerInterceptor，定义我们的拦截器

保存用户到ThreadLocal然后放行

 

其实这个UserHolder类就是把我们线程池的方法给封装起来

拦截器注册到配置类

我们要多写一个MVC的配置类来配置我们的拦截器

 

把我们刚刚的拦截器写进去

 

 

但我们不可能每一个路径都要拦截，所以我们要配置我们不拦截的路径，我们要放行类似登录和注册的路径
 

exclude
 

• 

  
  
  因为ThreadLocal线程池

因为我们拦截器的时候，已经把用户放到了UserHolder里面去了，那我们就直接从UserHodler里面取就行了

 

隐藏用户敏感信息

思路，我们一些隐私信息返回的时候我们不用给前端，例如密码

所以我们多谢一个DTO实体类，当作返回的类型

然后我们就用BeanUtil中的一个方法，复制属性，然后就把我们的user拷贝到我们的userDTO里面去

 

Session共享问题

我们要是用负载均衡，因为多台tomcat不共享session，所以我们用Redis修改我们的登录逻辑

优化登录逻辑

我们就用手机号来存，这样子就可以保证每个手机号都有key，然后没有冲突

我们用手机号来做Key然后UUID来做token

登录逻辑和平常一样，但是它不使用jwt，所以我们使用手机号作为key来存储的验证码，发送验证码时，验证码存储到我们的Redis中

如果输入的验证码和Redis中取出来的一样，就用“login:token”作为前缀+toekn的值，来存我们的Map，这个Map里面保存的是我们的用户信息

UUID的toString的不同，下面这种带中划线，上面的不带

（流式编程beanToMap）这个是将User对象转成HashMap存储

beanToMap方法可能是将Java对象转换为Map对象的自定义实现。

在beanToMap方法中，传入了三个参数：userDTO对象、一个空的HashMap对象作为目标Map，以及一个CopyOptions对象。

CopyOptions.create()用于创建CopyOptions对象，该对象可以用于设置属性拷贝的选项。在这里，通过链式调用.setIgnoreNullValue(true)设置了忽略源对象中的空值属性，.setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString())设置了一个字段值编辑器，用于将字段值转换为字符串类型。

• 

  
   

我们转成Map，使用putAll简化存储步骤

putAll用map存储，那样的话不管是不是有多个字段都可以一次存储
 

所以我们要将我们的User转成我们的map来存储

我们一样使用我们的BeanUtil

 

我们还要设置我们的token的有效期

 

设置token的有效期

expire

 

弄一个和Session一样的不断刷新逻辑

但我们的session是用户30分钟里不停访问的时候，我们的那个30分钟会不停的刷新
 

所以我们要有一个更新token的逻辑

 

所以我们重新写一下我们的拦截器

数据存到ThreadLocal

然后刷新我们的token的有效期

我们试试在拦截器注入我们的StringRedisTemplate发现失败了，因为这个不是交给IOC容器管理的

---

自定义类依赖注入小问题

我们无法手动注入了，因为这个类是我们自己手动new出来的，不是我们通过Component等等注解去获取的，就是说这个类的对象不是由spring创建的而是我们手动创建的，手动创建的话我们不能直接依赖注入

 

所以我们要生成构造方法

 

然后有个报错，因为我们之前再springmvc里面配置的，所以我们重新放个StringRedisTemplate变量进去

 

因为这个东西configuration是spring创建的，所以我们可以依赖注入

 

我们在MVC配置类Configuration引入了StringRedisTemplate变量

然后在拦截器inteceptor里面的StringRedisTemplate变量，我们也弄了一个构造方法

这样就可以成功注入了

 

---

取出字段时的get和entires的区别

使用get的话，我们只能取出我们存到HASH结构中的一个key，使用entries的话可以一次性取出所有字段，然后用Map的形式取出

 

fillbeanwithmap

我们将我们的Map转换成我们的UserDTO类型

用map来填充我们的bean，第三个参数是是否忽略填充过程中的错误

那我们肯定false不忽略啊

---

StringRedisTemplate要确保存入的类型是String类型的

你看我们存的这个Map是String和Object类型，但是我们StringRedisTemplate用PutAll的时候，

要两个都是String类型才行，所以我们用流式编程来解决

beanToMap

意思是，将我们的userDTO转换成我么的呢HashMap类型

CopyOptions.create()

创建了一个CopyOptions对象

setIgnoreNullValue(true)

指示转换过程中忽略源对象中的空值属性，即不将空值属性复制到目标Map中。

setFieldValueEditor(...)方法接受一个函数作为参数，用于对字段值进行编辑。

将我们的字段值类型转成String类型

这里的编辑函数是(fieldName, fieldValue) -> fieldValue.toString()，它将字段值转换为字符串类型。这意味着转换后的userMap中的字段值将都是字符串类型。

CopyOptions

其实主要在于我们的CpoyOptions工具类，我们在这里是配置我们的Map类型的转换

---

多个拦截器使用

一个拦截全部路径，一个拦截登录路径

前面的拦截器做了很多东西，我们后面的拦截器就只用判断是否需要放行就行了

我们之前把信息存到ThreadLocal中了

我们后面那个拦截器就只要判断我们是否需要拦截就行了（判断ThredLocal里面是否有东西）

如果没有东西的话，就是登录失败，所以我们设置状态码为401

 

重新在MVC配置里面配置一下

registry添加拦截器

我们默认是拦截所有请求，如果我们不放心，我们addPathPatterns("/**") 这样子就是拦截我们的所有请求

order来设置我们的拦截器的优先级

数值越小优先级越高

总结

思路就是两个拦截器一个负责刷新一个负责拦截

---

第二部分缓存

例如我们把商户信息转换成JSON，存在我们的Redis里面

JSONUtil.toJsonStr（）

因为我们取出来的时候，要变回对应的类型，所以我们的JSONUtil.toBean做类型转换

JsonUtil.toBean（）

 

---

缓存更新的最佳实践方案

主动编写业务逻辑，在数据库更新的同时主动更新缓存

实现商铺缓存与数据库的双写一致

• 

  
   
• 
   

• 

  
   

Transational注解

我们的因为涉及到更新这种事务操作，所以要加Transational注解
 

• 

  
   

记得加事务Transational注解，这样子我们错误了就会回滚
 

解决缓存穿透

缓存不存在，然后我们的请求打到数据库

思路

缓存空对象

造成额外的内存消耗

造成短期不一致

布隆过滤器

实现复杂

存在误判的可能性，判断存在的不一定存在，但判断不存在的一定存在

解决商户查询的缓存穿透问题

思路是

如果数据库中不存在，那么我们就写入空值

如果缓存时空值，就结束

如果缓存正常，就返回商户数据

空值“”

不存在的时候我们写入空值

 

isNotBlank的坏处

我们这个isNotBlank的坏处是，我们之前存储了空值“”，所以导致我们还会往下执行代码

所以下面我们还要加一个！=null的判断

 

所以就是

1.所以先从缓存查
 

2.缓存查不到再从数据库查
 

3.查到的数据或者空值，存到缓存里面

解决缓存雪崩
 

原因，redis突然宕机或者大量缓存同时过期

不同key添加不同的TTL

利用Redis集群

给缓存业务添加限流测流

运用多级缓存

 

缓存击穿问题

高并发的key突然失效了

解决方案

互斥锁

逻辑过期

如果获取锁失败后，我们就先返回过期的数据，反正无伤大雅

互斥锁解决缓存击穿问题
 

setnx指令设计一个锁

setnx就是值不存在的时候，我们才执行

String类型的话有一个setnx命令

 

当且仅当这个key不存在的时候执行
 

我们多次setnx我们的lock，我们发现我们保存的值其实没有改变

 

所以setnx就是在key不存在的时候才能往里面写
 

setIfAbsent

使用StringRedisTemplate来操作，setnx其实就是setIfAbsent

---

return的小细节（用BooleanUtil拆箱）

我们不要直接这样子return flag

 

因为直接这样子的话，我们就会进行拆箱，那样的话就会报空指针异常
 

所以我们就要使用一个工具类，BooleanUtil
 

然后我们这样子使用，如果获取锁失败就休眠一小会

然后继续搜索

finally

我们要用try catch来写，因为不管怎么样我们的finally最后都要释放

 

逻辑过期来解决缓存击穿问题

因为我们使用了逻辑过期，所以存null值得这个思路没有了、

我们之前的思路是先找缓存，如果缓存没有的话就去找数据库，如果有就正常写入缓存

如果没有的话，就把空值写入缓存但是这样子大量key同时过期了的话，数据库压力也会过大

所以我们现在的思路就是，我们就算过期的数据，我们也不设置过期时间，我们就把他存在那

因为我们一直存着的，所以如果Redis中缓存没有命中，就是没有，所以返回null

当命中缓存的时候，我们呢就要判断它是否过期，因为我们存了个时间戳进去，然后争夺互斥锁

如果没抢到互斥锁的话，我们就先把Redis中旧的数据返回，反正不要伤害到我们的Mysql数据库就可以了

这个是我们转换成JSON存进Redis的实体类，一个是过期时间，一个是Object

判断是否过期

JSONUtil.toBean，我们拿出来的时候判断是否过期

获取互斥锁

上面的是后面封装好的代码，这个是我们一开始的代码

使用线程池开启新线程，我们开启新线程是用来更新我们的缓存同时释放锁的

然后我们的返回过期的数据

也就是说，我们如果过期了，就更新缓存，然后返回过期的数据

Executors.newFixedThreadPool(10)

1.Executors是Java标准库中的工具类，提供了创建和管理线程池的方法。

2.newFixedThreadPool(10)是Executors类的静态方法之一，用于创建一个固定大小的线程池。

3.该方法接受一个整数参数，表示线程池的大小，这里设置为10，即线程池中同时运行的线程数为10。

4.返回的结果是一个ExecutorService对象，它表示创建的线程池。

---

（涉及泛型）封装Redis工具类

既然是封装我们的方法，所以我们的返回的类型是不确定的，我们要使用泛型

调用我们的人它肯定数据库要怎么做，所以我们这个函数还有个形参，用来传递我们的数据库的逻辑
 

所以我们要传个Function，有参数有返回值的函数，我们叫Function

 

Function回退函数

前面的<R,ID>代表我们这里有两个泛型,我们定义为ID和R

Function：表示数据库回退函数

ID作为参数，R是返回的类型

Function传的其实是方法

它是一个接收标识符ID作为参数并返回结果R的函数

我们这样子使用

然后看看我们使用这个工具类的时候怎么传参数

• 

  
  这个是我们传给Function的函数

• 

  
   

---

第三部分优惠卷秒杀

Redis实现全局唯一ID

这是我们的代码

生成时间戳

首先获取当前时间的LocalDateTime对象now。

通过调用toEpochSecond方法将now转换为自UTC 1970-01-01T00:00:00以来的秒数，得到nowSecond。

将nowSecond减去一个起始时间戳（BEGIN_TIMESTAMP），得到相对于起始时间的偏移量timestamp。

我们这样子是不对的，因为这样子我们就只有一个key，那样子后面量太多导致数值过大怎么办呢

生成序列号

因为我们可能超过序列号上限，所以

我们拼接日期字符串，不同的日期就是不同的key了

首先将当前时间的日期部分格式化为字符串，使用yyyy:MM:dd的格式，得到date。

通过调用Redis的opsForValue().increment方法，对键为icr: + keyPrefix + : + date的值进行自增操作，返回自增后的结果count。

拼接并返回ID

将timestamp左移COUNT_BITS位，通过位运算将时间戳占据高位。

将count与左移后的timestamp进行位或操作，将序列号占据低位，形成最终的ID。

Redis自增ID策略

每天一个key，方便统计订单量（我们设定的是统计当天的，因为我们的拼接的日期是天）

ID构造是 时间戳+计数器

库存超卖问题分析

乐观锁

版本号法

我们可以在表加多一个version字段来进行判断

我们可以加多一个version版本控制

CAS法

记录票数stock，判断前后我们的票数是否一样

cas法修改

我们的逻辑是判断之前的数据是否有被修改过

• 因为假设两个人同时在版本号为1的时候抢100
   
• 那么一个快一个慢
   
• 按理来说慢的那个人还有99个票能抢
   
• 但是按照我们的那个版本更新逻辑来说
   
• 我们是失败了
   
• 但是我们根据业务来分析，其实这并没有什么问题
   
• 所以我们要修改

---

所以我们就只要判断票数stock大于0的时候，那么我们就可以执行就可以了

gt是大于的意思

所以我们把eq改成gt，gt是大于的意思

缺陷

所以乐观锁虽然性能好，但是抢票的成功率会很低

---

悲观锁实现一人一单

乐观锁是我们更新数据的时候用的，我们看看我们的数据是否经过修改

在方法上加锁

我们在我们的这个方法这里加锁synchronized

 

我们事务的范围其实就只有我们封装的那段逻辑，所以我们把我们的Transactional注解拿下来

 

多线程

但是我们这个锁也有问题，我们这样子就是串行的了

我们只是不想同一个用户多次获取，这个是直接把整个方法锁住了，其他用户也拿不到

所以不同的用户我们要用不同的线程才行啊

 

不在方法上加锁，给我们的用户名加锁

因为我们每次的UserId都是不同的，所以我们使用toString

使用toString的问题

但是我们tostring难道就一样了吗
 

我们看看我们的底层，我们其实是new了一段字符串

 

所以tostring后还是不一样的，即使我们的变量值是一样的，但是因为它是new出来的，所以我们内存地址不一样，这样toString（）之后还是不一样
 

每调用一次都是一个全新的字符串对象

intern（）

所以我们加一个intern（）方法

 

我们值一样那么就是一样的了

 

但是要是我们释放锁，然后提交数据，要是这个过程有其他线程进来了呢？

所以我们应该要在我们的函数调用这里上锁

 

这样子就保证了我们是事务提交之后我们才把锁释放的
 

我们现在是只在方法上加了Transactional注解，而不是在整个类上加的

但是我们直接这样return的话，因为在一个类里面，所以它是用this调用的

this调用导致不是代理对象

如果我们用this调用这个方法的话就错误了，因为我们的Transaction注解能生效是因为我们spring来代理对象，我们用this调用的话，那么就不是代理对象了，是非代理对象，它是没有事务功能的

 

使用AopContext.currentProxy()来解决
 

这样子就可以拿到当前对象的代理对象了
 

然后我们用代理对象来调用这个函数

 

引入依赖org.asectj

但是如果我们要这样子做的话，我们要引入这个依赖

 

 

启动类暴露代理对象

我们的代理对象默认是不暴露的

然后我们在启动类配置暴露代理对象，因为我们的代理对象默认是不暴露的

 

这样子我们就可以保证事务生效了
 

---

集群下线程并发安全问题

• 

  
   

我们两个不同的端口都发起请求，发现我们两个都进去了

 

我们的两个服务都收到了请求

 

集群模式下，虽然我们用了sychronic锁
 

但我们还是出问题了

 

集群模式下，不同端口
 

其实是因为我们的JVM不同，所以导致我们的锁不共享

 

---

 

Redis实现全局唯一ID

这是我们的代码

生成时间戳

首先获取当前时间的LocalDateTime对象now。

通过调用toEpochSecond方法将now转换为自UTC 1970-01-01T00:00:00以来的秒数，得到nowSecond。

将nowSecond减去一个起始时间戳（BEGIN_TIMESTAMP），得到相对于起始时间的偏移量timestamp。

我们这样子是不对的，因为这样子我们就只有一个key，那样子后面量太多导致数值过大怎么办呢

生成序列号

因为我们可能超过序列号上限，所以

我们拼接日期字符串，不同的日期就是不同的key了

首先将当前时间的日期部分格式化为字符串，使用yyyy:MM:dd的格式，得到date。

通过调用Redis的opsForValue().increment方法，对键为icr: + keyPrefix + : + date的值进行自增操作，返回自增后的结果count。

拼接并返回ID

将timestamp左移COUNT_BITS位，通过位运算将时间戳占据高位。

将count与左移后的timestamp进行位或操作，将序列号占据低位，形成最终的ID。

Redis自增ID策略

每天一个key，方便统计订单量（我们设定的是统计当天的，因为我们的拼接的日期是天）

ID构造是 时间戳+计数器

库存超卖问题分析

乐观锁

版本号法

我们可以在表加多一个version字段来进行判断

我们可以加多一个version版本控制

CAS法

记录票数stock，判断前后我们的票数是否一样

cas法修改

我们的逻辑是判断之前的数据是否有被修改过

• 因为假设两个人同时在版本号为1的时候抢100
   
• 那么一个快一个慢
   
• 按理来说慢的那个人还有99个票能抢
   
• 但是按照我们的那个版本更新逻辑来说
   
• 我们是失败了
   
• 但是我们根据业务来分析，其实这并没有什么问题
   
• 所以我们要修改

---

所以我们就只要判断票数stock大于0的时候，那么我们就可以执行就可以了

gt是大于的意思

所以我们把eq改成gt，gt是大于的意思

缺陷

所以乐观锁虽然性能好，但是抢票的成功率会很低

---

悲观锁实现一人一单

乐观锁是我们更新数据的时候用的，我们看看我们的数据是否经过修改

在方法上加锁

我们在我们的这个方法这里加锁synchronized

 

我们事务的范围其实就只有我们封装的那段逻辑，所以我们把我们的Transactional注解拿下来

 

多线程

但是我们这个锁也有问题，我们这样子就是串行的了

我们只是不想同一个用户多次获取，这个是直接把整个方法锁住了，其他用户也拿不到

所以不同的用户我们要用不同的线程才行啊

 

不在方法上加锁，给我们的用户名加锁

因为我们每次的UserId都是不同的，所以我们使用toString

使用toString的问题

但是我们tostring难道就一样了吗
 

我们看看我们的底层，我们其实是new了一段字符串

 

所以tostring后还是不一样的，即使我们的变量值是一样的，但是因为它是new出来的，所以我们内存地址不一样，这样toString（）之后还是不一样
 

每调用一次都是一个全新的字符串对象

intern（）

所以我们加一个intern（）方法

 

我们值一样那么就是一样的了

 

但是要是我们释放锁，然后提交数据，要是这个过程有其他线程进来了呢？

所以我们应该要在我们的函数调用这里上锁

 

这样子就保证了我们是事务提交之后我们才把锁释放的
 

我们现在是只在方法上加了Transactional注解，而不是在整个类上加的

但是我们直接这样return的话，因为在一个类里面，所以它是用this调用的

this调用导致不是代理对象

如果我们用this调用这个方法的话就错误了，因为我们的Transaction注解能生效是因为我们spring来代理对象，我们用this调用的话，那么就不是代理对象了，是非代理对象，它是没有事务功能的

 

使用AopContext.currentProxy()来解决
 

这样子就可以拿到当前对象的代理对象了
 

然后我们用代理对象来调用这个函数

 

引入依赖org.asectj

但是如果我们要这样子做的话，我们要引入这个依赖

 

 

启动类暴露代理对象

我们的代理对象默认是不暴露的

然后我们在启动类配置暴露代理对象，因为我们的代理对象默认是不暴露的

 

这样子我们就可以保证事务生效了
 

---

集群下线程并发安全问题

• 

  
   

我们两个不同的端口都发起请求，发现我们两个都进去了

 

我们的两个服务都收到了请求

 

集群模式下，虽然我们用了sychronic锁
 

但我们还是出问题了

 

集群模式下，不同端口
 

其实是因为我们的JVM不同，所以导致我们的锁不共享

---

第四分布式锁

原理

让多个JVM使用同一个锁监视器

分布式锁实现思路

获取锁，SETNX

释放锁，DEL

EXPIRE,为锁添加我们的过期时间

TTL +锁的名字 ，查看我们的锁的过期时间

分布式锁误删问题

原因：释放了别人的锁

解决方法：释放锁的时候进行判断

• 

  
   

问题是，因为业务阻塞，导致我们超时释放锁
 

然后我们的业务突然跑起来了，然后我们的逻辑上再释放锁
 

但是我们的不小心释放的是别的业务的锁，那么就出问题了
 

所以我们释放锁的时候也要进行一下判断

---

不同JVM线程ID冲突问题

我们之前用的不是UUID，而是我们的线程的id
 

我们的线程id，是根据我们创建的线程然后依次递增的
 

但这个是我们的JVM，那么要是我们是两个JVM，那我们的数字不就冲突了？

解决方法

那么我们的就UUID+线程id，这样子就保证了不冲突
 

UUID，如果我们的参数是true的话，其实就是把我们的uuid的横线去掉

 

锁标识一样就释放锁

先得到自己的标识，然后得到锁的标识，如果自己的标识和锁的标识一样，那么我们就释放锁

 

分布式锁原子性问题

• 我们其实还是会有阻塞问题，因为有垃圾回收

  

  
   

我们释放锁的时候阻塞了，导致我们的锁超时，然后释放锁了，
 

但是我们这个线程还是认为我们的锁没有释放，然后释放
 

导致我们的另一个事务的锁，被上一个线程释放了

 

因为判断锁标识和释放锁其实是两个动作，这两个动作之间产生了阻塞导致出现了问题
 

所以我们要让判断锁和释放锁的操作，变成一个原子性的操作

Lua脚本解决原子性问题

RedisTemplate调用Lua脚本

这个是我们自己写的lua脚本

 

execute读取文件

我们是每次都读取文件呢还是提前把我们的文件读取好

因为会产生io流会导致性能不好，所以我们最好提前读取好

 

实现类DefaultRedisScript

是我们的实现类
 

我们用static包括起来，让它把我们的文件提前读取

DefaultRedisScript是Spring Data Redis提供的一个类，用于执行Redis的Lua脚本。

setLocation

设置脚本位置：使用setLocation()方法设置脚本的位置。这里通过new ClassPathResource("unlock.lua")创建了一个ClassPathResource对象，指定了Lua脚本文件的路径。unlock.lua是一个位于类路径下的Lua脚本文件。

setResultType

设置结果类型：使用setResultType()方法设置脚本执行结果的类型。这里将结果类型设置为Long.class，表示脚本执行的返回值将被解析为Long类型。

调用Lua脚本

三个参数

Lua脚本

UNLOCK_SCRIPT：作为第一个参数，表示要执行的Lua脚本。根据之前的代码解析，UNLOCK_SCRIPT是一个已经初始化好的Redis脚本对象。

脚本所需的第一个参数

Collections.singletonList(KEY_PREFIX + name)：作为第二个参数，表示Lua脚本中所需的键参数列表。KEY_PREFIX + name是一个键的字符串，用于构建键参数列表。

所需的第二个参数

ID_PREFIX + Thread.currentThread().getId()：作为第三个参数，表示Lua脚本中所需的其他参数。ID_PREFIX + Thread.currentThread().getId()是一个字符串，用于构建其他参数列表。

这两个参数是通过逗号分隔的单个参数，因此它们实际上是作为 stringRedisTemplate.execute() 方法的第二个参数一起传递的。

这里显示，我们的第二个参数要是List<String>，第三个参数要是Object

Collections.singletonList

Collections.singletonList(KEY_PREFIX+name)

我们运用这个工具就可以把我们的字符串变成我们的集合了

Redisson

1.不可重入

 

A中要先去获取锁然后调用方法B
而B方法里又要去获取同一把锁

如果我们的锁是不可重复的，我们在方法A里获取的锁，等我在方法b的时候又想来获取这把锁，显然是无法获取的这样子就出现死锁的情况了

2.不可重试

 

不能重复获取锁

3.超时释放

 

4.主从一致性

 

我们之前用senx来实现锁，但是实际开发中肯定不是自己写锁的，我们的锁用Redisson来实现

Redisson快速入门

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

配置类配置Redisson客户端

用Configuration注解和Bean注解，将我们的RedissonClient交给我们的IOC容器管理

三步

一，new 一个Config

二，userSingleServer来setAddress配置我们的Redis的地址和setPassword来配置我们的密码

三，Redisson.create()，创建我们的Redisson

我们要用锁的时候，我们就注入RedissonClient对象，然后使用来获取锁

 

getLock获创建

tryLock获取锁

这个tryLock，我们也可以传三个参
也可以选择无参，这样子我们就是默认值了

---

Redisson可重入锁原理

可重入：如果一个线程可以在锁没释放的时候多次获取锁，那么就是可重入了

我们可以多次获取锁，然后我们底层有个东西

底层可以记录我们的重入的次数

可重入锁是使用Hash结构来存储的

Hash结构

Hash结构可以记录多个数据，我们可以记录我们的锁，然后再记录我们的重入的次数

String结构

之前我们使用setnx和setex
 

setnx是判断我们锁是否互斥锁
 

setex是我们的锁的时间
 

这是String结构

 

可是我们的Hash结构可没有这种组合命令
 

所以我们把之前的两步分开
 

先判断锁是否存在
 

再手动来设置我们的获取时间
 

这样的逻辑我们不能再用java逻辑来实现，我们要用lua脚本来实现

lua脚本

Redisson底层

我们看看我们Redisson的底层，发现是把我们的lua脚本给写死了

 

---

锁重试和看门狗机制

锁重试

我们传输了waitTime

第一个参数是获取锁的最大等待时长

 

我们获取锁的时候，我们失败了不直接返回，而是在规定的等待时间里面不断去尝试
 

这个时间结束了我们还是没有获取成功的话，我们就返回false
 

我们如果加了第一个参数，我们就成了可重试的锁了
 

 

第二个参数，锁自动释放失效的时间

最后一个参数就是时间的单位

我们发现他的底层也是类似于得到线程ID，因为防止不同jvm的线程id相同，我们也加一个类似于uuid的那种

 

我们这里有一个WacthDog的超时时间

 

这个是我们看门狗超时时间的默认值

 

默认的锁的超时释放时间是30秒

我们成功了返回空，要是失败了就是执行了一个pttl的命令

 

pttl命令返回的是锁的剩余有效期

 

我们返回的是null，那么就是获取成功，所以我们return true

 

如果不等于null，那就是获取锁失败，失败了那就要继续尝试获取锁
 

如果超时了，那就return false

 

我们不可能直接去尝试获取锁，因为事务还没执行完我们就去获取锁，除了增加cpu负担没有啥用

 

所以我们这里有个subscribe来订阅，看看我们的事务是否结束，结束我们就获取锁
但如果我们在超时时间内，仍然没有等到，那我们就返回false

 

如果我们在等待时间内成功获取到锁，那我们就可以去重试了

 

然后我们这样子不断尝试等待尝试等待，形成一个循环
 

---

看门狗机制
 

当leaseTime等于-1的时候，它会使用看门狗的默认时间

 

putIfabsent，我们用EntryName来放进去

 

这样子不管我们这个锁被冲入了几次，我们拿到的永远是同一个Entry
 

如果第一次进来，我们就有一个renew更新有效期

 

如果我们设计了leaseTime，那我们就没有看门狗了

看门狗机制就是，不断刷新我们锁的超时时间，来等到我们的事务结束

联锁mutilock

联锁的本质其实就是多个独立锁

我们的主节点失效，redis选出新的主节点

 

 

因为之前主从同步未完成，锁已经丢失了

 

所以访问新的节点的时候，我们就会发现我们的锁已经没有了
 

Redisson
 

既然我们的主从节点会出现这种问题，那我们就不用主从节点了咯
 

注册多个RedissonClient

这样子我们就把三个独立的RedissonClient给配置好了

mutilock使用多个RedissonClient

然后我们有三个RedissonClient，就是有了三个不同的锁

 

我们创建联锁，multiLock

这是我们的底层

发现我们的联锁底层是个List集合，那我们就要把这个集合里面的锁都尝试获取一遍，都获取成功了才算是成功

 

所谓的联锁，其实就是多个独立锁
 

 

 

---

 

第五秒杀优化

异步秒杀思路

把我们的之前的同步操作弄成异步操作，缩短了秒杀业务的流程，加快了时间

 

 

把我们的之前的同步操作弄成异步操作，缩短了秒杀业务的流程，加快了时
 

 

我们的这四个步骤我们都是走数据库的，数据库的并发能力很差

 

更何况我们减去库存，做的还是我们的写入操作
 

其实我们举一个简单的例子
 

就是我们是顾客，服务员接待我们从头到尾，然后包括后厨做饭的时间，我们上了菜才算服务结束
 

但是这样子顾客多了怎么办？
 

所以我们就弄一个菜单，给后厨，后厨跟着菜单依次做，然后我们服务员再接待其他用户，然后继续给菜单
 

这样子服务员的效率是不是就大大提高了？

我们判断库存和一人一单还是用去mysql弄，但是这样子会很慢，所以我们就交给我们的redis去做

 

但是我们还是要交给两个人去做，如果我们是这样子串行执行的话，我们反而性能更差了

 

所以我们需要创建我们的独立线程去执行
我们抢单成功后先返回ID
虽然此时我们的订单还没有创建，但是我们保证我们之后会创建
 

• 

  
  因为异步+redis，所以我们的执行流程变短了，性能也得到了很大的提升

  

  
  并发能力大大提高
  
  如何在redis中判断库存充足和一人一单？
   

我们的话肯定要把这个东西缓存到我们的Redis里面才可以

key是我们的优惠卷ID，Value是我们的库存数
将来我们只要判断我们的库存数是否大于0，我们就可以下单了
当我们判断我们的用户确实有购买资格后，我们的库存数记得要减一，相当于我们要在我们的Redis里面提前预减库存

 

Redis完成秒杀资格判断

保存秒杀库存到redis当中

 

lua脚本

这个是我们的lua脚本

重写业务流程

第一个参数lua脚本

第二个参数key（集合类型）

但我们刚刚写脚本的时候，发现我们lua脚本里面是没有key这个参数的，而是ARGV其他类型参数

所以我们不需要传key，所以我们传一个空的集合进去

 

第三个参数其他类型

这是我们的lua脚本写的其他类型参数

 

所以我们的第三个参数，我们穿了三个进去

拿到我们的lua脚本的返回值

我们lua脚本里面写了

0是下单

1是库存不足

2是重复下单

阻塞队列实现秒杀下单优化

阻塞队列的特点
 

当一个线程往这个队列获取元素的时候，如果这个队列里面没有元素，那么我们就阻塞，直到这里有元素后

我们线程下单，所以我们还要写两个东西
 

一个是我们的线程池

一个是我们的线程任务

---

线程池

Executors.newSingleThreadExecutor()

ExecutorService SECKILL_ORDER_EXCUTOR= Executors.newSingleThreadExecutor()

我们先弄一个线程池

BlockingQueue

连接Runnable接口重写run（）方法

线程池实现线程任务

我们的任务应该是在我们的秒杀抢购之前开始
我们项目一启动，我们的用户马上就可以抢购，所以我们的这个类应该在一初始化后马上执行，那该怎么让我们的类一初始化后就马上执行这个任务呢
 

PostConstruct注解

PostConstruct,这个注解就表示在当前类初始化完毕后就立刻执行

我们就是初始化结束后，我们就用线程池开启我们的线程任务

其实我们的线程任务，就是处理我们的订单

BlockingQueue

我们之前没修改的时候，我们的代码就是把我们的订单放到我们的BlockingQueue，然后后面处理

上面的代码是经过后面处理过的

这个是我们修改之后使用的

---

第六消息队列

分为

消费者

生产者

三种结构

List

PubSub

Stream

基于List

我们利用LPUSH和RPOP来实现放进和取出

因为它不想阻塞队列那样阻塞然后等待信息

但是我们有BRPUSH和BRPOP这两个命令可以来实现阻塞效果

优点

基于Redis持久化机制

满足消息的有序性

缺点

无法避免数据丢失

只支持单消费者

基于PubSub

PubSub其实就是我们的发布和订阅

SUBSCRIBE channel xxx订阅一个或者多个评到

PUBLISH channal msg 向某个频道发送消息

PSUBSCRIBE pattern 订阅于Pattern格式匹配的所有频道

优点

支持多生产和消费

缺点

不支持数据持久化

无法避免消息丢失

消息堆积有上限，超出时数据丢失

基于Stream

单消费者模式

XADD

NOMKSTREAM

如果队列不存在，是否自动创建队列，默认是自动创建

MAXLEN

设置消息队列的最大消息数量

ID

标识消息的唯一ID，*代表我们的ID由Redis自动生成

field value

发送到队列中的消息，成为Entry，格式是key-value键值对

block

阻塞时间，如果是0的话就是我们什么时候有就什么时候结束

XRead

阻塞方式来查询最新消息

处理消息的过程中，如果有一条以上的消息到达队列，那我们会出现漏读问题

特点

消息可回溯

可被多个消费者读取

可以阻塞读取

有消息漏读风险

多消费者模式-消费者组

创建消费者组XGROUP

XGROUP CREATE key groupName ID MKSTREAM

key

队列名称

gropuName

消费者组名称

ID

起始ID标识

0标识队列中的第一个信息

$标识队列中的最后一个信息

MKSTREAM

队列不存在时自动创建队列

其他命令

DESTORY

删除指定消费组

CREATECONSUMER ......CONSUMERNAME

给指定消费组添加消费者

DELCONSUMER ......CONSUMERNAME

删除消费组中指定消费者

从消费者组读取信息

group 消费者组名称

consumer 消费者名称，如果消费者不存在，我们就会自动创建一个消费者

count 本次查询的最大数量

BLOCK milliseconds 当没有消息的时候的最长等待时间

NOACK 无需手动ACK，获取到信息后自动确认

STREAMS key 指定队列名称

ID 获取消息的起始ID，“>”表示从下一个未消费的信息开始，0代表的是第一个开始

 

 

所以我们可以自己把刚才基于JVM的bolckqueue的逻辑代码，改成基于Redis的Stream来实现

---

第七达人探店

添加探店笔记

+号，添加探店笔记

MutiPartFile类

这个是我们的上传图片存到本地的逻辑

获取原始文件名

生成新文件名

保存文件到本地

这个其实就是我们封装的静态类，是我们的路径

nginx目录

我们把前端内容部署到我们的nginx的html目录里面

我们的照片部署在了我们的前端的服务器里面

 

Bolg博客实体类

我们的实体类有很多的东西，例如我们要显示的博客图片，是否点赞，我们的姓名，我们的笔记这些很多东西

所以我们点进《我的》后，就要把这些都展示出来

TableField(exist = false)

 

这个表示这两个字段，不属于我们的Blog类对应数据库的那张表tb-blog

查看探店笔记

我们点进探店笔记后，要显示的东西有很多

点赞功能

这个就是我们的点赞的接口

用set字段来记录我们的用户是否点赞过

 

首先为我们的Blog类加一个isLike字段，因为这个我们只是用来前端展示，所以我们表里面没有。所以TableField(exist = false)

 

实现思路

首先获取当前登录用户的ID,存储在userId变量中

然后根据博客ID和用户ID构建一个Redis集合的key (BLOG_LIKED_KEY + id)。使用这个key从Redis中查询该用户是否已经点赞过这篇博客,结果存储在score变量中。

如果我们点赞了，写入数据库，然后我们就存进Redis里面

如果我们取消点赞，我们也重写数据库，然后Redis里面的数据就remove

所以我们如果拿到的是null，那我们就写入

拿到的不是null，我们就remove

判断当前用户是否点赞了

---

点赞排行榜功能

我们想要展示前5位点赞的用户，所以我们使用SortedSet来做

Zset的结构是有score值的

zrange来进行查询

 

sortedSet

我们用zset来操作我们的sortedSet然后添加

 

System.currentTimeMillis()，这个是我们存储的时间戳

如果我们score==null，那么就说明不存在，说明我们没点赞

我们存入zset的时候，我们把当前的时间戳来作为我们的score值

range命令查询top5点赞用户

其实我们的SortedSet他只是用我们的score排序，但是我们存进去的时候还是我们的Set<String>

这个是我们存进去的时候

这个是我们用Stream流拿出来的时候

top5.stream()

将 Set<String> 类型的 top5 转换成一个 Stream

.map(Long::valueOf)

对 Stream 中的每个元素(String 类型的用户 ID)使用 Long::valueOf 方法进行转换,将其转换成 Long 类型。

.collect(Collectors.toList())

将转换后的 Long 类型的元素收集到一个 List 中,最终得到 List<Long> 类型的 ids。

点赞顺序出了问题

这是数据库的问题

in的问题

我们是先传5再传1，但我们用in的时候我们看看数据库里面

先1再5
 

所以我们要order by FIELD,这样子我们手动指定

 

order by FIELD手动指定

我们自己来指定我们的id的顺序

.last("ORDER BY FIELD(id," + idStr + ")"):

这里使用了 .last() 方法,它是 MyBatis-Plus 中的一个拓展方法,

用于在 SQL 查询语句的末尾添加自定义的 SQL 片段。

---

第八好友关注

关注和取关

User之间的关系是博主和粉丝之间的关系，所以我们用一个tb_follow来表示

这是我们点击关注的时候，它判断到底我们是关注还是取关

我们把我们当前用户关注的人存入到我们的Redis里面，我们的用户ID作为key，然后关注的那一堆人是我们的Value

这个是判断我们是否已经关注了

共同关注

我们点进博主首页的时候，我们要发送两个请求

一个是获我们点击的用户的信息

一个是我们笔记的分页查询

共同关注

实现思路

我们要把我们关注了谁放到我们的redis当中去
 

key是当前用户id，value是我们的关注的人的id

 

intersect(key1, key2)，求交集

Feed流的两种模式

Timeline

智能排序

Feed流的三种实现方案

拉模式

推模式

推拉结合

活跃粉丝才推过去，不活跃的粉丝看的时候再临时拉下来看

实现推送到粉丝收件箱

新增Blog保存到数据库的同时，我们推送到粉丝的收件箱

收件箱满足时间戳排序

查询收件箱时，用分页查询

feed流不能采用传统分页

feed流的数据在不断更新，我们的角标也在不断变化，所以不能用传统的分页查询

List能否实现

分页查询也是关键，我们的redis的list能实现分页查询吗？

List结构只能按照角标查询，不支持这样子的滚动分页

使用SortedSet

数据会变动的话，我们可以用sortedset，这样子的话我们可以根据score进行排序
 

我们按照时间戳进行排列，下次查询时我再找比这个时间戳更小的，这样子我们就实现了滚动分页了

我们的博客保存到我们的数据库的同时

我们查询我们的数据库，查到当前博主的所有粉丝

然后创一个新的SortSet结构，来保存博主推送资料的，然后我们推送我们的新的博客给我们的粉丝

滚动分页查询实现收件箱思路

因为我们用角标查询的话，其中有新数据进去，我们查询的话，我们的角标会混乱，我们下一页会查到某些上一页的博客，这样子明显不好

所以我们把从角标查询，变成用我们的score分数来排序查询

我们其实是往我们的socre里面存我们的时间戳，然后根据时间戳大小来排序

我们一开始从最大的开始查3条，例如654，然后我们下一次就找比4小的数据就行了，这样就可以避免角标混乱从而产生的问题了

我们在关注中查询我们关注的博主的Blog的信息

RANGE其实是升序排列的

ZREVRANGE其实是降序排列的

ZREVRANGE

我们这个其实是按照角标进行查询的，所以我们新增数据进来后，我们的查询会出现一些失误

ZREVRANGEBYSCORE

ZREVRANGEBYSCORE

 

WITHSCORES

WITHSCORES是查询我们的数据后要带上我们的分数

OffSet

这个offset其实就是我们的偏移量，就是从最大值开始的第几个开始查询

这个其实就是我们刚刚的那个记住我们之前在哪

我们如果是刚开始的话，我们肯定offset就是0

Count

这个是我们要查询几条

offset变化思路

我们后面要查比5小的，所以我们的max就改成5，min还是0，因为是REVRANGE所以是倒序排序

然后我们的比5小肯定补包括5，所以我们的offset要是1，我们偏移一位

这就是滚动查询，每一次都记录上一次查询的score的最小值

也就是我们的offset第一次是0，其他情况下是1？？？？？真是这样吗

看看当我们的score，也就是我们的时间戳一样的情况，例如我们的score有3个6，

那么我们要查比6小的，我们的offset为1，偏移一位的话？那我们不就是从第二个socre为6的开始查询了吗？

所以我们的offset的值，应该是和我们上一次查询的score的一样的个数

所以第一次offset为0，其他情况下我们的offset的值取决于上一次查询的最后的socre，有几个score是和他一样的

参数

所以我们的请求的参数有两个

一个是lastId

一个是offset偏移量

其他的例如页面大小那些，是写死的

返回值

List<Blog>

minTime

offset

实现滚动分页查询

POJO实体类

Controller

 

ServiceImpl

Set<ZSetOperations.TypedTuple<String>>

typedTuples: 最终得到的结果是一个 Set 集合,集合元素是 ZSetOperations.TypedTuple<String> 类型,它包含了元素(String 类型)和分数。

---

第九附近商户

GEO数据结构的基本用法

GEOADD 添加一个地理空间信息

GEODIST 计算指定的两个点之间的距离并返回

GEOHASH 将指定的member的坐标转为hash字符串形式并返回

GEOPOS 返回指定member的坐标

GEOSEARCH 在指定范围内搜索member，并按照与指定点之间的距离排序后返回，范围可以是圆形或者矩形

GEOSEARCHSTORE 和GEOSEARCH一样的功能，不过可以把结果存储到一个指定的key

GEOADD

 

我们GEO的底层是zset

GEODIST

相差多远的距离

 

 

• 

  
   

 

GEORADIUS

GEOSEARCH

 

WITHDIST

显示相差的距离

 

 

导入我们的GEO数据

 

我们把我们的商户信息分组写入我们的redis

实现附近商户功能

引入依赖

 

lettuce-core 库来连接 Redis 服务器、执行基本的 Redis 操作

实现代码解析

有一些参数可以传，也可以不传

没传坐标时

我们就不根据我们的坐标，我们就之就从我们的数据库查询

计算分页参数

Redis按照距离排序和分页

我们之前存储的时候，是根据我们的类型来分组放进我们的Redis中的

GeoResults

GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo()

这一行代码获取了一个 GeoResults 对象,它包含了 Redis 地理位置查询的结果。

stringRedisTemplate 是一个 Spring Data Redis 提供的模板类,用于简化与 Redis 的交互。

opsForGeo() 方法返回了一个 Redis 地理位置操作接口,用于执行地理位置相关的 Redis 命令。

search（）

这一行开始调用 search() 方法,执行地理位置查询操作。

GeoReference.fromCoordinate(x, y)

这一行创建了一个 GeoReference 对象,表示查询的中心坐标点。

x 和 y 是查询中心点的经度和纬度坐标。

new Distance(5000)

这一行创建了一个 Distance 对象,表示查询的半径距离。

在本例中,查询半径为 5000 米。

RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end)

这一行构建了 Redis 地理位置查询的参数对象。

includeDistance() 表示需要返回每个结果的距离信息。

limit(end) 表示限制返回结果的数量为 end

我们刚刚的那个最大分页参数就是end

results.getContent()

这一行代码从之前获取的 GeoResults 对象中提取出实际的查询结果列表。GeoResults 是一个包装类,内部包含了一个 List<GeoResult<RedisGeoCommands.GeoLocation<String>>> 类型的结果列表。

其实就是，假设我们是查30这里后面，然后我们拿到的数据只有28或者30，那我们就是下一页没有内容

截取我们需要的部分

list.stream().skip(from)

从from这个坐标之后开始

然后添加我们的商户ID到我们的List集合中

然后获取我们的距离Distance

然后放到我们的Map集合中

然后遍历我们的商户，然后返回

---

第十用户连续签到实现

BitMap功能展示

BitMap的底层实现是基于String结构的

数据库存储签到数据的弊端
 

这种方式既耗费内存，对数据库来讲又消耗过大

 

BitMap的用法

操作命令

SETBIT 向指定位置（offset）存入一个0或者1

GETBIT 获取指定位置（offset）的bit值

BITCOUNT 统计BitMap中值为1的bit位的数量

BITFIELD 操作BitMap中bit数组中的指定位置（offset）的值

BITFIELD_RO 获取BitMap中bit数组，并且以十进制形式返回

BITOP 将多个BitMap的结果做位运算

BITPOS 查找bit数组中指定范围内第一个0或者1出现的位置

offset是我们的角标，我们的value可以这样，我们签到了就写1，没签到就写0

 

BITFIELD

我们看看BITFIELD的命令

 

type

是有符号还是没符号

offset

是我们要从多少位开始读

我们是以十进制的方式返回，有正负

 

u代表无符号返回
 

i代表有符号返回

我们u2，就是获取两个bit位

 

 

所以，我们的BITFIELD是读取多个bit位

 

实现签到功能

因为BitMap是基于String结构的，所以我们的相应的操作也封装到我们的字符串当中了

这个是我们的实现代码

 

我们是按月统计签到的

使用 stringRedisTemplate.opsForValue().setBit() 方法在 Redis 中设置一个位值。

键名为前面拼接好的 key。

偏移量为当前日期在本月的第几天减 1(因为数组下标从 0 开始)。

设置的值为 true。

---

统计连续签到

代码解析

获取本月截止到今天位置的所有签到记录

BitFieldSubCommands.create()

这是创建一个 BitFieldSubCommands 对象,用于定义对位图的具体操作。

.get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth))

这里指定要获取位图中当前日期在本月的第几天(从 1 开始)对应的位值。

BitFieldType.unsigned(dayOfMonth) 表示将位图中的值解释为无符号整数。

.valueAt(0)

这里指定要获取位图中指定位置的值,0 表示获取第一个值(也就是当前日期对应的位值)。

遍历循环

---

第十一UV统计

HyperLogLog

PFADD

 测试数百万数据

• 

• 

• 这是我们的统计结果，有误差