点评项目核心内容

目录

拦截器设置

集群的session共享问题

 基于redis实现共享session登录

 创建bean对象技巧

什么是缓存

使用缓存来处理对象

使用String类型缓存来处理集合

缓存更新策略

主动更新策略

缓存穿透

 空串""和null的区别

缓存null值解决穿透问题

缓存雪崩

 缓存击穿

互斥锁和逻辑过期介绍 

基于互斥锁解决缓存穿透问题

​编辑

 下载JMeter模拟线程测试

redis缓存工具类封装

优惠卷秒杀

全局唯一id

优惠券添加

 优惠券秒杀下单

 JMeter线程测试遇到401错误

超卖问题分析

乐观锁

版本号法

 CAS法

一人一单的并发安全问题

分布式锁

​编辑 什么是分布式锁

 分布式锁的实现

基于Redis实现分布式锁的初级版本 

线程存在问题

线程阻塞超时自动删除后，线程完成释放别的线程的锁

 改进分布式锁（判断线程和存的是否一致）

有并发安全分析 

 Redis的Lua脚本

 基于Redis的分布式锁实现思路

 还存在的问题

Redission实现分布式锁

Redisson可重入锁原理

获取锁的Lua脚本 

 释放锁的Lua脚本

Redisson分布式锁的原理

Redis秒杀优化（暂未实现）

 达人探店

点赞功能

 实现点赞排行榜功能

关注和取关

查看共同关注

关注推送

feed流模式

Redis最佳实践

Redis键值设计

---

拦截器设置

第一步，定义拦截器

package com.hmdp.utils;import com.hmdp.dto.UserDTO;
import com.hmdp.entity.User;
import org.springframework.web.servlet.HandlerInterceptor;import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;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((UserDTO) user);//6、放行return true;}@Overridepublic void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {//移除用户，避免内存泄漏UserHolder.removeUser();}
}

userhold类下

package com.hmdp.utils;import com.hmdp.dto.UserDTO;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();}
}

第二步：配置文件（让拦截器生效）

package com.hmdp.config;import com.hmdp.utils.LoginInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;@Configuration
public class MvcConfig  implements WebMvcConfigurer {
//配置，添加拦截器，让之前的拦截器生效@Overridepublic void addInterceptors(InterceptorRegistry registry) {registry.addInterceptor(new LoginInterceptor())//排除掉不需要拦截的路径.excludePathPatterns("/shop/**","/voucher/**","/shop-type/**","/upload/**","/blog/hot","/usr/code","/usr/login");}
}

集群的session共享问题

session共享问题：多态Tomcat并不共享session存储空间，当请求切换到不同tomcat服务是导致数据丢失问题

session的代替为redis，满足数据共享，内存存储key、value结构

 基于redis实现共享session登录

保存登录的用户信息，可以使用String结构，以JSON字符串来保存，比较直观：

 Hash结构可以将对象中的每个字段独立存储，可以针对单个字段做CRUD，并且内存占用更少:

 

 

 创建bean对象技巧

自己创建的类，无法直接用@Autowire方式注入，因为他不属于 spring容器管理的。

需要在创建的加入一个构造方法，然后在由其他由spring管理的类调用，然后在注入传入这个属性即可

@Configuration
public class MvcConfig  implements WebMvcConfigurer {
//由spring管理的类注册@Resourceprivate StringRedisTemplate stringRedisTemplate;//配置，添加拦截器，让之前的拦截器生效@Overridepublic void addInterceptors(InterceptorRegistry registry) {//调用时传入这个即可registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))}

public class LoginInterceptor implements HandlerInterceptor {private StringRedisTemplate stringRedisTemplate;public LoginInterceptor(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}
}

什么是缓存

缓存就是数据交换的缓冲区（称作Cache，是存储数据的临时笛梵，一般读写性能较高）

使用缓存来处理对象

 //1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串String shopJson = stringRedisTemplate.opsForValue().get(key);//2、判断是否存在if (StrUtil.isNotBlank(shopJson)){//3、存在，直接返回缓存中的//是json数据，则需要通过JSONUtil返回指定的对象即可Shop shop= JSONUtil.toBean(shopJson, Shop.class);return  Result.ok(shop);}//4、缓存中不存在，根据id查询数据库Shop shop = this.getById(id);//5、数据库中不存在，返回错误if (shop==null){return Result.fail("商铺不存在");}//6、存在，写入缓存stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));//7、返回

使用String类型缓存来处理集合

@Resource
private StringRedisTemplate stringRedisTemplate;@Overridepublic Result shopTypeList() {String key="shop_Type_List";//查询缓存String shopTypeJson = stringRedisTemplate.opsForValue().get(key);if (StrUtil.isNotBlank(shopTypeJson)){//查到了，直接返回，json转list集合List<ShopType> shopTypes = JSONUtil.toList(shopTypeJson, ShopType.class);return Result.ok(shopTypes);}//缓存没查到，查数据库List<ShopType> typeList = this.query().orderByAsc("sort").list();//数据库没查到，返回错误if (typeList==null){return Result.fail("没有列表信息");}//数据库查到，缓存下//list集合转jsonString json = JSONUtil.toJsonStr(typeList);stringRedisTemplate.opsForValue().set(key, json);

缓存更新策略

 业务场景

低一致性需求：使用内存淘汰机制，例如店铺类型的查询

高一致性需求；主动更新，并以超时剔除作为兜底方案，例如店铺的详细信息

主动更新策略

操作缓存和数据库的三个问题需要考虑：

1，删除缓存还是更新缓存？

• 更新缓存：每次更新数据都会更新缓存，无效写操作较多（×）
• 删除缓存：更新数据库时让缓存失效，查询时在更新缓存（√）

2、如何保证缓存与数据库的操作同时成功或者失败？

• 单体系统，将缓存与数据库操作放在一个事务
• 分布式系统，利用TCC操作等分布式事务方案

3、先操作缓存还是先操作数据库

• 先操作数据库，在删除缓存

 缓存更新策略的最佳实践方案：

1、低一致性需求：使用Redis自带的内存淘汰机制

2、高一致性需求：主动更新，并以超时提出作为兜底方案

读操作：

• 缓存命中则直接返回
• 缓存未命中则查询数据库，并写入缓存，设置超时时间

写操作::

• 先写数据库，然后在删除缓存
• 要确保数据库与缓存操作的原子性

缓存穿透

我们使用Redis大部分情况都是通过Key查询对应的值，假如发送的请求传进来的key是不存在Redis中的，那么就查不到缓存，查不到缓存就会去数据库查询。假如有大量这样的请求，这些请求像“穿透”了缓存一样直接打在数据库上，这种现象就叫做缓存穿透。

解决方案：

1、缓存空对象（把无效的Key存进Redis中）。如果Redis查不到数据，数据库也查不到，我们把这个Key值保存进Redis，设置value="null"，当下次再通过这个Key查询时就不需要再查询数据库。这种处理方式肯定是有问题的，假如传进来的这个不存在的Key值每次都是随机的，那存进Redis也没有意义，会占用redis的内存空间，所以设置过期时间是有必要的，其次，当这个值一开始没有内容，我们查询数据库后 ，将null赋值给这个值，并存在redis中，而之后我们数据库新增了这个值，但是缓存中还是为null，这就会导致短期数据不一致，可以使用更新数据库删除那个缓存就可以解决。

2、使用布隆过滤器。布隆过滤器的作用是某个 key 不存在，那么就一定不存在，它说某个 key 存在，那么很大可能是存在(存在一定的误判率)。于是我们可以在缓存之前再加一层布隆过滤器，在查询的时候先去布隆过滤器查询 key 是否存在，如果不存在就直接返回，这个布隆过滤器也存在一定的穿透风险。

 项目中解决缓存穿透的思路

 空串""和null的区别

null表示的是一个对象的值，而非一个字符串。例如声明一个对象的引用，String aaa = null ;
""表示的是一个长度为0的空字符串。例如声明一个字符串String bbb = "" ;
所以：null不指向任何对象，相当于没有任何值;而""代表一个长度为0的字符串。

缓存null值解决穿透问题

 public Result queryById(Long id) {String key=CACHE_SHOP_KEY+id;//1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串String shopJson = stringRedisTemplate.opsForValue().get(key);//2、判断是否存在if (StrUtil.isNotBlank(shopJson)){//3、存在，直接返回缓存中的//是json数据，则需要通过JSONUtil返回指定的对象即可Shop shop= JSONUtil.toBean(shopJson, Shop.class);return  Result.ok(shop);}//4、缓存中不存在，判断是否命中的为空串""if (shopJson!=null){return Result.fail("商铺不存在");}Shop shop = this.getById(id);//5、数据库中不存在，并返回错误if (shop==null){//插入一个空串,设置过期时间stringRedisTemplate.opsForValue().set(key,"", CACHE_NULL_TTL, TimeUnit.MINUTES);return Result.fail("商铺不存在");}//6、存在，写入缓存stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);//7、返回return  Result.ok(shop);}

缓存穿透产生的原因是什么?

用户请求的数据在缓存中和数据库中都不存在，不断发起这样的请求给数据库带来巨大压力

缓存穿透的解决方案有?

• 缓存null值
• 布隆过滤
• 增强id的复杂度，避免被猜测id规律
• 做好数据的基础格式校验
• 加强用户权限校验
• 做好热点参数的限流

缓存雪崩

当某一个时刻出现大规模的redis缓存失效的情况，就会导致大量的请求直接打在数据库上面，导致数据库压力巨大，如果在高并发的情况下，可能瞬间就会导致数据库宕机。这时候如果运维马上又重启数据库，马上又会有新的流量把数据库打死。这就是缓存雪崩。缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

解决方案:

• 给不同的Key的TTL添加随机值
• 利用Redis集群提高服务的可用性
• 给缓存业务添加降级限流策略
• 给业务添加多级缓存

1、在原有的失效时间上加上一个随机值，这样就避免了因为采用相同的过期时间导致的缓存雪崩。

如果真的发生了缓存雪崩，有没有什么兜底的措施？

2、使用熔断机制。当流量到达一定的阈值时，就直接返回“系统拥挤”之类的提示，防止过多的请求打在数据库上。至少能保证一部分用户是可以正常使用，其他用户多刷新几次也能得到结果。

3、提高数据库的容灾能力，可以使用分库分表，读写分离的策略。

4、为了防止Redis宕机导致缓存雪崩的问题，可以搭建Redis集群，提高Redis的容灾性

 缓存击穿

其实跟缓存雪崩有点类似，缓存雪崩是大规模的key失效，而缓存击穿是一个热点的Key，有大并发集中对其进行访问，突然间这个Key失效了，导致大并发全部打在数据库上，导致数据库压力剧增。这种现象就叫做缓存击穿。

解决方案：

• 互斥锁
• 逻辑过期（不过期）

1、业务允许的话，对于热点的key可以设置永不过期的key。

2、使用互斥锁。如果缓存失效的情况，只有拿到锁才可以查询数据库，降低了在同一时刻打在数据库上的请求，防止数据库打死。当然这样会导致系统的性能变差。

多条线程同时访问数据库 

互斥锁和逻辑过期介绍 

 

基于互斥锁解决缓存穿透问题

修改id查询店铺 ，基于互斥锁来解决缓存击穿问题

    /*** 互斥锁解决缓存击穿* @param id* @return*/public Shop queryWithMutex(Long id){String key=CACHE_SHOP_KEY+id;//1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串String shopJson = stringRedisTemplate.opsForValue().get(key);//2、判断是否存在if (StrUtil.isNotBlank(shopJson)){//3、存在，直接返回缓存中的//是json数据，则需要通过JSONUtil返回指定的对象即可Shop shop= JSONUtil.toBean(shopJson, Shop.class);return  shop;}//4、缓存中不存在，判断是否命中的为空串""if (shopJson!=null){return null;}//4实现缓存重建//4.1获取互斥锁String lockKey=LOCK_SHOP_KEY+id;Shop shop = null;try {boolean isLock = tryLock(lockKey);//4.2判断是否获取成功if (!isLock){//4.3失败，则休眠并重试Thread.sleep(99);return queryWithMutex(id);}//4.4成功，根据id查询数据库shop = this.getById(id);//模拟重建延时Thread.sleep(366);//5、数据库中不存在，并返回错误if (shop==null){//插入一个空串,设置过期时间stringRedisTemplate.opsForValue().set(key,"", CACHE_NULL_TTL, TimeUnit.MINUTES);return null;}//6、存在，写入缓存stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {//7、释放互斥锁unLock(lockKey);}//8、返回return shop;}

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

  /*** 逻辑辑过期解决缓存击穿* @param id* @return*///创建一个线程池private static final ExecutorService CACHE_REBUILD_EXECUTOR= Executors.newFixedThreadPool(10);public Shop queryWithLogicalExpire(Long id){String key=CACHE_SHOP_KEY+id;//1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串String shopJson = stringRedisTemplate.opsForValue().get(key);//2、判断是否存在if (StrUtil.isBlank(shopJson)){//3不存在，直接返回nullreturn  null;}//4命中，需要把json反序列化为对象RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);JSONObject data = (JSONObject) redisData.getData();Shop shop = JSONUtil.toBean(data, 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){//6.3成功，开启独立线程，实现缓存重建CACHE_REBUILD_EXECUTOR.submit(()->{try {//重建缓存this.saveShop2Redis(id,20L);} catch (Exception e) {e.printStackTrace();} finally {//释放锁unLock(lockKey);}});}return shop;}

 下载JMeter模拟线程测试

 修改数据据库中的一点信息，会发现某一时刻重建前是旧数据，完成后是新数据

 

 控制台中只有一数据重建，一次是查询旧数据，一次为新数据重建

redis缓存工具类封装

基于stringRedisTemplate封装一个缓存工具类，满足下列需求:

方法1:将任意va对象序列化为json并存储在string类型的key中，并且可以设置TTL过期时间

方法2:将任意/ava对象序列化为json并存储在string类型的key中，并且可以设置逻辑过期时间，用于处理缓存击穿问题

方法3:根据指定的key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题

方法4:根据指定的key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题

package com.hmdp.utils;import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;import static com.hmdp.utils.RedisConstants.LOCK_SHOP_KEY;@Slf4j
@Component/*** 缓存工具类1*/
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)));//写入RedisstringRedisTemplate.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){//将空值写入redisstringRedisTemplate.opsForValue().set(key,"",2,TimeUnit.MINUTES);//返回错误信息return null;}//6.存在写入redisthis.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.存在，缓存中存的nullreturn 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);//写入redisthis.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);}
}

优惠卷秒杀

全局唯一id

全局id生成器

全局ID生成器，是一种在分布式系统下用来生成全局唯一id的工具，一般满足：

• 唯一性
• 高可用
• 高性能
• 递增性
• 安全性

为了增加ID的安全性，我们可以不直接使用Redis自增的数值，而是拼接一些其它信息：

 ID的组成部分:
符号位:1bit，永远为0
时间戳:31bit，以秒为单位，可以使用69年
序列号:32bit，秒内的计数器，支持每秒产生2^32个不同ID

@Component
public class RedisWorker {private static final long BEGIN_TIMESTAMP=1640995000L;private static final long COUNT_BITS=32;@Resourceprivate 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);//拼接并返回(先向左移把右边空出，然后在或，相当于加上)return timestamp<<COUNT_BITS|count;}}

 测试类中

    private ExecutorService es= Executors.newFixedThreadPool(500);@Testvoid testIdWordker() throws InterruptedException {CountDownLatch latch = new CountDownLatch(300);Runnable task=()->{for (int i=0;i<100;i++){long id = redisWorker.nextId("order");System.out.println("id="+id);}latch.countDown();};long begin = System.currentTimeMillis();for (int i = 0; i <100 ; i++) {es.submit(task);}latch.await();long end = System.currentTimeMillis();System.out.println("time="+(end-begin));}

全局唯一id生成策略

• UUID
• redis自增
• snowflake算法
• 数据库自增 

redis自增id策略

• 每天一个key，方便统计订单量
• id构造是 时间戳+计数器

优惠券添加

没有后台只能通过postman添加

{
"shopId": 1,
"title": "200元代金券",
"subTitle": "周六周末可用",
"rules": "全场通用\\n可以叠加\\n仅限制堂食",
"payValue": 16000,
"actualValue": 20000,
"type": 1,
"stock": 100,
"beginTime": "2023-08-01T00:00:00",
"endTime": "2024-08-01T00:00:00"
}

 

 优惠券秒杀下单

 下单时需要判断两点:
秒杀是否开始或结束，如果尚未开始或已经结束则无法下单

库存是否充足，不足则无法下单

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resourceprivate ISeckillVoucherService seckillVoucherService;
@Resource
private RedisWorker redisWorker;/*** 优惠券下单* @param voucherId* @return*/@Override@Transactionalpublic 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 = redisWorker.nextId("order");voucherOrder.setId(orderId);//6.2用户idLong userId = UserHolder.getUser().getId();voucherOrder.setUserId(userId);//6.3代金券idvoucherOrder.setVoucherId(voucherId);//添加订单save(voucherOrder);//7返回订单idreturn Result.ok(orderId);}
}

 

 JMeter线程测试遇到401错误

这是未授权问题
用 F12 打开开发者工具
在network网络 里寻找相关信息:

添加一个信息管理器

 订单出现了超卖

超卖问题分析

超卖问题是多线程安全问题，即在一个线程还没执行完，其他线程抢先执行，对同一个数据进行修改

乐观锁

 乐观锁的关键是判断之前查询到的数据是否被修改，常见的方式有

版本号法

在修改之前查询一次版本号，若版本号不变则说明没有被其他线程修改，则正常进行数据修改，并让版本加一，若是版本号不一致，则不会执行

 

 CAS法

直接比较数据是否发生了改变，若不变则说明安全

如果弄数据是否和之前一致，会导致成功率低

设置200个线程只卖出23个 

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resourceprivate ISeckillVoucherService seckillVoucherService;
@Resource
private RedisWorker redisWorker;/*** 优惠券下单* @param voucherId* @return*/@Override@Transactionalpublic 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") //set stock=stock-1.eq("voucher_id", voucherId).gt("stock", 0)//where id=? and stock=?只要大于0即可.update();if (!success)return Result.fail("库存不足");//6创建秒杀券订单VoucherOrder voucherOrder = new VoucherOrder();//6.1订单idlong orderId = redisWorker.nextId("order");voucherOrder.setId(orderId);//6.2用户idLong userId = UserHolder.getUser().getId();voucherOrder.setUserId(userId);//6.3代金券idvoucherOrder.setVoucherId(voucherId);//添加订单save(voucherOrder);//7返回订单idreturn Result.ok(orderId);}
}

 超卖这样的线程安全问题，解决方案有哪些?

1.悲观锁:添加同步锁，让线程串行执行

• 优点:简单粗暴
• 缺点:性能一般

乐观锁:不加锁，在更新时判断是否有其它线程在修改

• 优点:性能好
• 缺点:存在成功率低的问题

给整个this对象上锁

优点是简单安全

缺点是性能低，因为这样所有用户都被锁上了，我们的初衷是，单个用户中，只能单卖，这样就会导致其他用户也会受到影响

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resourceprivate ISeckillVoucherService seckillVoucherService;
@Resource
private RedisWorker redisWorker;/*** 优惠券下单* @param voucherId* @return*/@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("已经被抢完");}//7返回订单idreturn createVoucherOrder(voucherId);}@Transactionalpublic  synchronized Result createVoucherOrder(Long voucherId){//5、一人一单Long userId = UserHolder.getUser().getId();//5.1查询订单Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();//判断是否已经存在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("库存不足");//6创建秒杀券订单VoucherOrder voucherOrder = new VoucherOrder();//6.1订单idlong orderId = redisWorker.nextId("order");voucherOrder.setId(orderId);//6.2用户idvoucherOrder.setUserId(userId);//6.3代金券idvoucherOrder.setVoucherId(voucherId);//添加订单save(voucherOrder);//返回订单idreturn Result.ok(orderId);}
}

悲观锁升级后 

@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();synchronized (userId.toString().intern()){//intern获取源对象，new多少次都只是从常量池中寻找//拿到当前对象的代理对象(事务的对象)IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);//要想让事务生效，必须要有代理对象}
}@Transactionalpublic   Result createVoucherOrder(Long voucherId){//5、一人一单Long userId = UserHolder.getUser().getId();//5.1查询订单Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();//判断是否已经存在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("库存不足");//6创建秒杀券订单VoucherOrder voucherOrder = new VoucherOrder();//6.1订单idlong orderId = redisWorker.nextId("order");voucherOrder.setId(orderId);//6.2用户idvoucherOrder.setUserId(userId);//6.3代金券idvoucherOrder.setVoucherId(voucherId);//添加订单save(voucherOrder);//返回订单idreturn Result.ok(orderId);}

一人一单的并发安全问题

通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。
1.我们将服务启动两份，端口分别为8081和8082:

2.然后修改nginx的conf目录下的nginx.conf文件，配置反向代理和负载均衡

nginx配置实现了反代理和负载均衡 

 

 发现线程不安全

集群下的锁监听器tomcat等不是同一个

分布式锁

 什么是分布式锁

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁

 分布式锁的实现

分布式锁的核心是是实现多进程之间互斥，常见的有三种 

 实现分布式锁时需要实现两个基本方法

获取锁：

• 互斥:确保只能有一个线程获取锁
• 非阻塞：尝试一次，成功返回true，失败返回false

添加锁，nx是互斥（当存在则不执行），ex是设置超时时间

local是键 thread1是值

• 

释放锁：

• 手动释放
• 超时释放：获取锁时添加一个超时时间

释放锁，直接删除即可

 业务流程

基于Redis实现分布式锁的初级版本 

锁的类 

public class SimpleRedisLock  implements ILock{private StringRedisTemplate stringRedisTemplate;private String name;public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {this.stringRedisTemplate = stringRedisTemplate;this.name = name;}private static final String KEY_PREFIX="lock:";//获取锁@Overridepublic boolean tryLock(long timeoutSec) {//获取线程标识long threadId = Thread.currentThread().getId();//获取锁，这里的返回值是一个布尔型的包装类，直接返回有时会出现空指针异常Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);//和真比较，就不会出现异常return Boolean.TRUE.equals(success) ;}//释放锁@Overridepublic void unlock() {stringRedisTemplate.delete(KEY_PREFIX+name);}
}

    Long userId = UserHolder.getUser().getId();//创建锁对象SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);//获取锁boolean isLock = lock.tryLock(12);//判断是否获取锁成功if (!isLock){//获取锁失败return Result.fail("一个人只能下一单");}try {//拿到当前对象的代理对象(事务的对象)IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);//要想让事务生效，必须要有代理对象}  finally {//释放锁lock.unlock();}

线程存在问题

线程阻塞超时自动删除后，线程完成释放别的线程的锁

存在的线程阻塞超时自动删除后，线程释放别的线程的锁

 改进分布式锁（判断线程和存的是否一致）

需求:修改之前的分布式锁实现，满足:

1.在获取锁时存入线程标示(可以用UUID表示)

2.在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致如果一致则释放锁
如果不一致则不释放锁

  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);}

有并发安全分析 

 这里有个并发问题，即当判断完相同时，发生了阻塞，没来得及删除锁，被redis超时释放后，下一个线程来获取后，之前那个线程阻塞完成，就会释放掉锁，但是此时这把锁的拥有者不是他。

所以我们改进的是时候，应该保证，判断和删除在同一条语句中，即使用lua脚本可以保证原子性

 Redis的Lua脚本

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站: https://www.runoob.com/lua/lua-tutorial.html语法如下:

 如，我们要执行set name jack则脚本是这样

 列如。我们要先执行set name Rose，在执行get name ，则脚本如下

 需要用Redis命令来调用脚本，调用脚本的常见命令如下:

例如，我们要执行 redis.call('set','name','jack') 这个脚本，语法如下: 

脚本中可以从KEYS和ARGV数组获取这些参数:
如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在

 resoure下创建unlock.lua

--比较线程标识与锁中是否一致
if (redis.call('get',KEYS[1])==ARGV[1]) then-- 释放锁 del keyreturn redis.call('del',KEYS[1])
end
return 0

    private static final String KEY_PREFIX = "lock:";private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}//获取锁@Overridepublic 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);}//释放锁@Overridepublic void unlock() {//调用LiastringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX+name),ID_PREFIX+Thread.currentThread().getId());}

 基于Redis的分布式锁实现思路

• 利用set nxex获取锁，并设置过期时间，保存线程标示
• 释放锁时先判断线程标示是否与自己一致，一致则删除锁

特性:

• 利用setnx满足互斥性
• 利用set ex保证故障时锁依然能释放，避免死锁，提高安全性
• 利用Redis集群保证高可用和高并发特性

 还存在的问题

基于setnx实现的分布式锁还存在下面的问题

不可重入：同一个线程无法多次获取同一把锁

不可重试：获取锁只尝试一次就返回false，没有重试机制

超时释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患

主从一致性：如果Redis提供了主从集群主从同步存在延迟，当主宕机时，如果从并同步主中的
锁数据，则会出现锁实现

Redission实现分布式锁

Redisson是一个在Redis的基础上实现的ava驻内存数据网格 (In-Memory Data Grid)。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

官网地址: https://redisson.org
GitHub地址: https://github.com/redisson/redisso 

Redisson入门

导入maven地址

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

创建配置文件

package com.hmdp.config;import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient(){//配置Config config=new Config();config.useSingleServer().setAddress("redis://localhost:6379");//创建RedissonClient对象return Redisson.create(config);
}
}

       //创建锁对象
//        SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);RLock lock = redissonClient.getLock("lock:order:" + userId);//获取锁，数量分别是：获取锁的最大等待时间（期间会重试），锁自动释放时间，时间单位boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);if(isLock){try {System.out.println("执行业务");}finally {// 释放锁lock.unlock();}

Redisson可重入锁原理

获取锁的Lua脚本 

 释放锁的Lua脚本

Redisson分布式锁的原理

 Redisson分布式锁原理：

• 可重入：利用hash结构记录线程id和重入次数
• 可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
• 超时续约：用于watchDog.每隔一段时间（releaseTime/3），重置超时时间

总结 

1)不可重入Redis分布式锁
原理:利用setnx的互斥性;利用ex避免死锁;释放锁时判断线程标示
缺陷:不可重入、无法重试、锁超时失效
2)可重入的Redis分布式锁:
原理:利用hash结构，记录线程标示和重入次数;利用watchDog延续锁时间;利用信号量控制锁重试等待
缺陷: redis宕机引起锁失效问题
3)Redisson的multiLock:
原理:多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功
缺陷:运维成本高、实现复杂

Redis秒杀优化（暂未实现）

 

 达人探店

点赞功能

/*** 博客点赞** @param id* @return*/@Overridepublic Result likeBlog(Long id) {//获取blog实体Blog blog = getById(id);//获取博客idLong blogId = blog.getId();//获取登录当前用户idLong userId = UserHolder.getUser().getId();//拼接keyString key = BLOG_LIKED_KEY + blogId;//去redis中看有没有点赞过，查询这个key是否存在Boolean isMember = stringRedisTemplate.opsForSet().isMember(key, userId.toString());//不存在，即没点赞过if (BooleanUtil.isFalse(isMember)) {//修改数据库，让liked加1boolean isSuccess = update().setSql("liked=liked+1").eq("id", id).update();if (isSuccess) {//修改成功后，将自定义的blog的isLike(是否点赞过)改为true// blog.setIsLike(true);不能，因为这不是数据库中的字段存不了//将这个点赞信息加到redis缓存中stringRedisTemplate.opsForSet().add(key, userId.toString());}} else {//存在，即点赞过,则取消赞boolean isSuccess = update().setSql("liked=liked-1").eq("id", id).update();if (isSuccess) {//修改，删除stringRedisTemplate.opsForSet().remove(key, userId.toString());}}return Result.ok();}

 实现点赞排行榜功能

  /*** 博客点赞排序* @param id* @return*/@Overridepublic Result queryBlogLikes(long id) {String key =BLOG_LIKED_KEY+id;//1查询top5的点赞用户 zrange key 0 4Set<String> topRange = stringRedisTemplate.opsForZSet().range(key, 0, 4);if (topRange==null||topRange.isEmpty()){return Result.ok(Collections.emptyList());}//2解析出其中的用户idList<Long> ids = topRange.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)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());return Result.ok(userDTOS);}

关注和取关

 @Overridepublic Result follow(Long followUserId, boolean isFollow) {//1获取当前登录用户Long userId = UserHolder.getUser().getId();if (userId==null)return Result.fail("还没有登录");Follow follow = new Follow();//2判断为关注还是取关if (isFollow){//为关注，新增follow数据follow.setUserId(userId);follow.setFollowUserId(followUserId);save(follow);}else {//为取关，删除follow数据LambdaUpdateWrapper<Follow> updateWrapper=new LambdaUpdateWrapper();updateWrapper.eq(Follow::getUserId, userId).eq(Follow::getFollowUserId,followUserId);//删除数据this.remove(updateWrapper);}return Result.ok();}@Overridepublic Result isFollow(Long followUserId) {//1获取当前登录用户Long userId = UserHolder.getUser().getId();if (userId==null)return Result.fail("还没有登录");Integer count = query().eq("user_id", userId).eq("follow_user_id", followUserId).count();//判断是否有数据return Result.ok(count>0);}

查看共同关注

/*** 被查看的人和我的共同关注** @param checkedUserId* @return*/@Overridepublic Result commonConcernPerson(Long checkedUserId) {//1获取当前登录用户Long userId = UserHolder.getUser().getId();if (userId == null) return Result.fail("还没有登录");String key = "follows:" + userId;//2求交集String key2 = "follows:" + checkedUserId;Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key, key2);if (intersect==null||intersect.isEmpty()){//没有交集return Result.ok(Collections.emptyList());}//解析出id集合List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());//4、根据id查询用户，转为userDtoList<UserDTO> userDTOS = userService.listByIds(ids).stream().map(user -> BeanUtil.copyProperties(user, UserDTO.class )).collect(Collectors.toList());return Result.ok(userDTOS);}

关注推送

关注推送也叫做Feed流，直译为投喂。为用户持续的提供“沉浸式”的体验，通过无限下拉刷新获取新的信息。

feed流模式

Feed流产品有两种常见模式:
Timeline:不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈优点:信息全面，不会有缺失。并且实现也相对简单
缺点:信息噪音较多，用户不一定感兴趣，内容获取效率低

智能排序:利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户> 优点: 投喂用户感兴趣信息，用户粘度很高，容易沉迷
缺点:如果算法不精准，可能起到反作用

 

对于普通人：直接发送到他的粉丝下 

对于大v：活跃粉丝直接给他推送，不活跃粉丝放在收件箱，等他要读的时候推

feed流实现方案

 需求

 修改新增探店笔记的业务，在保存blog到数据库的同时，推送到粉丝的收件箱

收件箱满足可以根据时间戳排序，必须用Redis的数据结构实现

查询收件箱数据时，可以实现分页查询

不能使用传统的分页，得使用滚动分页

    
@GetMapping("/of/follow")
public Result queryBlogOfFollow(
@RequestParam("lastId") Long max,@RequestParam(value = "offset",defaultValue = "0") Integer offset){return blogService.queryBlogOfFollow(max,offset);
}

    @Overridepublic Result queryBlogOfFollow(Long max, Integer offset) {//1获取当前用户Long userId = UserHolder.getUser().getId();//2查询收件箱 ZREVRANGEBYSCORE key Max Min LIMiT offset countString key =FEED_KEY+userId;Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, 0, max, offset, 2);//3判断是否为空if (typedTuples==null||typedTuples.isEmpty()){return Result.ok();}//4解析数据：blogId，minTime(时间戳)，offsetList<Long> ids=new ArrayList<>(typedTuples.size());long minTime=0;int offNum=1;//统计有多次offNum（和最小的相同的个数）for (ZSetOperations.TypedTuple<String> tuple : typedTuples) {//获取idids.add(Long.valueOf(tuple.getValue()));//获取分数（时间戳）long time = tuple.getScore().longValue();if (time==minTime){offNum++;}else {minTime=time;offNum=1;}}//5,根据id查询blogString idStr =StrUtil.join(",", ids);List<Blog> blogs = query().in("id", ids).last("order by field (id," + idStr + ")").list();//把博客相关信息点赞填充for (Blog blog : blogs) {//查询blog有关用户queryBlogUser(blog);isLikeBlog(blog);}//5封装并返回ScrollResult scrollResult = new ScrollResult();scrollResult.setList(blogs);scrollResult.setOffset(offNum);scrollResult.setMinTime(minTime);return Result.ok(scrollResult);}

Redis最佳实践

Redis键值设计

优雅的key结构

Redis的key虽然可以自定义，但最好遵循下面结构最佳实践的约定：

• 遵循基本格式：[业务名称]:[数据名]:[id]
• 长度不超过44字节
• 不包含特殊字符

例如：在登录业务，保存用户信息，key是这样：login:user:1

优点：

• 可读强
• 避免key冲突
• 方便管理

更节省内存：key是string类型，底层编码包含int、embstr、和raw三种，embstr在小于44字节使用，采用连续的空间，内存占用更小