一.项目详情
1.1 项目简介
黑马点评项目是前后端分离项目，前端部署在nginx服务器上，后端部署在tomcat上。整个项目需要实现的功能如下图。
1.2 数据库表设计
1.3 前端部署
由于是前后端分离项目，前端页面设计样式都已经完成好了，我们后端只需要部署前端项目即可。
start nginx.exe
然后访问 http://localhost:8080 ，即可看到页面：如果访问失败，详细解决方法我写在另一篇文章了。地址：nginx部署前端项目访问失败解决方法
1.4 后端搭建
后端搭建还是分3步
1.创建Springboot工程项目
2.pom文件导入依赖坐标
3.yml配置文件添加配置信息
这里我直接导入黑马的项目工程来做脚手架，整个架构其实不复杂，就是用mybatis–plus做持久层，业务层，控制层，加上对应注解，调mp提供的api就完了，实体类这些更不用多说，按表映射成实体类。
项目整体结构如下图
二.短信登录
2.1 发送验证码
这里虽然后端配置的端口时8081，但这里的请求仍然时8080，是因为前端使用nginx，反向代理，请求先发送到了nginx服务，然后反向代理再去请求tomcat服务器
@Resource有两个属性name和type。Spring将@Resource注解的name属性解析为bean的名字，而type属性则解析为bean的类型。所以如果使用name属性，则使用byName的自动注入策略，而使用type属性时则使用byType自动注入策略。如果既不指定name也不指定type属性，这时将通过反射机制使用byName自动注入策略。
@Autowired只根据type进行注入，不会去匹配name。
2.2 验证码登录
这里验证码登录逻辑有点复杂，先捋清楚
1.校验提交过来的手机号是否为空
2.获取表单提交过来的手机号和验证码
3.根据key获取redis存的验证码
4.验证码匹配判断
5.不匹配，直接返回提示信息
6.匹配，根据phone查询用户信息
7.判断用户是否存在
8.不存在，创建该用户到数据库中
9.将用户信息存入redis
重点来了
9.1生成随机的token
9.2将User对象转为HashMap存储
9.3存入redis中
9.4设置key有效时间
注意：这里设置key的有效时间是死的，就是不管如何，到时间就key就失效，而我们想要的效果是和session一样，每次登录后，key的有效时间就刷新，只有一直不登录时间达到了设置的时间才会失效（这项功能我们在拦截器里去实现）
10.返回token
//8.1随机生成token，作为登录令牌; 生成key
String token = UUID.randomUUID().toString();
//8.2将User对象转为HashMap存储； 生成value
UserDTO userDTO = new UserDTO();
BeanUtils.copyProperties(user,userDTO);
Map<String, Object> userMap = BeanUtil.beanToMap(userDTO);
//8.3存储
stringRedisTemplate.opsForHash().putAll(LOGIN_USER_KEY+token,userMap);//用定义的常量拼接key值
stringRedisTemplate.expire(LOGIN_USER_KEY+token,30,TimeUnit.MINUTES);//设置有效时间
//返回token
return Result.ok(token);
2.3 登录校验拦截器
填入手机号和验证码之后，点击登录，前端发送如下请求，我们需要把登录用户的基本信息封装传递过去，获取当前登录的用户并返回
那么接下来，如何获取登录用户的信息呢？
答案是我们在登录时就将用户的基本信息封装为userDto对象，存入ThreadLocal中，在me这里直接调用get方法，获取ThreadLocal存的对象基本信息
public class UserHolder {
private static final ThreadLocal<UserDTO> 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();
}
}
如果直接传递一个user对象，会造成信息泄露，用户的所有信息随着请求提交都可以在负载中看到，如下图
所以我们需要创建一个userDto，只定义几个简单属性，这样只是显示一些公开字段，不会泄露密码等敏感信息，返回的时候也是返回userDto对象
使用session时的登录验证凭证，就是session–id，根据客户端发送的session–id获取session，再从session中获取数据
由于session不适合集群模式，就是好几个tomcat服务器，运行时采用轮询策略，每个请求处理的服务器不同，不可能每个服务器都存一份session的用户数据，同份数据存多次，既造成了访问延迟数据更新不及时，也造成了内存浪费。而redis是非常适合的，redis支持横向扩展。
使用redis代替session作为缓存后，无法从客户端发送的请求来获取key，在不明确key的情况下，无法获取存入的value值，如下是使用redis缓存来代替session存user
我们使用token来作为唯一key值，token是自动生成且不重复的值
接下来分析前端代码，如何发送请求时携带key值，其实就是相当于session的发送请求时携带cookie一样。
前端的意思是设置了一个拦截器，当要访问后端请求时拦截器会拦截住，在上面绑定一个token请求，这样到后端就会携带token信息，后端就让你访问对应的redis数据
至此虽然我们获取到了对象信息，但是不只一个业务需要获取登录用户的信息，比如发布博客，发表评论，都需要获取用户名称等信息。我们也不可能每个业务都去添加这段代码，要想再方法执行前统一做一些操作，就用到了我们的拦截器，也可以拦截未登录用户。
拦截器开发：我们使用springmvc提供的拦截器，拦截请求，在执行前做一些操作
如果要使用redis作为缓存对象信息，首先拦截器要获取到stringRedisTemplate对象，才能调用缓存api
需要注意
拦截器在Bean初始化之前它就执行了，所以它肯定是无法获取SpringIOC容器中的内容的。那么我们就让拦截器执行的时候实例化拦截器Bean，在拦截器配置类里面先实例化拦截器，然后再获取
用构造器方法注入
不能使用autowried或resource的原因，这个类的对象是在注入拦截器时，自己new出来的，不是由spring创建的，用构造器注入
而注入拦截器的MvcConfig是由spring构建的，它可以使用自动装配，所以在这里获取stringRedisTemplate，在new的时候作为形参传进去
业务逻辑分析
1.获取请求头的token，并判断是否为空
2.根据请求头的token，获取redis存的value–>用户信息，以map类型接收
3.判断用户是否存在
4.不存在，返回错误信息
5.将查询到value–>map数据转换为userDTO对象
6.将userDTO对象存入ThreadLocal中
7.刷新token有效期
8.放行
拦截器代码很通用，我就直接上代码，里面也给了很详细的注释
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
//1.获取请求头的token
String token = request.getHeader("authorization");//前端页面定义的请求头
if (StrUtil.isBlank(token)){
//不存在拦截,返回401状态码
response.setStatus(401);
return false;
}
//2.基于token获取redis中的用户
Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(RedisConstants.LOGIN_USER_KEY + token);
//3.判读用户是否存在
if (userMap.isEmpty()){
//4.不存在拦截到登录页面,返回401状态码，未授权
response.setStatus(401);
return false;
}
//5.将查询到的Hash数据转为UserDTO对象
//将map中的数据自动填充到对象中，false是忽视错误，有异常直接往外抛就行了
UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
//6.存在，保存用户信息到TreadLocal中
UserHolder.saveUser(userDTO);
//7.刷新token的有效期
stringRedisTemplate.expire(RedisConstants.LOGIN_USER_KEY + token,30, TimeUnit.MINUTES);
//8.放行
return true;
}
拦截器开发完毕后还需要注入，不然不会起作用
@Configuration
public class MvcConfig implements WebMvcConfigurer {
//刚刚拦截器是编写好了，但是还没有注册
@Override
public void addInterceptors(InterceptorRegistry registry) {
registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))
.excludePathPatterns(//排除哪些不被拦截,放行
"/user/code",
"/voucher/**",
"/blog/hot",
"/shop/**",
"/shop-type/**",
"/upload/**",
"/user/login"
);
}
}
拦截器优化
由于之前那个拦截器不是拦截所有请求，比如访问首页，拦截器直接放行，这样使得token登录令牌的有效时间没有被刷新，以下可以解决这个问题。
拦截器的执行顺序是，先注入的先执行，最严谨的还是.order()，设置的值越小，执行顺序越前
登录验证总结：以前用的是session，用户登录后将数据放在session中，拦截器那边判断登录验证，可以根据请求传来的cookie获取session-id，从而获取数据
使用redis后，就不能根据请求获取key了，这里用到前端的token来作为key，登录时存入数据，拦截器做登录验证时，根据key取value，判断value是否为空，来判断是否拦截
而存入ThreadLocal中，是因为后续业务需要获取到用户信息
2.4 退出登录（补充）
用户点击退出登录，我们需要清除ThreadLocal里存放的用户信息，这样前端发送的请求获取不到用户信息，登录拦截器从ThreadLocal中获取用户也为null，被拦截跳到登录界面。
/**
* 登出功能
* @return 无
*/
@PostMapping("/logout")
public Result logout(){
// TODO 实现登出功能
UserHolder.removeUser();
return Result.ok();
}
三.商家查询缓存
3.1 添加商家缓存
业务逻辑
1.先根据id查询redis中的value
2.判断value是否为空
3.若不为空，返回value值
注意，这里从redis中获取的value是json字符串，需要转换成对象才能返回。
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
4.若redis未命中，从数据库查询数据
5.判断数据是否为空
6.为空返回错误信息
7.非空存入redis
注意存入redis时，需要将数据库查询出的对象转换为json字符串
stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(shop));
@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Result queryById(Long id) {
//1.从redis根据id查数据
String shopJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
//2.判断是否存在
if (StrUtil.isNotBlank(shopJson)) {
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
//3.若存在直接返回数据
return Result.ok(shop);
}
//4.从数据库中根据id查数据
Shop shop = getById(id);
//5.若不存在，返回错误信息
if (shop == null){
return Result.fail("店铺不存在");
}
//6.将数据写入redis
stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id,JSONUtil.toJsonStr(shop));
stringRedisTemplate.expire(CACHE_SHOP_KEY + id,30, TimeUnit.MINUTES);
//7.返回数据
return Result.ok(shop);
}
}
3.2 首页列表缓存（p37作业）
首页的这块列表信息是不变动的，因此我们可以将它存入缓存中，避免每次访问时都去查询数据库
前端发送的请求
请求URL：http://localhost:8080/api/shop-type/list（GET）
3.2.1 Stirng缓存策略实现
@Override
public Result queryShopTypeString() {
// 1.从 Redis 中查询商铺缓存
String shopTypeJson = stringRedisTemplate.opsForValue().get(CACHE_SHOP_TYPE_LIST_KEY);
// 2.判断 Redis 中是否存在数据
if (StrUtil.isNotBlank(shopTypeJson)) {
// 2.1.存在，则返回
List<ShopType> shopTypes = JSONUtil.toList(shopTypeJson, ShopType.class);
return Result.ok(shopTypes);
}
// 2.2.Redis 中不存在，则从数据库中查询
List<ShopType> shopTypes = query().orderByAsc("sort").list();
// 3.判断数据库中是否存在
if (shopTypes == null) {
// 3.1.数据库中也不存在，则返回 false
return Result.fail("分类不存在！");
}
// 3.2.数据库中存在，则将查询到的信息存入 Redis
stringRedisTemplate.opsForValue().set(CACHE_SHOP_TYPE_LIST_KEY, JSONUtil.toJsonStr(shopTypes));
// 3.3返回
return Result.ok(shopTypes);
}
3.2.2 List缓存策略实现
@Slf4j
@Service
public class ShopTypeServiceImpl extends ServiceImpl<ShopTypeMapper, ShopType> implements IShopTypeService {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Result queryShopType() {
// 1.从 Redis 中查询商铺缓存
List<String> shopTypeJsonList = stringRedisTemplate.opsForList().range(CACHE_SHOP_TYPE_LIST_KEY, 0, -1);
// 2.判断 Redis 中是否有该缓存
if (shopTypeJsonList != null && !shopTypeJsonList.isEmpty()) {
// 2.1.若 Redis 中存在该缓存，则直接返回
ArrayList<ShopType> typeList = new ArrayList<>();
for (String str : shopTypeJsonList) {
typeList.add(JSONUtil.toBean(str, ShopType.class));
}
//也可以用下面的stream流的方式，其实大差不差，都是遍历，每个数据都转换类型后再操作
// List typeList = shopTypeJsonList.stream().map((shopTypeJson)->{
// ShopType shopType = JSONUtil.toBean(shopTypeJson, ShopType.class);
// return shopType;
// }).collect(Collectors.toList());
return Result.ok(typeList);
}
// 2.2.Redis 中若不存在该数据，则从数据库中查询
List<ShopType> typeList = query().orderByAsc("sort").list();
// 3.判断数据库中是否存在
if (typeList == null || typeList.isEmpty()) {
// 3.1.数据库中也不存在，则返回 false
return Result.fail("分类不存在！");
}
// 3.2数据库中存在，则将查询到的信息存入 Redis
for (ShopType shopType : typeList) {
stringRedisTemplate.opsForList().rightPushAll(CACHE_SHOP_TYPE_LIST_KEY, JSONUtil.toJsonStr(shopType));
}
//下面是stream流的方式
// List shopTypeJson = typeList.stream().map((shopType)-> {
// String jsonStr = JSONUtil.toJsonStr(shopType);
// return jsonStr;
// }).collect(Collectors.toList());
// stringRedisTemplate.opsForList().rightPushAll(CACHE_SHOP_TYPE_LIST_KEY,shopTypeJson);
// 3.3返回
return Result.ok(typeList);
}
}
3.2.3 Zset缓存策略实现
@Override
public Result queryShopTypeZSet() {
// 1.从 Redis 中查询商铺缓存
Set<String> shopTypeJsonSet = stringRedisTemplate.opsForZSet().range(CACHE_SHOP_TYPE_LIST_KEY, 0, -1);
// 2.判断 Redis 中是否有该缓存
if (shopTypeJsonSet.size() != 0) {
// 2.1.若 Redis 中存在该缓存，则直接返回
List<ShopType> shopTypes = new ArrayList<>();
for (String str : shopTypeJsonSet) {
shopTypes.add(JSONUtil.toBean(str, ShopType.class));
}
return Result.ok(shopTypes);
}
// 2.2.若 Redis 中无该数据的缓存，则查询数据库
List<ShopType> shopTypes = query().orderByAsc("sort").list();
// 3.判断数据库中是否存在
if (shopTypes == null || shopTypes.isEmpty()) {
// 3.1.数据库中也不存在，则返回 false
return Result.fail("分类不存在！");
}
// 3.2.数据库中存在，则将查询到的信息存入 Redis
for (ShopType shopType : shopTypes) {
stringRedisTemplate.opsForZSet().add(CACHE_SHOP_TYPE_LIST_KEY,JSONUtil.toJsonStr(shopType),shopType.getSort());
}
// 3.3返回
return Result.ok(shopTypes);
}
3.2.4 列表图片不显示的问题解决
首先看看你的路径改了没
图片路径往nginx目录的html下找
如果你的图片路径改正确了，可以看看下面的另一种情况
而我们无论是断点debug还是redis中查，都发现redis已经缓存了查询的数据，而且数据和从数据库查出来的一致
首先是controller 返回的时候不要再次返回result.ok（xxx）不然会出现嵌套 的问题返回两个data，serviceImpl返回即可。
如果你的service实现类中最后返回的是Result的对象，这里只需要返回调用方法即可，如下
3.3 商家缓存主动更新策略
超时剔除
在根据id查询商家信息时，把信息存入缓存，同时设置有效时间
主动更新策略
先更新数据库，再删除缓存，下次查询就把更新的数据存入缓存。因为数据库更新的时间是较长的，而删除缓存，写入缓存是很快速的，在多线程并发情况下，若使用先删除缓存，再更新数据库的策略，这时注意，在该线程更新数据库时，另一线程进行查询操作，又把旧数据写入到缓存了，后续再查询，直接命中缓存，不查询数据库，造成缓存和数据库不一致。
3.4 缓存穿透
不断发送这样的请求，redis一直未命中，一直查询数据库，给数据库造成很大压力
两种解决方法
缓存空对象
对于不存在的数据也在redis中建立缓存，并设置一个较短的TTL时间，实现比较简单。
根据id查询店铺信息上应用
当数据库查询出的数据为空时，返回错误信息还要把空对象存入缓存中，并设置较短的有效时间。用户再次发送该请求时，直接命中redis缓存的空对象，返回错误信息，不再向下查询数据库，降低数据库压力。
布隆过滤
利用布隆过滤算法，在请求进入Redis之前判断是否存在，不存在直接拒绝请求，但布隆过滤器实现比较复杂。
3.5 缓存雪崩
3.6 缓存击穿
逻辑过期设置的一般不是TTL，设置缓存基本上是一直有效到活动结束后，才移除缓存中数据
之所以会逻辑过期，不是因为有效时间，而是因为数据更新了，缓存也需要更新数据，这时逻辑过期。
获取锁
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);
}
3.7 封装Redis工具类
工具类代码，老师封装的很好，有很多知识点，封装类值得反复咀嚼消化
@Slf4j
@Component
public class CacheClient {
private final StringRedisTemplate stringRedisTemplate;
public CacheClient(StringRedisTemplate stringRedisTemplate){
this.stringRedisTemplate = stringRedisTemplate;
}
public void set(String key, Object value, Long time, TimeUnit unit){
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value),time,unit);
}
public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit){
//设置逻辑过期
RedisData redisData = new RedisData();
redisData.setData(value);
redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
//写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
}
public <R,ID> R queryWithPassThrough(
String keyPrefix, ID id, Class<R> type, Function<ID,R>dbFallback,Long time, TimeUnit unit){
String key = keyPrefix + id;
//1.从redis查询商铺缓存
String 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){
//将空值写入redis
stringRedisTemplate.opsForValue().set(key,"",2,TimeUnit.MINUTES);
//返回错误信息
return null;
}
//6.存在写入redis
this.set(key,r,time,unit);
return r;
}
private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
public <R,ID> R queryWithLogicalExpire(
String keyPrefix, ID id, Class<R> type, Function<ID,R> dbFallback, Long time, TimeUnit unit) {
String key = keyPrefix + id;
//1.从redis查商铺缓存
String json = stringRedisTemplate.opsForValue().get(key);
//2.判断是否存在
if (StrUtil.isBlank(json)) {
//3.存在，缓存中存的null
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 成功，再进行二次判断，查看缓存中是否有数据，因为有可能是别人刚刚重建完释放锁，刚好获取到了
//6.4 开启独立线程，实现缓存重建
CACHE_REBUILD_EXECUTOR.submit(() -> {
try {
//查询数据库
R r1 = dbFallback.apply(id);
//写入redis
this.setWithLogicalExpire(key, r1, time, unit);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
//释放锁
unlock(lockKey);
}
});
}
//6.6返回过期的商铺信息
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);
}
}
四.达人探店
涉及基于List的点赞列表
基于SortedSet的点赞排行榜
4.1 发布博客
根据前端页面提交的数据，我们用blog对象来接收，调用mp提供的接口向博客表里save一条数据，但是用户id前端没有传递过来，需要自己去ThreadLocal中去获取
4.2 查看博客
这里可以使用Dto对象，把两表查询结果放入dto，然后返回dto对象，但是过于繁琐，这里直接在Blog实体类，添加用户头像和用户名称两个属性，并加上mp提供的注解@TableField(exist = false) //当前属性不属于表中字段
逻辑分两步，一是根据传过来的id调用mp提供方法查到blog对象，二是根据userId查询user对象，将user中的icon头像，nickname昵称，存入set到blog对象中，再将blog对象返回
4.3 点赞
这里使用到redis的set集合，key为blog的id，value为user的id，用set的ismembet方法判断，当前集合是否有userId，来判读该博客，用户是否已经点赞过了。每个key代表每条博客，每个key下的value集合代表所有点赞的用户id集合。
点赞功能实现及逻辑代码
/**
* 点赞功能实现及判读逻辑
* @param id
* @return
*/
@Override
public Result likeBlog(Long id) {
//1.获取登录用户
Long userId = UserHolder.getUser().getId();
//2.判读登录用户是否点赞
String key = BLOG_LIKED_KEY + id;
Boolean isMembet = stringRedisTemplate.opsForSet().isMember(key, userId.toString());
//3.若未点赞，可以点赞
if (BooleanUtil.isFalse(isMembet)){
//3.1 数据库点赞数+1
boolean 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 数据库点赞数-1
boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update();
//4.2 清除redis的set集合里的userId
if (isSuccess){
stringRedisTemplate.opsForSet().remove(key, userId.toString());
}
}
return Result.ok();
}
分页查询下的点赞业务
@Override
public Result queryHotBlog(Integer current) {
// 根据用户查询
Page<Blog> page = this.query()
.orderByDesc("liked")
.page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
// 获取当前页数据
List<Blog> records = page.getRecords();
// 查询用户
records.forEach(blog ->{
this.queryBlogUser(blog);
this.isBlogLinked(blog);
});
return Result.ok(records);
}
根据id查询博客详情页下的点赞业务
@Override
public Result queryBlogById(Long id) {
//获取博客对象
Blog blog = this.getById(id);
if (blog == null){
return Result.fail("博客不存在");
}
//存入用户名和头像
queryBlogUser(blog);
//存入是否点赞信息
isBlogLinked(blog);
return Result.ok(blog);
}
调用的工具方法
/**
* 网blog对象存入用户名和头像
* @param blog
*/
private void queryBlogUser(Blog blog) {
Long userId = blog.getUserId();
User user = userService.getById(userId);
blog.setIcon(user.getIcon());
blog.setName(user.getNickName());
}
/**
* 往blog对象填入isLike信息（是否点赞）
* @param blog
*/
private void isBlogLinked(Blog blog) {
//1.获取登录用户
Long userId = UserHolder.getUser().getId();
//2.判读登录用户是否点赞
String key = BLOG_LIKED_KEY + blog.getId();
Boolean isMembet = stringRedisTemplate.opsForSet().isMember(key, userId.toString());
//3.将是否点赞信息set到blog中
blog.setIsLike(BooleanUtil.isTrue(isMembet));
}
4.4 点赞排行榜
我们点击博客，进入详情页，其实是发送了两个请求，一个是根据id返回博客详细信息，另一个是根据id返回点赞排行榜，两个请求如下。
这里使用的策略是用redis的sortSet来代替set，因为set集合是无序的，但排行榜需要显示前5个点赞的用户，要想有序且不重复，只能用sortSet了，把之前的点赞功能，用户id存入set集合改为存入ZSet，使用sorce（key，value）方法来获取该键值的sorce，若没有则返回null，用来代替之前set的ismembet方法
判读是否点赞的工具方法也做了改动
/**
* 往blog对象填入isLike信息（是否点赞）
* @param blog
*/
private void isBlogLinked(Blog blog) {
//1.获取登录用户
UserDTO user = UserHolder.getUser();
if (user == null){
return;//用户未登录，无需查询是否点赞
}
Long userId = user.getId();
//2.判读登录用户是否点赞
String key = BLOG_LIKED_KEY + blog.getId();
Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
//3.将是否点赞信息set到blog中
blog.setIsLike(score != null?true:false);
}
注意：sql语句的 list.in（…, …）查询出来的是后点赞的在前，先点赞的在后，需要我们自定义sql查询,用last最后一条sql语句，手写order by的sql，具体代码如下
/**
* 根据博客id查询点赞排行榜
* @param id
* @return
*/
@Override
public Result queryBlogLikes(Long id) {
String key = BLOG_LIKED_KEY + id;
//1.查询top5的点赞用户 zrange key 0 4
Set<String> top5 = stringRedisTemplate.opsForZSet().range(key, 0, 4);
if (top5 == null || top5.isEmpty()){
return Result.ok(Collections.emptyList());
}
//2.解析除其中的用户id
List<Long> ids = top5.stream().map(Long::valueOf).collect(Collectors.toList());
String idStr = StrUtil.join(",", ids);
//3.根据用户id查询用户 将user处理为userDTO对象 where id (5 , 1) order by field(id, 5, 1)
List<UserDTO> 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);
}
五.好友关注
5.1 关注和取关
业务逻辑：
①关注用户功能的实现，根据前端传递的isFollow的值判断，用户是否已关注该博主，若未关注，传过来的是true，则有关注资格，new出follow对象，将userId和userFollowId都set到follow对象里，再将follow对象save到数据库表中
②查看博客详情页的另一个请求，判断用户是否关注了博主，根据ThreadLocal中获取的userId和传递过来的userFollowId查询count是否大于0，结果返回true或false
代码比较简单，如下
@Override //关注取关功能
public Result follow(Long userFollowId, boolean isFollow) {
Long userId = UserHolder.getUser().getId();
if (isFollow){
Follow follow = new Follow();
follow.setUserId(userId);
follow.setFollowUserId(userFollowId);
save(follow);
}else {//取关，删除 delete from tb_follow where user_id = ? and follow_user_id = ?
remove(new QueryWrapper<Follow>().eq("user_id", userId).eq("follow_user_id", userFollowId));
}
return Result.ok();
}
@Override //用户是否关注了博主
public Result isFolow(Long userFollowId) {
Long userId = UserHolder.getUser().getId();
//查询是否关注 select count(*) from tb_follow where user_id = ? and follow_user_id = ?
Integer count = query().eq("user_id", userId).eq("follow_user_id", userFollowId).count();
return Result.ok(count>0);
}
5.2 查看他人主页
一个根据userId查询用户，一个根据userId对博客分页查询
// 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
@GetMapping("/of/user")
public Result queryBlogByUserId(
@RequestParam(value = "current", defaultValue = "1") Integer current,
@RequestParam("id") Long id) {
// 根据用户查询
Page<Blog> page = blogService.query()
.eq("user_id", id).page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
// 获取当前页数据
List<Blog> records = page.getRecords();
return Result.ok(records);
}
5.3 共同关注
共同关注可以使用redis中的set数据结构，来求两个用户关注集合的交集，那么我们就需要更改关注功能的接口了，不仅要把数据存入数据库follow表中，还有把userId存入redis的set集合里
到这里，我们把关注的用户存入了set集合里，key为用户id，value为被关注用户id。下面需要根据上面图片的请求，编写求交集的方法，交集里都是用户id，先将String转为Long型，再根据useId批量查询user转为userDTO，返回集合。代码如下
@Override
public Result followCommons(Long id) {
Long userId = UserHolder.getUser().getId();
String key1 = FOLLOW_USER_ID + userId; //当前登录用户的关注列表集合
String key2 = FOLLOW_USER_ID + id; //点击查看的用户的关注列表集合
//求交集
Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key1, key2);
if (intersect == null || intersect.isEmpty()){
//无交集
return Result.ok(Collections.emptyList());
}
//解析id集合
List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());
//批量查询用户并转换为userDTO对象
List<UserDTO> userDTOList = userService.listByIds(ids).stream().map(user ->
BeanUtil.copyProperties(user, UserDTO.class))
.collect(Collectors.toList());
return Result.ok(userDTOList);
}
最后结果图
5.4 关注推送
关注推送也叫做 Feed 流，直译为投喂。为用户持续的提供 “沉浸式” 的体验，通过无限下拉刷新获取新的信息。
5.4.1 Feed 流分析
Feed 流产品有两种常见模式：
Timeline：不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。
例如朋友圈
优点：信息全面，不会有缺失。并且实现也相对简单
缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低
智能排序：利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户
例如抖音，快手
优点：投喂用户感兴趣信息，用户粘度很高，容易沉迷
缺点：如果算法不精准，可能起到反作用
本例中的个人页面，是基于关注的好友来做 Feed 流，因此采用 Timeline 的模式。
拉模式：也叫做读扩散
每次读的时候获取消息，内存消耗小，但读操作过于频繁，若用户关注了许多博主，一次要读的消息也是十分多，造成延迟较高
推模式：也叫做写扩散。
发消息时写入粉丝收件箱，内存占用更高，写操作频繁，若博主有许多粉丝，写操作更加繁重
推拉结合模式：也叫做读写混合，兼具推和拉两种模式的优点。
普通博主，粉丝少，可以采用推模式，写操作并不是很繁重
大v博主，粉丝多；分两种粉丝，活跃粉，普通粉；活跃粉，数量少，可以采用推模式；普通粉，数量多，但上线查看少，采用拉模式，什么时候看什么时候拉取。
5.4.2 推送到粉丝收件箱
①修改新增探店笔记的业务，在保存 blog 到数据库的同时，推送到粉丝的收件箱
②收件箱满足可以根据时间戳排序，必须用 Redis 的数据结构实现
③查询收件箱数据时，可以实现分页查询
- Feed 流中的数据会不断更新，所以数据的角标也在变化，因此不能采用传统的分页模式。
这里使用sortSet来实现收件箱 ，先将之前新增保存博客的功能接口修改一下，使得博客发布就能推送到粉丝
代码修改如下
public Result saveBlog(Blog blog) {
// 获取登录用户
UserDTO user = UserHolder.getUser();
blog.setUserId(user.getId());
// 保存探店博文
boolean isSuccess = save(blog);
if (!isSuccess){
return Result.fail("发布失败，请检查重试");
}
// 查询博文作者的所有粉丝
List<Follow> follows = followService.query().eq("follow_user_id", user.getId()).list();
for (Follow follow : follows) {
// 获取粉丝id
Long userId = follow.getUserId();
// 推送笔记id给所有粉丝
String key = "feed:" + userId;
stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());
}
// 返回id
return Result.ok(blog.getId());
}
5.4.3 滚动分页
redis中分析滚动查询
详细代码，分析与注解都在代码上
/**
* 滚动查询，展示博主推送的笔记， 新发布的滚动查询查不到，但是往上滚，前端做了处理，就是刷新重新查询，开始位置在当前最新位置
* @param max
* @param offset
* @return
*/
@Override
public Result queryBlogOfFollow(Long max, Integer offset) {
//获取当前用户
Long userId = UserHolder.getUser().getId();
//查询收件箱 ZREVRANGEBYSCORE key Max Min LIMIT offset count limit是小于等于的意思，小于等于查询的最后时间戳
String key = "feed:" + userId;
Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet()
.reverseRangeByScoreWithScores(key, 0, max, offset, 2);
//非空判断
if (typedTuples == null || typedTuples.isEmpty()){
return Result.ok();
}
//解析数据： blogId，minTime（时间戳）， offset
ArrayList<Object> ids = new ArrayList<>(typedTuples.size());
long minTime = 0; //这个minTime是上次查询的最小时间戳，作为当次查询的最大时间戳来开始查
int os = 1;
for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
//获取博客id转换为Long型并存入ids数组中
ids.add(Long.valueOf(typedTuple.getValue()));
//获取分数 判读得到最后一次的时间戳，以及偏移量
long time = typedTuple.getScore().longValue();
if (time == minTime){
os++;
}else {
minTime = time;
os = 1;
}
}
//根据id查询blog，先把前面保存id的ids数组转为字符串
String idStr = StrUtil.join(",", ids); //由于用mp提供的listByIds是用in方法查，不能保证顺序
List<Blog> blogs = query().in("id", ids).last("order by field(id," + idStr + ")").list();
for (Blog blog : blogs) {
//查询blog有关用户信息
queryBlogUser(blog);
//查询blog是否已被点赞
isBlogLinked(blog);
}
//封装并返回
ScrollResult r = new ScrollResult();
r.setList(blogs);
r.setOffset(os);
r.setMinTime(minTime);
return Result.ok(r);
}
六.优惠券秒杀
涉及Redis的计数器，Lua脚本Redis
分布式锁
Redis的三种消息队列
6.1 全局ID生成器
全局id生成器，是一种在分布式系统下用俩生成全局唯一ID的工具，满足以下特性
ID生成器工具类代码，效率基本可以达到每秒1w个
@Component
public class RedisIdWorker {
//开始时间戳
private static final long BEGIN_TIMESTAMP = 1674086400L;
//序列号位数
private static final int COUNT_BITS = 32;
private StringRedisTemplate stringRedisTemplate;
public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
public long nextId(String keyPrefix){
//1.生成时间戳
LocalDateTime time = LocalDateTime.now();
long nowSecond = time.toEpochSecond(ZoneOffset.UTC);
long timestamp = nowSecond - BEGIN_TIMESTAMP;
//2.生成序列号,redis自增长,redis单个key自增长有上限，2的64次方
//2.1获取当前日期，精确到天
String date = time.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
//3.拼接并返回,不能使用字符串方式拼接
return timestamp << COUNT_BITS | count;//先向左移32位，那么低32位全为0，跟序列号进行或操作
}
/**
* 生成开始时间戳
* @param args
*/
public static void main(String[] args) {
LocalDateTime time = LocalDateTime.of(2023, 1, 19, 0, 0, 0);
long second = time.toEpochSecond(ZoneOffset.UTC);
System.out.println(second);
}
}
6.2 秒杀下单
下单业务本质上就是修改优惠券表中的number字段，再在order表中新增一条订单数据；点击抢购优惠券，发送请求如下
以上是业务逻辑，其中涉及到的表操作
1.seckill_voucher表的根据id查询操作
2.seckill_voucher表的根据id修改剩余数量的操作
3.voucher_order表的新增订单数据操作
6.3 超卖问题
通过加锁的方式去处理多线程问题
悲观锁实现比较简单，操作前获取锁，操作结束才释放锁，让多个线程串行执行，但是你让并发线程串行，效率十分低下
第一种加版本号的方式
逻辑如下
第二种方式CAS法
用版本号方式，发现，用每次查的版本跟操作前查的版本作对比是否一致，那还不如直接查库存，用库存对比，更加简化，逻辑如下
主要是在扣减操作上加上条件，查询的值不一致，就不执行操作
在idea中，代码修改，在更新数据库操作，也就是减库存的时候，加上判断条件，eq比较
//5.扣减库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1")//set stock = stock -1
.eq("voucher_id",voucherId).eq("stock",voucher.getStock()) //where id = ? and stock =?
.update();
如果是匹配库存，成功率很低，如下压力测试，100张只售出23张，但是我是开的有200个线程
成功率低的原因
所以，我们更改库存操作时，加的条件不必是必须前面查到的库存数和更改时查到的库存数一致，我们只需要让更改时的条件为查到的库存数大于0即可，代码修改如下
//5.扣减库存
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();
6.4 一人一单
以下是最后结果图。这个一人一单的单体项目这一节，知识点太多了，像spring框架事务失效、aop代理对象、synchronized锁对象等等，这一节值得多看几遍，我自认为我是没有能力将这些知识点阐述的非常清晰的，所以直接多看这节视频，知识点我就不记录了，就算记录也是模糊不清不能像老师那么浅显易懂。
通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。
我们将服务启动两份，端口分别为 8081 和 8082：ctrl+d复制启动类，添加端口配置
然后修改 nginx 的 conf 目录下的 nginx.conf 文件，配置反向代理；更改完毕后需要重新启动nginx服务
nginx.exe -s reload
现在，用户请求会在这两个节点上负载均衡，再次测试下发现存在线程安全问题。
但是在集群模式下，加锁只是该台jvm给当前这台服务器处理的请求加锁，而集群是多台服务器轮询处理请求，会造成每台服务器都有一个加锁的线程，每台服务器都会有一个新订单创建处理
6.5 分布式锁
分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。
分布式锁的特点：多进程可见、互斥、高可用、高性能（高并发）、安全性
分布式锁的核心是实现多进程之间互斥，而满足这一点的方式有很多，常见的分布式锁有以下三种
6.5.1 初级版本
用redis实现分布式锁
Redis分布式锁原理：基于setnx命令–>key存在的情况下，不更新value，而是返回nil
那么利用key是唯一的特性来加锁，比如一人一单业务，key名称精确到userId，那么同一个用户无论发多少次请求，能成功创建键值的只有一个，因为setnx命令，后面的请求在获取锁创建键值就会失败
redis获取锁命令测试，设置过期时间为10秒的锁，nx–>唯一
以下是结果图，两台机器发送请求，只有一台能够获取锁成功，这就是分布式锁的作用，它的作用域不再是单体项目，单机模式，而是在整个集群模式下，它的锁都生效。
数据库的表中数据变化
6.5.2 误删问题
解决逻辑如下图
其实就是在每次释放锁的时候进行判断，判断当前锁与自己是否一致，不一致可能是别人的锁，不释放
业务流程图
6.5.3 原子性问题
Redis 提供了 Lua 脚本功能，在一个脚本中编写多条 Redis 命令，确保多条命令执行时的原子性。
Lua是一种编程语言，它的基本语法大家可以参考网站：https://www.runoob.com/lua/lua-tutorial.html
我们将释放锁的操作写到Lua脚本中去，直接调用脚本
释放锁的业务逻辑是这样的：
①获取锁中的线程标示
②判断是否与指定的标示（当前线程标示）一致
③如果一致则释放锁（删除）
④如果不一致则什么都不做
Lua脚本代码
-- 这里的 KEYS[1] 就是锁的 key，这里的 ARGV[1] 就是当前线程标识
-- 获取锁中的线程标识 get key
local id = redis.call('get', KEYS[1]);
-- 比较线程标识与锁中的标识是否一致
if (id == ARGV[1]) then
-- 释放锁 del key
return redis.call('del', KEYS[1])
end
return 0
6.5.4 Redisson 分布式锁
基于 setnx 实现的分布式锁存在下面的问题
1.不可重入：同一个线程无法多次获取同一把锁
2.不可重试：获取锁只尝试一次就返回 false，没有重试机制
3.超时释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患
4.主从一致性：如果 Redis 提供了主从集群，主从延同步在延迟，当主机宕机时，如果从机同步主机中的数据，则会出现锁失效
Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网格
它不仅提供了一系列的分布式的 Java 常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。
说白了就是一个封装各种锁且十分完善的工具，前面又白雪了，分布式锁轮子人家已经做的很完善了，咱学的就是个思想
官网地址：https://redisson.org
GitHub 地址：https://github.com/redisson/redisson
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
Redisson类
@Configuration
public class RedisConfig {
@Bean
public RedissonClient redissionClient() {
// 配置类
Config config = new Config();
// 添加 Redis 地址，此处添加了单点的地址，也可以使用 config.useClusterServers() 添加集群地址
config.useSingleServer().setAddress("redis://192.168.2.12:6379").setPassword("123321");
// 创建客户端
return Redisson.create(config);
}
}
使用 Redisson 的分布式锁
@Resource
private RedissonClient redissonClient;
@Test
void testRedisson() throws InterruptedException {
// 获取锁（可重入），指定锁的名称
RLock lock = redissonClient.getLock("anyLock");
// 尝试获取锁，参数分别是：获取锁的最大等待时间（期间会重试过），锁自动释放时间，时间单位
boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
// 判断锁是否获取成功
if (isLock) {
try {
System.out.println("执行业务");
} finally {
//释放锁
lock.unlock();
}
}
}
Redisson 分布式锁原理
如果多个主节点保证锁的话，一个主节点宕机了，其它线程只能获得一个新主节点的锁，获取不到其它两个锁，还会获取失败
这里主要是防止主节点宕机后，其它线程获得新主节点的锁，引起线程安全问题
①不可重入Redis 分布式锁
原理：利用 setnx 的互斥性；利用 ex 避免死锁；释放锁时判断线程标示
缺陷：不可重入、无法重试、锁超时失效
②可重入的 Redis 分布式锁
原理：利用 hash 结构，记录线程标示和重入次数；利用 watchDog 延续锁时间；利用信号量控制锁重试等待
缺陷：Redis 宕机引起锁失效问题
③Redisson 的 multiLock连锁
原理：多个独立的 Redis 节点，必须在所有节点都获取重入锁，才算获取锁成功
缺陷：运维成本高、实现复杂
6.5.5 秒杀优化
为避免所有操作都在数据库上执行，分离成两个线程，一个线程判断用户的购买资格，发现用户有购买资格后再开启一个独立的线程来处理耗时较久的减库存、写订单的操作。
可以将耗时较短的两步操作放到 Redis 中，在 Redis 中处理对应的秒杀资格的判断。Redis 的性能是比 MySQL 要好的。此外，还需要引入异步队列记录相关的信息。
redis部分处理逻辑， Lua脚本封装操作保证原子性， redis这里选择的存储类型为set，因为key不能重复，而set恰好是无序不重复的
需求：
1.新增秒杀优惠券的同时，将优惠券信息保存到 Redis 中
2.基于 Lua 脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
3.如果抢购成功，将优惠券 id 和用户 id 封装后存入阻塞队列
4.开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能
分步实现如下
1.在新增优惠券的业务实现类中，把秒杀优惠券的库存信息保存到redis里
VoucherServiceImpl类中
2.编写lua脚本，按照下面的业务流程逻辑，在脚本中完成业务实现
以下是代码及详细注解
-- 1.参数列表
-- 1.1 优惠券id
local voucherId = ARGV[1]
-- 1.2 用户id
local userId = ARGV[2]
-- 2.数据key
-- 2.1 库存key key 是优惠的业务名称加优惠券id value 是优惠券的库存数
local stockKey = 'seckill:stock:' .. voucherId
-- 2.2 订单key key 也是拼接的业务名称加优惠权id 而value是用户id， 这是一个set集合，凡购买该优惠券的用户都会将其id存入集合中
local orderKey = 'seckill:order:' .. voucherId
-- 3.脚本业务
-- 3.1 判断库存是否充足 get stockKey
if (tonumber(redis.call('get', stockKey)) <= 0) then --将get的value先转为数字类型才能判断比较
-- 3.2 库存不足，返回1
return 1
end
-- 3.3 判断用户是否下单 sismember orderKey userId命令，判断当前key集合中，是否存在该value；返回1存在，0不存在
if (redis.call('sismember', orderKey, userId) == 1) then
--3.4 存在说明是重复下单，返回2
return 2
end
-- 3.5 扣库存
redis.call('incrby', stockKey, -1)
-- 3.6 下单（保存用户）
redis.call('sadd', orderKey, userId)
return 0
3.java代码中执行lua脚本，并判断，抢购成功的生成订单并存入阻塞队列
首先注入脚本
private IVoucherOrderService proxy;//定义代理对象，提前定义后面会用到
//注入脚本
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}
其次运行脚本，且判断不满足的请求直接返回提示信息
@Override
public Result seckillVoucher(Long voucherId) { //使用lua脚本
//获取用户
Long userId = UserHolder.getUser().getId();
//1.执行lua脚本
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT,
Collections.emptyList(), //这里是key数组，没有key，就传的一个空集合
voucherId.toString(), userId.toString()
);
//2.判断结果是0
int r = result.intValue();//Long型转为int型，便于下面比较
if (r != 0){
//2.1 不为0，代表没有购买资格
return Result.fail(r == 1?"优惠券已售罄":"不能重复购买");
}
最后是将满足条件的给存放进阻塞队列中
创建一个BlockingQueue阻塞队列
BlockingQueue这个阻塞队列特点：当一个线程尝试从队列获取元素的时候，如果没有元素该线程阻塞，直到队列中有元素才会被唤醒并获取元素
//创建阻塞队列 这个阻塞队列特点：当一个线程尝试从队列获取元素的时候，如果没有元素该线程阻塞，直到队列中有元素才会被唤醒获取
private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);//初始化阻塞队列的大小
接下来就是将满足条件的请求，给生成订单，并把订单对象add到阻塞队列中，接上面的代码，完成整个第三步
//2.2 为0，有购买资格，把下单信息保存到阻塞队列中
//7.创建订单 向订单表新增一条数据，除默认字段，其他字段的值需要set
VoucherOrder voucherOrder = new VoucherOrder();
//7.1订单id
long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
//7.2用户id
voucherOrder.setUserId(userId);
//7.3代金券id
voucherOrder.setVoucherId(voucherId);
//放入阻塞对列中
orderTasks.add(voucherOrder);
//获取代理对象
proxy = (IVoucherOrderService) AopContext.currentProxy();
//3.返回订单id
return Result.ok(orderId);
}
4.开启线程任务，实现异步下单功能
首先创建一个线程池
再定义一个线程任务，但是注意，线程任务需要在用户秒杀订单之前开始，用户一但开始秒杀，队列就会有新的订单，线程任务就应该立即取出订单信息，这里利用spring提供的注解，在类初始化完毕后立即执行线程任务，详细代码如下
//创建线程池
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
//利用spring提供的注解，在类初始化完毕后立即执行线程任务
@PostConstruct
private void init(){
SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
线程任务代码如下
//创建线程任务，内部类方式
private class VoucherOrderHandler implements Runnable{
@Override
public void run() {
//1.获取队列中的订单信息
try {
VoucherOrder voucherOrder = orderTasks.take();
//2.创建订单，这是调之前那个创建订单的方法，需要稍作改动
handleVoucherOrder(voucherOrder);
} catch (Exception e) {
log.info("异常信息:",e);
}
}
}
创建调用的handleVoucherOrder方法，这里的获取锁操作只是做最后的兜底，以防万一，因为前面lua脚本都已经判断过了
private void handleVoucherOrder(VoucherOrder voucherOrder) {
Long userId = voucherOrder.getUserId();
//创建锁对象
SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
//获取锁
boolean isLock = lock.tryLock(1200);
//判断是否获取锁成功
if (!isLock){
log.error("您已购买过该商品，不能重复购买");
}
try {
proxy.createVoucherOrder(voucherOrder);//使用代理对象，最后用于提交事务
} catch (IllegalStateException e) {
throw new RuntimeException(e);
} finally {
lock.unlock();//释放锁
}
}
createVoucherOrder创建订单方法，这里一人一单的其实也不必判读了，lua脚本都写好了，这里只是兜底
@Transactional
public void createVoucherOrder(VoucherOrder voucherOrder){
Long voucherId = voucherOrder.getVoucherId();
//5.一人一单
Long userId = voucherOrder.getId();
//5.1查询订单
int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
//5.2判断是否存在
if (count > 0){
log.error("您已经购买过了");
}
//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){
log.error("库存不足！");
}
this.save(voucherOrder);
}
最后来分析以下整个优化思路
①编写lua脚本，对于超卖问题和一人一单进行解决处理，超卖用CAS方法判断库存是否大于0，一人一单用redis的set集合的sismenber判读该优惠券（key）下的用户id（value）是否唯一
②Java代码中注入脚本，并执行脚本判断脚本返回结果，若不为脚本结果0，直接返回错误提示
③若脚本结果为0，代表有购买优惠券资格，将new VoucherOrder创建订单对象，并set orderId，userId，voucherId。再把订单对象放入阻塞队列中，返回订单id给用户
④创建线程池，并定义线程任务，但注意，线程任务必须在方法执行前执行，使用到spring提供的注解在类初始化完成后执行线程任务
⑤线程任务中获取阻塞队列的订单对象，然后调用handleVoucherOrder方法传入voucherOrder
⑥handleVoucherOrder方法其实是再次获取锁，这个就是个纯兜底，作用不大。并在获取锁成功后调用createVoucherOrder方法扣减库存创建订单，由于都是对数据库的操作，因此要提交事务
至此，整个秒杀业务优化完毕
秒杀业务的优化思路是什么？
先利用 Redis 完成库存余量、一人一单判断，完成抢单业务
再将下单业务放入阻塞队列，利用独立线程异步下单
基于阻塞队列的异步秒杀存在哪些问题？
内存限制问题 —> 我们使用的是JDK里的阻塞队列，是基于JVM的内存，高并发海量请求下造成内存溢出还有服务宕机情况下内存数据丢失
数据安全问题
个人认为还存在的问题
队列满了怎么办 ？
子线程下单失败怎么办？
订单太多了超过阻塞队列大小了怎么办？
拒绝策略怎么设计？
待消费的消息是否应该持久化，不然宕机了消息不就丢失了？
还有如何确保消息确实被消费成功了，不然消费失败了无法重试
6.5.6 消息队列
由于前面的阻塞队列是基于JVM的内存实现，那么不可避免的两个大问题，①高并发海量访问，创建订单，队列很快就超出上限造成内存溢出；②JVM内存没有持久化机制，若服务出现重启或宕机，阻塞队列中的所有任务都会丢失。所以我们使用MQ
MQ是JVM以外的服务，不受JVM内存限制，且MQ中的所有消息会做持久化，这样即使重启或宕机，数据不会丢失。消息投递给消费者后需要消费者确认，未确认消息会一直存在下一次继续投递，确保消息至少被消费一次
6.5.6.1 基于 List 结构模拟消息队列
Redis 的 list 数据结构是一个双向链表
队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP 来实现。
不过要注意的是，当队列中没有消息时 RPOP 或 LPOP 操作会返回 null，并不像 JVM 的阻塞队列那样会阻塞并等待消息。
因此这里应该使用 BRPOP 或者 BLPOP 来实现阻塞效果。
优点
利用 Redis 存储，不受限于 JVM 内存上限
基于 Redis 的持久化机制，数据安全性有保证
可以满足消息有序性
6.5.6.2 基于 PubSub 的消息队列
PubSub（发布订阅） 是 Redis 2.0 版本引入的消息传递模型。
顾名思义，消费者可以订阅一个或多个channel，生产者向对应 channel 发送消息后，所有订阅者都能收到相关消息。
SUBSCRIBE channel [channel] ：订阅一个或多个频道
PUBLISH channel msg ：向一个频道发送消息
PSUBSCRIBE pattern[pattern] ：订阅与 pattern 格式匹配的所有频道
pattern – 通配符方式
?：匹配一个字符
*：匹配多个字符
ae：匹配括号内存在的字符
优点：采用发布订阅模型，支持多生产、多消费
缺点：
不支持数据持久化
无法避免消息丢失
消息堆积有上限，超出时数据丢失
6.5.6.3 基于 Stream 的消息队列
单消费模式
发送消息的命令
XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value …]
key：队列名称
[NOMKSTREAM]：如果队列不存在时，确定是否自动创建队列，默认自动创建
[MAXLEN|MINID [=|~] threshold [LIMIT count]]：设置消息队列的最大消息数量
|ID：消息的唯一 ID， 代表由 Redis 自动生成，格式是 ”时间戳-递增数字“，例如：”1666161469358-0“
field value [field value …]：发送到队列中的消息，称为 Entry。格式为多个 Key-Value 键值对。
例如：创建名为 users 的队列，并向其中发送一个消息，内容是：{name=jack,age=21}，并且使用 Redis 自动生成 ID
127.0.0.1:6379> XADD users * name jack age 21 “1644805700523-0”
读取消息命令
读取消息的方式之一：XREAD
[COUNT count]：每次读取消息的最大数量；
[BLOCK milliseconds]：当没有消息时，确定是否阻塞，阻塞则添加具体的 milliseconds （阻塞时长）
STREAMS key [key …]：从哪个队列读取消息，Key 就是队列名；
ID [ID …]：起始 ID，只返回大于该 ID 的消息；0 代表从第一个消息开始，$ 代表从最新的消息开始。
例如，使用 XREAD 读取第一个消息
127.0.0.1:6379> XREAD COUNT 1 STREAMS users 0
1) 1) "queue"
2) 1) 1) "1666169070359-0"
2) 1) "name"
2) "jack"
3) "age"
4) 20
XREAD COUNT 1 BLOCK STREAMS queue $
注意：
当我们指定起始 ID 为 $ 时，代表读取最新的消息
如果我们处理一条消息的过程中，又有超过 1 条以上的消息到达队列，则下次获取时也只能获取到最新的一条
如此便会出现漏读消息的问题
STREAM 类型消息队列的 XREAD 命令特点：
1.消息可回溯（消息永久的保存在消息队列中）
2.一个消息可以被多个消费者读取
3.可以阻塞读取
4.有消息漏读的风险（缺点）
消费者组模式
消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。
其具备下列特点：
消息分流：队列中的 消息会分流给组内不同的消费者，而不是重复消费，从而加快消息处理的速度。
消息标示：消费者组会维护一个标示，记录最后一个被处理的消息，即使消费者宕机重启，还会从标示之后读取消息，确保每一个消息都会被消费。（解决漏读问题）
消息确认：消费者获取消息后，消息处于 pending 状态，并存入一个 pending-list。
当处理完成后需要通过 XACK 命令来确认消息，标记消息为已处理，才会从 pending-list 中移除。（解决消息丢失问题)
创建消费者组
XGROUP CREATE key groupName ID [MKSTREAM]
key：队列名称
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 命令特点
消息可回溯
可以多消费者争抢消息，加快消费速度
可以阻塞读取
没有消息漏读的风险
有消息确认机制，保证消息至少被消费一次
Redis 三种消息队列的对比
Stream消息队列异步秒杀下单
需求：
①创建一个 Stream 类型的消息队列，名为 stream.orders
②修改之前的秒杀下单 Lua 脚本，在认定有抢购资格后，直接向 stream.orders 中添加消息，内容包含 voucherId、userId、orderId
③项目启动时，开启一个线程任务，尝试获取 stream.orders 中的消息，完成下单
XGROUP CREATE stream.orders g1 0 MKSTREAM
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
local stockKey_value = redis.call('get', stockKey)
-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if (tonumber(stockKey_value) <= 0) then
-- 3.2.库存不足，返回 1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if (redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在，则说明该用户是重复下单（这是不允许的），则返回 2
return 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
七.附近商户
Redis的GeoHash应用
7.1 GEO 数据结构基本用法
GEO 就是 Geolocation 的简写形式，代表地理坐标。
Redis 在 3.2 版本中加入了对 GEO 的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。
常见的命令有：
-
GEOPOS：返回指定member的坐标
-
GEORADIUS：指定圆心、半径，找到该圆内包含的所有 member，并按照与圆心之间的距离排序后返回。6.2 以后已废弃
-
GEOSEARCH：在指定范围内搜索 member，并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。6.2 新功能
-
GEOSEARCHSTORE：与 GEOSEARCH 功能一致，不过可以把结果存储到一个指定的 key。 6.2.新功能
Redis中使用示范
7.2 查找附近商铺功能
7.2.1导入店铺数据到Redis的GEO
@Test
void loadShopData(){
//查询店铺信息
List<Shop> list = shopService.list();
//把店铺分组，按照typeId分组，typeId一致的放到一个集合
Map<Long, List<Shop>> map = list.stream().collect(Collectors.groupingBy(Shop::getTypeId));
//分批完成写入Redis
for (Map.Entry<Long, List<Shop>> entry : map.entrySet()) {
//获取类型id
Long typeId = entry.getKey();
String key = "shop:geo" + typeId;
//获取同类型的店铺的集合
List<Shop> value = entry.getValue();
List<RedisGeoCommands.GeoLocation<String>> locations = new ArrayList<>(value.size());
//写入redis GEOADD key 经度 纬度 member
for (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);
}
}
7.2.2 实现查找附近商铺功能
SpringDataRedis 的 2.3.9 版本并不支持 Redis 6.2 提供的 GEOSEARCH 命令
因此我们需要提示其版本，修改自己的 pom.xml，内容如下
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
</exclusion>
<exclusion>
<artifactId>lettuce-core</artifactId>
<groupId>io.lettuce</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
<version>2.6.2</version>
</dependency>
<dependency>
<artifactId>lettuce-core</artifactId>
<groupId>io.lettuce</groupId>
<version>6.1.6.RELEASE</version>
</dependency>
这其中逻辑我也没消化完全，等后期我功力有成，再来详细表述。
最后效果图
八.用户签到
Redis的BitMap数据统计功能
8.1 BitMap
假如我们用一张表来存储用户签到信息，其结构应该如下
假如用户数量庞大，平均每人每年签到次数为 10 次，则这张表一年的数据量为 1 亿条
每签到一次需要使用（8 + 8 + 1 + 1 + 3 + 1）共 22 字节的内存，一个月则最多需要 600 多字节
我们按月来统计用户签到信息，签到记录为 1，未签到则记录为 0
把每一个 bit 位对应当月的每一天，形成了映射关系。用0和1标示业务状态，这样一个月也只消耗31位（4字节）这种思路就称为位图（BitMap）
Redis 中 是利用 string 类型数据结构实现 BitMap，因此最大上限是 512M，转换为 bit 则是 2^32个 bit 位。
BitMap 的操作命令有：
8.2 签到功能
因为 BitMap 底层是基于 String 数据结构，因此其操作也都封装在字符串相关操作中了
实现类代码
@Override
public Result sign() {
//获取登录用户
Long userId = UserHolder.getUser().getId();
//获取当前日期
LocalDateTime now = LocalDateTime.now();
//拼接key
String format = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
String key = "sign:"+ userId + format;
//获取今天是本月的第几天
int dayOfMonth = now.getDayOfMonth();
//写入redis select key offset 1
stringRedisTemplate.opsForValue().setBit(key, dayOfMonth -1, true);//true代表 1为签到，0为未签到
return Result.ok();
}
测试通过
8.3 签到统计
/**
* 统计签到次数
* @return
*/
@Override
public Result signCount() {
//获取登录用户
Long userId = UserHolder.getUser().getId();
//获取当前日期
LocalDateTime now = LocalDateTime.now();
//拼接key
String format = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
String key = "sign:"+ userId + format;
//获取今天是本月的第几天
int dayOfMonth = now.getDayOfMonth();
//获取本月截至今天为止的所有签到记录，返回的是一个十进制的数字 BITFIELD sign：5：202203 GET u14 0
List<Long> 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);
}
//循环遍历
int count = 0;
while (true){
//让这个数字与1做与运算，得到数字的最后一个bit位 //判读这个bit位是否为0
if ((num & 1) == 0){
//如果为0，说明未签到，结束
break;
}else {
//如果不为0，说明已签到，计数器加1
count++;
}
//把数字右移一位，抛弃最后一个bit位，继续下一个bit位
num >>>= 1;
}
return Result.ok(count);
}
九.UV统计
Redis的HyperLogLog的统计功能
-
UV：全称 Unique Visitor，也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1 天内同一个用户多次访问该网站，只记录1次。
-
PV：全称Page View，也叫页面访问量或点击量，用户每访问网站的一个页面，记录 1 次PV，用户多次打开页面，则记录多次PV。
Hyperloglog（HLL）是从 Loglog 算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。
相关算法原理大家可以参考：https://juejin.cn/post/6844903785744056333#heading-0
Redis 中的 HLL 是基于 string 结构实现的，单个 HLL 的内存永远小于 16 kb，内存占用低，但相对的其测量结果是概率性的，有小于 0.81％ 的误差。不过对于 UV 统计的庞大数量来说，这完全可以忽略。
127.0.0.1:6379> PFADD hl1 e1 e2 e3 e4 e5
(integer) 1
127.0.0.1:6379> pfcount hl1
(integer) 5
127.0.0.1:6379> PFADD hl1 e1 e2 e3 e4 e5
(integer) 0
127.0.0.1:6379> pfcount hl1
(integer) 5
由上可以看出，Hyperloglog天生就适合做UV统计，相同元素只能统计一次
通过单元测试，向 HyperLogLog 中添加 100 万条数据，看看内存占用和统计效果如何
@Test
void testHyperLogLog() {
String[] values = new String[1000];
int j = 0;
for (int i = 0; i < 1000000; i++) {
j = i % 1000;
values[j] = "user_" + i;
if (j == 999) {
// 发送到 Redis
stringRedisTemplate.opsForHyperLogLog().add("hl2", values);
}
}
// 统计数量
Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2");
System.out.println("count = " + count);
}
总结：
尚未完结
视频中虽然完结了，但是对于这个项目而言，还有一些功能未实现
- 地图功能和消息功能也没有实现，等我会前端了，就给他办了
已经5w字了，再写有点卡，后面功能完善了我再补充吧。
至此感谢您的阅览，个人能力有限，文中不足之处诸位多多包涵，也欢迎您的指正。等功能都实现了，我再来补充。
原文地址:https://blog.csdn.net/giveupgivedown/article/details/128723748
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如若转载，请注明出处：http://www.7code.cn/show_13927.html
如若内容造成侵权/违法违规/事实不符，请联系代码007邮箱：suwngjj01@126.com进行投诉反馈，一经查实，立即删除！
