黑马点评_知识点

将手机验证码保存到HttpSession中进行验证（感觉已经过时）

Controller中的参数有HttpSession，存验证码session.setAttribute(SystemConstants.VERIFY_CODE, code);
其他的都是逻辑代码

Cookie的缺点

什么是Session集群共享问题？

在分布式集群环境中，会话（Session）共享是一个常见的挑战。默认情况下，Web 应用程序的会话是保存在单个服务器上的，当请求不经过该服务器时，会话信息无法被访问。

Session集群共享问题造成哪些问题？

服务器之间无法实现会话状态的共享。比如：在当前这个服务器上用户已经完成了登录，Session中存储了用户的信息，能够判断用户已登录，但是在另一个服务器的Session中没有用户信息，无法调用显示没有登录的服务器上的服务。

如何解决Session集群共享问题？

方案一：Session拷贝（不推荐）

Tomcat提供了Session拷贝功能，通过配置Tomcat可以实现Session的拷贝，但是这会增加服务器的额外内存开销，同时会带来数据一致性问题

方案二：Redis缓存（推荐）

Redis缓存具有Session存储一样的特点，基于内存、存储结构可以是key-value结构、数据共享

Spring Boot中使用Redis

注意配置完之后就可以直接将RedisTemplate拿来用了

StringRedisTemplate也是直接注入使用

验证码使用Redis

随机一个验证码，用手机号做key，value为验证码存到redis中；设置一个过期时间；

public Result sendCode(String phone, HttpSession session) {// 1、判断手机号是否合法if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号格式不正确");}// 2、手机号合法，生成验证码，并保存到Redis中String code = RandomUtil.randomNumbers(6);stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY + phone, code,RedisConstants.LOGIN_CODE_TTL, TimeUnit.MINUTES);// 3、发送验证码log.info("验证码:{}", code);return Result.ok();}

登录使用Redis

token的产生

从redis中取出以电话号码作为key的验证码和用户输入的验证码进行对比；
然后使用UUID随机出一个token当作key，value为用户类的信息，保存到redis中
用户调用接口的时候带token

public Result login(LoginFormDTO loginForm, HttpSession session) {String phone = loginForm.getPhone();String code = loginForm.getCode();// 1、判断手机号是否合法if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号格式不正确");}// 2、判断验证码是否正确String redisCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);if (code == null || !code.equals(redisCode)) {return Result.fail("验证码不正确");}// 3、判断手机号是否是已存在的用户User user = this.getOne(new LambdaQueryWrapper<User>().eq(User::getPhone, phone));if (Objects.isNull(user)) {// 用户不存在，需要注册user = createUserWithPhone(phone);}// 4、保存用户信息到Redis中，便于后面逻辑的判断（比如登录判断、随时取用户信息，减少对数据库的查询）UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);// 将对象中字段全部转成string类型，StringRedisTemplate只能存字符串类型的数据Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),CopyOptions.create().setIgnoreNullValue(true).setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));String token = UUID.randomUUID().toString(true);String tokenKey = LOGIN_USER_KEY + token;stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);return Result.ok(token);}

拦截器中检查token有没有过期，延长token使用时间

定义拦截器（拦截器类不用注入IOC容器中），在webMVC配置类（注入IOC容器中）中会new一个拦截器
创建两个拦截器，Refresh拦截器来拦截所有接口，redis根据token能获得用户，将值保存到ThreadLocal中，刷新token有效期，放行。
第二个登录拦截器，放行登录和游客可以获得信息的接口，从ThreadLocal中查询用户，不存在就拦截，存在就放行；

刷新token拦截器

public class RefreshTokenInterceptor implements HandlerInterceptor {// new出来的对象是无法直接注入IOC容器的（LoginInterceptor是直接new出来的）// 所以这里需要再配置类中注入，然后通过构造器传入到当前类中private StringRedisTemplate stringRedisTemplate;public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {// 1、获取token，并判断token是否存在String token = request.getHeader("authorization");if (StrUtil.isBlank(token)){// token不存在，说明当前用户未登录，不需要刷新直接放行return true;}// 2、判断用户是否存在String tokenKey = LOGIN_USER_KEY + token;Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(tokenKey);if (userMap.isEmpty()){// 用户不存在，说明当前用户未登录，不需要刷新直接放行return true;}// 3、用户存在，则将用户信息保存到ThreadLocal中，方便后续逻辑处理，比如：方便获取和使用用户信息，Redis获取用户信息是具有侵入性的UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);ThreadLocalUtls.saveUser(BeanUtil.copyProperties(userMap, UserDTO.class));// 4、刷新token有效期stringRedisTemplate.expire(token, LOGIN_USER_TTL, TimeUnit.MINUTES);return true;}
}

登录拦截器

public class LoginInterceptor implements HandlerInterceptor {/*** 前置拦截器，用于判断用户是否登录*/@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {// 判断当前用户是否已登录if (ThreadLocalUtls.getUser() == null){// 当前用户未登录，直接拦截response.setStatus(HttpStatus.HTTP_UNAUTHORIZED);return false;}// 用户存在，直接放行return true;}
}

配置拦截器

优先级默认都是0，值越大优先级越低；同等优先级按照添加顺序进行拦截；

@Configuration
public class WebMvcConfig implements WebMvcConfigurer {// new出来的对象是无法直接注入IOC容器的（LoginInterceptor是直接new出来的）// 所以这里需要再配置类中注入，然后通过构造器传入到当前类中@Resourceprivate StringRedisTemplate stringRedisTemplate;@Overridepublic void addInterceptors(InterceptorRegistry registry) {// 添加登录拦截器registry.addInterceptor(new LoginInterceptor())// 设置放行请求.excludePathPatterns("/user/code","/user/login","/blog/hot","/shop/**","/shop-type/**","/upload/**","/voucher/**").order(1); // 优先级默认都是0，值越大优先级越低// 添加刷新token的拦截器registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);}
}

为商店详情添加缓存

缓存更新策略

• 内存淘汰（全自动）。利用Redis的内存淘汰机制实现缓存更新，Redis的内存淘汰机制是当Redis发现内存不足时，会根据一定的策略自动淘汰部分数据
  • Redis中常见的淘汰策略：
    • noeviction（默认）：当达到内存限制并且客户端尝试执行写入操作时，Redis 会返回错误信息，拒绝新数据的写入，保证数据完整性和一致性
    • allkeys-lru：从所有的键中选择最近最少使用（Least Recently Used，LRU）的数据进行淘汰。即优先淘汰最长时间未被访问的数据
    • allkeys-random：从所有的键中随机选择数据进行淘汰
    • volatile-lru：从设置了过期时间的键中选择最近最少使用的数据进行淘汰
    • volatile-random：从设置了过期时间的键中随机选择数据进行淘汰
    • volatile-ttl：从设置了过期时间的键中选择剩余生存时间（Time To Live，TTL）最短的数据进行淘汰
• 超时剔除（半自动）。手动给缓存数据添加TTL，到期后Redis自动删除缓存
• 主动更新（手动）。手动编码实现缓存更新，在修改数据库的同时更新缓存

我们可以同时将超时剔除和主动更新一起使用。

主动更新的策略

• 缓存调用者自行进行更新数据库的同时更新缓存
• 将缓存与数据库整合为一个服务，调用者只调用这个服务，不需要关心一致性的东西
• 只操作缓存中的数据，异步线程将缓存保存到本地

使用Cache Aside Pattern

• 为什么更新缓存中无效写操作较多，因为每次更新数据库之后都更新缓存，如果期间没有进行读那么无效写缓存的操作就会很多，同时出现线程安全问题；但是我是感觉有问题的，因为更改操作不都是先进行读然后再该么。视频中采用更新数据库的时候删除缓存，查询的时候再更新缓存的策略，有较少的线程安全问题。
• 将缓存和数据库操作放到一个事务中，先前我还以为只有数据库才能使用事务；同时分布式系统TCC是SpringCloud中的内容。
• 先删除缓存还是先操作数据库？举得例子为有两个线程，两个线程异步执行发生错误；视频中说使用先删除数据库再删除缓存的操作发生错误的概率非常小，同时设置缓存的有效时间；只是粗略的按照访问数据库时间比缓存时间多来解释，这是不太合理的。
  

最佳方案

缓存穿透

缓存空对象原理为上图所示，简单利于理解；
布隆过滤的原理后面讲，能保证不放行的一定不存在，放行的不一定存在。
上面两种方式都是被动的解决缓存穿透方案，此外我们还可以采用主动的方案预防缓存穿透，比如：增强id的复杂度避免被猜测id规律、做好数据的基础格式校验、加强用户权限校验、做好热点参数的限流。

注意

我们实际情况下redis中不能保存null，因此我们使用空字符串来表示请求的数据在数据库中不存在；（由于这个也是读操作，因此我们也要写上TTL）

    /*** 根据id查询商铺数据** @param id* @return*/@Overridepublic Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;// 1、从Redis中查询店铺数据String shopJson = stringRedisTemplate.opsForValue().get(key);Shop shop = null;// 2、判断缓存是否命中if (StrUtil.isNotBlank(shopJson)) {// 2.1 缓存命中，直接返回店铺数据shop = JSONUtil.toBean(shopJson, Shop.class);return Result.ok(shop);}// 2.2 缓存未命中，判断缓存中查询的数据是否是空字符串(isNotBlank把null和空字符串给排除了)if (Objects.nonNull(shopJson)){// 2.2.1 当前数据是空字符串（说明该数据是之前缓存的空对象），直接返回失败信息return Result.fail("店铺不存在");}// 2.2.2 当前数据是null，则从数据库中查询店铺数据shop = this.getById(id);// 4、判断数据库是否存在店铺数据if (Objects.isNull(shop)) {// 4.1 数据库中不存在，缓存空对象（解决缓存穿透），返回失败信息stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.SECONDS);return Result.fail("店铺不存在");}// 4.2 数据库中存在，重建缓存，并返回店铺数据stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);return Result.ok(shop);}

缓存雪崩

定义：同一时间几十万个缓存过期、redis宕机直接访问数据库

• 服务器做预热，将数据预批量存到redis中；如果设置为同一过期时间，那么就可能发生同时过期的问题；解决方式：设置随机过期时间30+(1~5)；
• 使用集群，什么哨兵可以发现主redis挂没挂掉，挂掉的话使用从redis去代替主节点；其中还要包含主从同步
• 给缓存业务添加降级限流策略，好像是挂掉之后；会暂停服务
• 给缓存业务添加降级限流策略；SpringCloud里面的功能
• 添加多级缓存：浏览器缓存、JVM缓存、Mysql等

缓存击穿

解决方式

互斥锁

获取互斥锁成功的例子：查询缓存，没有命中 -> 获取互斥锁成功 -> 去查询SQL -> 写入缓存 -> 释放锁
获取互斥锁失败的例子：查询缓存，没有命中 -> 获取互斥锁失败 -> 休眠一会 -> 循环直到缓存命中

逻辑过期

关于第一次查询缓存中没有数据，那么怎么进行？
个人理解：定义发生缓存击穿的情况为访问热点key的场景；所以关于热点信息应该是先手动缓存到redis中的；

有三种情况：
在缓存中设置一个逻辑的过期时间，
查询缓存 -> 发现逻辑过期 -> 获取一个互斥锁 -> 执行下面对应两种情况 -> 返回当前redis中的缓存获取互斥锁成功 -> 单开一个异步线程去从数据库中查询数据，并将查询到的数据写入缓存中+重置过期时间 -> 释放锁获取互斥锁失败 -> 返回当前redis中的缓存
查询缓存 -> 发现逻辑没有过期 -> 返回当前redis中的缓存

各有优点，看具体业务需求选择适合的方案

案例

互斥锁的技术实现

使用redis中的setnx外加过期时间，使用完之后删除；跟真正的还有差距，只是简单的实现；

获取锁

public boolean trylock(String key){//为使用@Autowire注入的Boolean flag = stringRedisTemplate.opsForvalue().setIfAbsent(key,"1",10,TimeUnit.SECOND);//不能直接返回flag，因为flag有可能为空return BooleanUtil.isTrue(flag);
}

解锁

public void unlock(key){stringRedisTemplate.delete(key);
}

查询

public Result queryById(Long id) {//先解决缓存穿透问题String key = CACHE_SHOP_KEY+id;String shopJson = stringRedisTemplate.opsForValue().get(key);String lockKey = LOCK_SHOP_KEY+id;System.out.println("shopJson为："+shopJson);//如果shopJson不为 null、""、"\t"，直接返回if(StringUtil.isNotBlank(shopJson)){return Result.ok(JSONUtil.toBean(shopJson, Shop.class));}//如果为"",代表之前请求没有存在的数据，发生穿透if (Objects.nonNull(shopJson)){return Result.fail("发生穿透，不存在");}//缓存不存在try {boolean isLock = trylock(lockKey);//没有抢到if(!isLock){//等待Thread.sleep(50);return queryById(id);}//获取锁成功//为null，则去数据库中查Shop shop = this.getById(id);if (shop == null) {//数据库中没有设置为""，过期时间为2SstringRedisTemplate.opsForValue().set(key,"", CACHE_NULL_TTL, TimeUnit.SECONDS);}//不为null，数据库中查到了，保存到redis中并返回stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.SECONDS);return Result.ok(shop);}catch (InterruptedException e) {throw new RuntimeException(e);}finally {unlock(lockKey);}}public boolean trylock(String key) {//为使用@Autowire注入的Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);//不能直接返回flag，因为flag有可能为空return BooleanUtil.isTrue(flag);}public void unlock(String key){stringRedisTemplate.delete(key);}

注意

• 递归的使用
• try中如果有return的话，先执行finally中的代码，再执行try中的return。
• trylock中锁这个key也要设置过期时间

Jmeter使用

Jmeter使用

逻辑过期的技术实现

这下面的流程图只是一个简单的核心案例，未命中直接返回空是为了简化流程，在真实使用中这地方是要着重处理的；
其他部分的逻辑是没问题的

之前的猜想是对的，刚开始的时候会提前往redis中存放逻辑过期

重新定义一个类RedisData，里面成员变量为data和LocalDateTime的过期时间

代码实现

创建一个逻辑过期数据类：

@Data
public class RedisData {/*** 过期时间*/private LocalDateTime expireTime;/*** 缓存数据*/private Object data;
}

ShopService中的代码：

    /*** 缓存重建线程池*/public static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);/*** 根据id查询商铺数据** @param id* @return*/@Overridepublic Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;// 1、从Redis中查询店铺数据，并判断缓存是否命中String shopJson = stringRedisTemplate.opsForValue().get(key);if (StrUtil.isBlank(shopJson)) {// 1.1 缓存未命中，直接返回失败信息return Result.fail("店铺数据不存在");}// 1.2 缓存命中，将JSON字符串反序列化未对象，并判断缓存数据是否逻辑过期RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);// 这里需要先转成JSONObject再转成反序列化，否则可能无法正确映射Shop的字段JSONObject data = (JSONObject) redisData.getData();Shop shop = JSONUtil.toBean(data, Shop.class);LocalDateTime expireTime = redisData.getExpireTime();if (expireTime.isAfter(LocalDateTime.now())) {// 当前缓存数据未过期，直接返回return Result.ok(shop);}// 2、缓存数据已过期，获取互斥锁，并且重建缓存String lockKey = LOCK_SHOP_KEY + id;boolean isLock = tryLock(lockKey);if (isLock) {// 获取锁成功，开启一个子线程去重建缓存CACHE_REBUILD_EXECUTOR.submit(() -> {try {this.saveShopToCache(id, CACHE_SHOP_LOGICAL_TTL);} finally {unlock(lockKey);}});}// 3、获取锁失败，再次查询缓存，判断缓存是否重建（这里双检是有必要的）shopJson = stringRedisTemplate.opsForValue().get(key);if (StrUtil.isBlank(shopJson)) {// 3.1 缓存未命中，直接返回失败信息return Result.fail("店铺数据不存在");}// 3.2 缓存命中，将JSON字符串反序列化未对象，并判断缓存数据是否逻辑过期redisData = JSONUtil.toBean(shopJson, RedisData.class);// 这里需要先转成JSONObject再转成反序列化，否则可能无法正确映射Shop的字段data = (JSONObject) redisData.getData();shop = JSONUtil.toBean(data, Shop.class);expireTime = redisData.getExpireTime();if (expireTime.isAfter(LocalDateTime.now())) {// 当前缓存数据未过期，直接返回return Result.ok(shop);}// 4、返回过期数据return Result.ok(shop);}/*** 将数据保存到缓存中** @param id            商铺id* @param expireSeconds 逻辑过期时间*/public void saveShopToCache(Long id, Long expireSeconds) {// 从数据库中查询店铺数据Shop shop = this.getById(id);// 封装逻辑过期数据RedisData redisData = new RedisData();redisData.setData(shop);redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));// 将逻辑过期数据存入Redis中stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(redisData));}/*** 获取锁** @param key* @return*/private boolean tryLock(String key) {Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);// 拆箱要判空，防止NPEreturn BooleanUtil.isTrue(flag);}/*** 释放锁** @param key*/private void unlock(String key) {stringRedisTemplate.delete(key);}

注意

RedisData中存放数据为Object类型，从redis中取出数据，第一次反序列化RedisData中data类型为JSONObject并不是Object类型，所以需要进行第二次反序列化；

缓存穿透和缓存击穿的工具类

单独缓存穿透工具类

@Slf4j
@Component
public class CacheClient_tmp {private final StringRedisTemplate stringRedisTemplate;public CacheClient_tmp(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 <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, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回错误信息return null;}// 6.存在，写入redisthis.set(key, r, time, unit);return r;}
}

注意

StringBoot中可以省略构造函数上面的@AutoWired，它可以自动添加上

逻辑过期时间

@Slf4j
@Component
public class CacheClient_tmp {private final StringRedisTemplate stringRedisTemplate;private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);public CacheClient_tmp(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 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.存在，直接返回return null;}// 4.命中，需要先把json反序列化为对象RedisData redisData = JSONUtil.toBean(json, RedisData.class);R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);LocalDateTime expireTime = redisData.getExpireTime();// 5.判断是否过期if(expireTime.isAfter(LocalDateTime.now())) {// 5.1.未过期，直接返回店铺信息return r;}// 5.2.已过期，需要缓存重建// 6.缓存重建// 6.1.获取互斥锁String lockKey = LOCK_SHOP_KEY + id;boolean isLock = tryLock(lockKey);// 6.2.判断是否获取锁成功if (isLock){// 6.3.成功，开启独立线程，实现缓存重建CACHE_REBUILD_EXECUTOR.submit(() -> {try {// 查询数据库R newR = dbFallback.apply(id);// 重建缓存this.setWithLogicalExpire(key, newR, time, unit);} catch (Exception e) {throw new RuntimeException(e);}finally {// 释放锁unlock(lockKey);}});}// 6.4.返回过期的商铺信息return r;}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);}
}

注意

使用了函数引用的知识，好像lamda表达式为一个函数引用；

秒杀业务

全局唯一ID

自增ID存在的问题

当用户抢购时，就会生成订单并保存到tb_voucher_order这张表中，而订单表如果使用数据库自增ID就存在一些问题：

• id的规律性太明显，容易出现信息的泄露，被不怀好意的人伪造请求；当ID规律过于明显时，存在以下一些缺点：

  • **安全性问题：**如果ID规律太明显，可能会使系统容易受到恶意攻击，例如暴力破解等。攻击者可以通过分析ID规律来推断出其他用户的ID，从而进行未授权的访问或操纵。
  • **隐私泄露风险：**如果ID规律太明显，可能导致用户的个人信息或敏感数据被曝光。攻击者可以根据规律推测出其他用户的ID，并通过这些ID获取到相应的数据，进而侵犯用户的隐私。
  • **数据可预测性：**当ID规律太明显时，使用这些规律的攻击者可以很轻易地猜测出其他实体（如订单、交易等）的ID。这可能破坏系统的数据安全性和防伪能力。
  • **扩展性受限：**如果ID规律太明显，可能会对系统的扩展性造成一定影响。当系统需要处理大量并发操作时，如果ID规律过于明显，可能导致多个操作同时对同一资源进行竞争，从而增加冲突和性能瓶颈。
  • **维护困难：**当ID规律太明显时，系统可能需要额外的资源和机制来保持规律的更新和变化，以确保安全性和数据完整性。这会增加系统的复杂度，并给维护带来挑战。

• 受单表数据量的限制，MySQL中表能够存储的数据有限，会出现分库分表的情况，id不能够一直自增

业界流传是500万行。超过500万行就要考虑分表分库了。阿里巴巴《Java 开发手册》提出单表行数超过 500 万行或者单表容量超过 2GB，就需要考虑分库分表了。

全局唯一ID（分布式ID）

• 全局唯一性：分布式ID保证在整个分布式系统中唯一性，不会出现重复的标识符。这对于区分和追踪系统中的不同实体非常重要。
• 高可用性：分布式ID生成器通常被设计为高可用的组件，可以通过水平扩展、冗余备份或集群部署来确保服务的可用性。即使某个节点或组件发生故障，仍然能够正常生成唯一的ID标识符。
• 安全性：分布式ID生成器通常是独立于应用程序和业务逻辑的。它们被设计为一个单独的组件或服务，可以被各种应用程序和服务所共享和使用，使得各个应用程序之间的ID生成过程互不干扰。
• 高性能：分布式ID生成器通常要求在很短的时间内生成唯一的标识符。为了实现低延迟，设计者通常采用高效的算法和数据结构，以及优化的网络通信和存储策略。
• 递增性：分布式ID通常可以被设计成可按时间顺序排序，以便更容易对生成的ID进行索引、检索或排序操作。这对于一些场景，如日志记录和事件溯源等，非常重要。

分布式ID的实现

• UUID
• Redis自增
• 数据库自增
• snowflake算法（雪花算法）
  这里我们使用自定义的方式实现：时间戳+序列号+数据库自增
  

@Component
public class RedisIdWorker {/*** 开始时间戳*/private static final long BEGIN_TIMESTAMP = 1640995200L;/*** 序列号的位数*/private static final int COUNT_BITS = 32;private StringRedisTemplate stringRedisTemplate;public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}public long nextId(String keyPrefix) {// 1.生成时间戳LocalDateTime now = LocalDateTime.now();long nowSecond = now.toEpochSecond(ZoneOffset.UTC);long timestamp = nowSecond - BEGIN_TIMESTAMP;// 2.生成序列号// 2.1.获取当前日期，精确到天String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));// 2.2.自增长long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);// 3.拼接并返回return timestamp << COUNT_BITS | count;}
}

测试类

@SpringBootTest
public class RedisIdWorkerTest {@Resourceprivate RedisIdWorker redisIdWorker;private ExecutorService es = Executors.newFixedThreadPool(500);/*** 测试分布式ID生成器的性能，以及可用性*/@Testpublic void testNextId() throws InterruptedException {// 使用CountDownLatch让线程同步等待CountDownLatch latch = new CountDownLatch(300);// 创建线程任务Runnable task = () -> {for (int i = 0; i < 100; i++) {long id = redisIdWorker.nextId("order");System.out.println("id = " + id);}// 等待次数-1latch.countDown();};long begin = System.currentTimeMillis();// 创建300个线程，每个线程创建100个id，总计生成3w个idfor (int i = 0; i < 300; i++) {es.submit(task);}// 线程阻塞，直到计数器归0时才全部唤醒所有线程latch.await();long end = System.currentTimeMillis();System.out.println("生成3w个id共耗时" + (end - begin) + "ms");}
}

http://localhost:8080/shop-detail.html?id=1访问店铺详情

实现优惠卷秒杀下单

业务流程

为什么产生超卖问题

乐观锁解决一人多单超卖问题

实现方式一：版本号法

首先我们要为 tb_seckill_voucher 表新增一个版本号字段 version ，线程1查询完库存，在进行库存扣减操作的同时将版本号+1，线程2在查询库存时，同时查询出当前的版本号，发现库存充足，也准备执行库存扣减操作，但是需要判断当前的版本号是否是之前查询时的版本号，结果发现版本号发生了改变，这就说明数据库中的数据已经发生了修改，需要进行重试（或者直接抛异常中断）

实现方式二：CAS法

CAS法类似与版本号法，但是不需要另外在添加一个 version 字段，而是直接使用库存替代版本号，线程1查询完库存后进行库存扣减操作，线程2在查询库存时，发现库存充足，也准备执行库存扣减操作，但是需要判断当前的库存是否是之前查询时的库存，结果发现库存数量发生了改变，这就说明数据库中的数据已经发生了修改，需要进行重试（或者直接抛异常中断）

        // 5、秒杀券合法，则秒杀券抢购成功，秒杀券库存数量减一boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>().eq(SeckillVoucher::getVoucherId, voucherId).eq(SeckillVoucher::getStock, voucher.getStock()).setSql("stock = stock -1"));

这又是什么原因呢？这就是乐观锁的弊端，我们只要发现数据修改就直接终止操作了，我们只需要修改一下判断条件，即只要库存大于0就可以进行修改，而不是库存数据修改我们就终止操作

        // 5、秒杀券合法，则秒杀券抢购成功，秒杀券库存数量减一boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>().eq(SeckillVoucher::getVoucherId, voucherId).gt(SeckillVoucher::getStock, 0).setSql("stock = stock -1"));

一人一单

添加业务：要求一个用户下单一张秒杀优惠券

SpringBoot中代理失效的集中方式

• 方法被非public（访问不到，SpringBoot中设定死了，不是public不执行）、final（使用AOP，所以方法不能被修改）、方法被static修饰
• 本类方法中即使调用本类中带有@Transactional的方法，也不会开启事务
• 其他方式就不记录了，反正也不决定使用了

注意

@Transactional的方法中调用非@Transactional的方法，那么非@Transactional的方法也有事务
事务只有提交之后，数据库中才更新

@Transactional
public void TUseNT() {Voucher voucher = new Voucher();voucher.setSubTitle("你好");save(voucher);TUseNTExistError();
}public void TUseNTExistError(){Voucher voucher = new Voucher();voucher.setSubTitle("你好1");save(voucher);Voucher voucher1 = new Voucher();voucher1.setSubTitle("你好2");save(voucher1);
}

---

上述方法中有坑点，就是事务的粒度太大了，直接在整个函数上面的。如果有访问数据库查找的代码非常耗时，并且有很多请求，那么直接加锁就会导致短时间内很多锁，数据库就挂掉了；例如：

@Transactional
public void TUseNT() {query1();//消耗很长时间query2();//消耗很长时间insert();
}

---

那么好啊，直接在insert()方法上面加事务；OK那么就又犯了经典错误导致事务不生效；没有事务的方法调用类内事务方法会导致事务不生效；下面结果为保存你好和你好1，并没有回滚；
原因：在TUseNT()调用TUseNTExistError();实际上为this.TUseNTExistError();
并不是SpringBoot创建的代理对象来调用TUseNTExistError()，所以就根本没有事务相关的逻辑。

public void TUseNT() {Voucher voucher = new Voucher();voucher.setSubTitle("你好");save(voucher);TUseNTExistError();
}
@Transactional
public void TUseNTExistError(){Voucher voucher = new Voucher();voucher.setSubTitle("你好1");save(voucher);Voucher voucher1 = new Voucher();voucher1.setSubTitle("你好2");save(voucher1);//假设在此时保存错误
}

---

其他失效场景：一口气怼完12种@Transactional的失效场景 - 程序yuan的文章 - 知乎
好像大家都不推荐使用@Transactional，推荐的有手动回滚；
为什么不推荐？

1. 12种情况稍微一不注意就跌进去了
2. 抛出异常才回滚，如果调用的逻辑是别人的一大堆，稍微捕获一下异常就GG
3. 各种传播范围搞错就GG

代码实现

引入AOP依赖，动态代理是AOP的常见实现之一

        <dependency><groupId>org.aspectj</groupId><artifactId>aspectjweaver</artifactId></dependency>

暴露动态代理对象，默认是关闭的

@EnableAspectJAutoProxy(exposeProxy = true)

    /*** 抢购秒杀券** @param voucherId* @return*/@Transactional@Overridepublic Result seckillVoucher(Long voucherId) {// 1、查询秒杀券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2、判断秒杀券是否合法if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 秒杀券的开始时间在当前时间之后return Result.fail("秒杀尚未开始");}if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 秒杀券的结束时间在当前时间之前return Result.fail("秒杀已结束");}if (voucher.getStock() < 1) {return Result.fail("秒杀券已抢空");}// 3、创建订单Long userId = ThreadLocalUtls.getUser().getId();synchronized (userId.toString().intern()) {// 创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(userId, voucherId);}}/*** 创建订单** @param userId* @param voucherId* @return*/@Transactionalpublic Result createVoucherOrder(Long userId, Long voucherId) {
//        synchronized (userId.toString().intern()) {// 1、判断当前用户是否是第一单int count = this.count(new LambdaQueryWrapper<VoucherOrder>().eq(VoucherOrder::getUserId, userId));if (count >= 1) {// 当前用户不是第一单return Result.fail("用户已购买");}// 2、用户是第一单，可以下单，秒杀券库存数量减一boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>().eq(SeckillVoucher::getVoucherId, voucherId).gt(SeckillVoucher::getStock, 0).setSql("stock = stock -1"));if (!flag) {throw new RuntimeException("秒杀券扣减失败");}// 3、创建对应的订单，并保存到数据库VoucherOrder voucherOrder = new VoucherOrder();long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);voucherOrder.setId(orderId);voucherOrder.setUserId(ThreadLocalUtls.getUser().getId());voucherOrder.setVoucherId(voucherOrder.getId());flag = this.save(voucherOrder);if (!flag) {throw new RuntimeException("创建秒杀券订单失败");}// 4、返回订单idreturn Result.ok(orderId);
//        }
}

注意点

下面代码虽然整个放到事务中，并且上锁，但是

    @Transactionalpublic Result createVoucherOrder(Long userId, Long voucherId) {synchronized (userId.toString().intern()) {//执行插入逻辑return Result.ok(orderId);}//这个地方释放锁
}//这个地方执行事务提交，数据库才更新

线程1执行到//这个地方释放锁；但是事务此时并没有提交，并释放锁
线程2获得锁，并执行锁里面的内容，但是锁里面的内容只能执行一次；因此就会执行两次插入，释放锁，线程2提交事务
线程1提交事务
此时数据库里面出现了两次插入的数据，按照业务逻辑只能插入一次；因此发生错误

---

锁住的对象要是同一个对象，此处不能用userId.toString()，它会返回一个新的String，好像还是在堆中的；不是String池中的，因此使用userId.toString().intern()。

上述方式在集群中存在致命问题

由于锁在JVM中，如果整个集群，就锁不住了；
1）搭建集群并实现负载均衡


然后在Nginx中配置负载均衡：

分布式锁

分布式锁的特点：

• 多进程可见
• 互斥。分布式锁必须能够确保在任何时刻只有一个节点能够获得锁，其他节点需要等待。
• 分布式锁应该具备高可用性，即使在网络分区或节点故障的情况下，仍然能够正常工作。（容错性）当持有锁的节点发生故障或宕机时，系统需要能够自动释放该锁，以确保其他节点能够继续获取锁。
• 高性能。分布式锁需要具备良好的性能，尽可能减少对共享资源的访问等待时间，以及减少锁竞争带来的开销。
• 安全性：（可重入性）如果一个节点已经获得了锁，那么它可以继续请求获取该锁而不会造成死锁。（锁超时机制）为了避免某个节点因故障或其他原因无限期持有锁而影响系统正常运行，分布式锁通常应该设置超时机制，确保锁的自动释放。
  

# 添加锁
set [key] [value] ex [time] nx

# 释放锁（除了使用del手动释放，还可超时释放）
del [key]

上锁逻辑和释放锁

为什么要写成创建对象然后再使用的原因：
因为后续优化需要使用创建对象。

public class SimpleRedisLock implements Lock {/*** RedisTemplate*/private StringRedisTemplate stringRedisTemplate;/*** 锁的名称*/private String name;public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {this.stringRedisTemplate = stringRedisTemplate;this.name = name;}/*** 获取锁** @param timeoutSec 超时时间* @return*/@Overridepublic boolean tryLock(long timeoutSec) {String id = Thread.currentThread().getId() + "";// SET lock:name id EX timeoutSec NXBoolean result = stringRedisTemplate.opsForValue().setIfAbsent("lock:" + name, id, timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(result);}/*** 释放锁*/@Overridepublic void unlock() {stringRedisTemplate.delete("lock:" + name);}
}

在上面代码的基础上，将synchronized包住的逻辑改成自己的分布式锁的形式；

  /*** 抢购秒杀券** @param voucherId* @return*/@Transactional@Overridepublic Result seckillVoucher(Long voucherId) {// 1、查询秒杀券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 2、判断秒杀券是否合法if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 秒杀券的开始时间在当前时间之后return Result.fail("秒杀尚未开始");}if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 秒杀券的结束时间在当前时间之前return Result.fail("秒杀已结束");}if (voucher.getStock() < 1) {return Result.fail("秒杀券已抢空");}//        synchronized (userId.toString().intern()) {// 创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效//        }// 3、创建订单Long userId = ThreadLocalUtls.getUser().getId();SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);boolean isLock = lock.tryLock(1200);if (!isLock) {// 索取锁失败，重试或者直接抛异常（这个业务是一人一单，所以直接返回失败信息）return Result.fail("一人只能下一单");}try {// 索取锁成功，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(userId, voucherId);} finally {lock.unlock();}}/*** 创建订单** @param userId* @param voucherId* @return*/@Transactionalpublic Result createVoucherOrder(Long userId, Long voucherId) {
//        synchronized (userId.toString().intern()) {// 1、判断当前用户是否是第一单int count = this.count(new LambdaQueryWrapper<VoucherOrder>().eq(VoucherOrder::getUserId, userId));if (count >= 1) {// 当前用户不是第一单return Result.fail("用户已购买");}// 2、用户是第一单，可以下单，秒杀券库存数量减一boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>().eq(SeckillVoucher::getVoucherId, voucherId).gt(SeckillVoucher::getStock, 0).setSql("stock = stock -1"));if (!flag) {throw new RuntimeException("秒杀券扣减失败");}// 3、创建对应的订单，并保存到数据库VoucherOrder voucherOrder = new VoucherOrder();long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);voucherOrder.setId(orderId);voucherOrder.setUserId(ThreadLocalUtls.getUser().getId());voucherOrder.setVoucherId(voucherOrder.getId());flag = this.save(voucherOrder);if (!flag) {throw new RuntimeException("创建秒杀券订单失败");}// 4、返回订单idreturn Result.ok(orderId);
//        }

注意点

try…finally…确保发生异常时锁能够释放，注意这给地方不要使用catch，A事务方法内部调用B事务方法，A事务方法不能够直接catch，否则会导致事务失效（createVoucherOrder的方法）

上述分布式锁存在的问题

执行业务逻辑太长会导致锁失效
Thread1 获得锁，并且执行业务逻辑要10S，锁的过期时间为5S；
当5S后突然来个Thread2，Thread2获得了锁。Thread2执行业务5S的时候，Thread1执行完业务逻辑并删除redis锁，那么如果还有Thread3，它也能获得锁；因此锁失效了下图所示：

稍微增强一下的版本

实现思路：在删除锁之前查看是不是自己当初的那把锁，是的话再删除
如何判断是不是当初那把锁？之前value中保存的为当前线程的ID，线程ID的产生逻辑为当前JVM创建一个线程就自增一下线程ID。那么在分布式服务器中，线程ID就会发生重复；因此视频中采用UUID（几乎不会重复，而且还有很多版本，没有深究到底会不会重复）+线程ID的解决方式。

key保持不动，之前key为 “lock:” + name(service中指定，用于和其他service方法区分，例如指定"order") + userId（为了实现一个用户只能下一单); value由原来的ThreadId变为UUID+ThreadId;

package com.hmdp.utils.lock.impl;import cn.hutool.core.lang.UUID;
import com.hmdp.utils.lock.Lock;
import org.springframework.data.redis.core.StringRedisTemplate;import java.util.concurrent.TimeUnit;/*** @author ghp* @title* @description*/
public class SimpleRedisLock implements Lock {/*** RedisTemplate*/private StringRedisTemplate stringRedisTemplate;/*** 锁的名称*/private String name;/*** key前缀*/public static final String KEY_PREFIX = "lock:";/*** ID前缀*/public static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {this.stringRedisTemplate = stringRedisTemplate;this.name = name;}/*** 获取锁** @param timeoutSec 超时时间* @return*/@Overridepublic boolean tryLock(long timeoutSec) {String threadId = ID_PREFIX + Thread.currentThread().getId() + "";// SET lock:name id EX timeoutSec NXBoolean result = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(result);}/*** 释放锁*/@Overridepublic void unlock() {// 判断 锁的线程标识 是否与 当前线程一致String currentThreadFlag = ID_PREFIX + Thread.currentThread().getId();String redisThreadFlag = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);if (currentThreadFlag != null || currentThreadFlag.equals(redisThreadFlag)) {// 一致，说明当前的锁就是当前线程的锁，可以直接释放stringRedisTemplate.delete(KEY_PREFIX + name);}// 不一致，不能释放}
}

使用Lua脚本

你可能还会想，判断锁和释放锁在同一个方法中，并且两者之间没有别的代码，为什么会发生阻塞呢？JVM的垃圾回收机制会导致短暂的阻塞；

那么我们该如何保障 判断锁 和 释放锁 这连段代码的原子性呢？答案是使用Lua脚本；
那么Lua脚本是如何确保原子性的呢？Redis使用（支持）相同的Lua解释器，来运行所有的命令。Redis还保证脚本以原子方式执行：在执行脚本时，不会执行其他脚本或Redis命令。这个语义类似于MULTI（开启事务）/EXEC（触发事务，一并执行事务中的所有命令）。从所有其他客户端的角度来看，脚本的效果要么仍然不可见，要么已经完成。

释放锁的业务流程是这样的：

• 获取锁中的线程标示
• 判断是否与指定的标示（当前线程标示）一致
• 如果一致则释放锁（删除）
• 如果不一致则什么都不做
  编写lua文件
  

-- 比较缓存中的线程标识与当前线程标识是否一致
if (redis.call('get', KEYS[1]) == ARGV[1]) then-- 一致，直接删除return redis.call('del', KEYS[1])
end
-- 不一致，返回0
return 0

package com.hmdp.utils.lock.impl;import cn.hutool.core.lang.UUID;
import com.hmdp.utils.lock.Lock;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;import java.util.Collections;
import java.util.concurrent.TimeUnit;/*** @author ghp* @title* @description*/
public class SimpleRedisLock implements Lock {/*** RedisTemplate*/private StringRedisTemplate stringRedisTemplate;/*** 锁的名称*/private String name;/*** key前缀*/private static final String KEY_PREFIX = "lock:";/*** ID前缀*/private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {this.stringRedisTemplate = stringRedisTemplate;this.name = name;}/*** 获取锁** @param timeoutSec 超时时间* @return*/@Overridepublic boolean tryLock(long timeoutSec) {String threadId = ID_PREFIX + Thread.currentThread().getId() + "";// SET lock:name id EX timeoutSec NXBoolean result = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(result);}/*** 加载Lua脚本；不应该在解锁的时候加载，提前加载*/private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;//泛型中存放返回的类型static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("lua/unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}/*** 释放锁*/@Overridepublic void unlock() {// 执行lua脚本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),ID_PREFIX + Thread.currentThread().getId());}
}

Redission

为什么使用Redission？
我们之前写的代码存在4个问题：

经过优化1和优化2，我们实现的分布式锁已经达到生产可用级别了，但是还不够完善，比如：

• 分布式锁不可重入：不可重入是指同一线程不能重复获取同一把锁。比如，方法A中调用方法B，方法A需要获取分布式锁，方法B同样需要获取分布式锁，线程1进入方法A获取了一次锁，进入方法B又获取一次锁，由于锁不可重入，所以就会导致死锁。（递归场景）
• 分布式锁不可重试：获取锁只尝试一次就返回false，没有重试机制，这会导致数据丢失，比如线程1获取锁，然后要将数据写入数据库，但是当前的锁被线程2占用了，线程1直接就结束了而不去重试，这就导致数据发生了丢失
• 分布式锁超时释放：超时释放机机制虽然一定程度避免了死锁发生的概率，但是如果业务执行耗时过长，期间锁就释放了，这样存在安全隐患。锁的有效期过短，容易出现业务没执行完就被释放，锁的有效期过长，容易出现死锁，所以这是一个大难题！
  我们可以设置一个较短的有效期，但是加上一个 心跳机制 和 自动续期：在锁被获取后，可以使用心跳机制并自动续期锁的持有时间。通过定期发送心跳请求，显示地告知其他线程或系统锁还在使用中，同时更新锁的过期时间。如果某个线程持有锁的时间超过了预设的有效时间，其他线程可以尝试重新获取锁。
• 主从一致性问题：如果Redis提供了主从集群，主从同步存在延迟，线程1获取了锁
  并且解决上述问题，代码复杂Redission已经提供了解决方案。

Redission快速入门(单节点)

• 引入Redisson依赖

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

• 配置Redisson客户端

@Configuration
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private String port;@Value("${spring.redis.password}")private String password;/*** 创建Redisson配置对象，然后交给IOC管理** @return*/@Beanpublic RedissonClient redissonClient() {// 获取Redisson配置对象Config config = new Config();// 添加redis地址，这里添加的是单节点地址，也可以通过 config.userClusterServers()添加集群地址config.useSingleServer().setAddress("redis://" + this.host + ":" + this.port).setPassword(this.password);// 获取RedisClient对象，并交给IOC进行管理return Redisson.create(config);}
}

温馨提示：此外还有一种引入方式，可以引入 redission 的 starter 依赖，然后在yml文件中配置Redisson，但是不推荐这种方式，因为他会替换掉 Spring官方 提供的这套对 Redisson 的配置

• 使用：我们只需要修改一下使用锁的地方，其它的业务代码都不需要改

        // 3、创建订单（使用分布式锁）Long userId = ThreadLocalUtls.getUser().getId();RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);boolean isLock = lock.tryLock();

boolean tryLock(long time, TimeUnit unit);:指定等待获取锁的最长时间。如果在这个时间内无法获取锁，则返回 false。
boolean tryLock(long waitTime, long leaseTime, TimeUnit unit);:尝试获取锁，如果在指定的 waitTime 内获取成功，返回 true，并在 leaseTime 到期后自动释放锁。
tryLock():waitTime的默认值是-1，代表不等待，leaseTime的默认值是30，unit默认值是 seconds ，也就是锁超过30秒还没有释放就自动释放。

Redis可重入、可重试、可续约的原理

Redisson内部释放锁，并不是直接执行del命令将锁给删除，而是将锁以hash数据结构的形式存储在Redis中，每次获取锁，都将value的值+1，每次释放锁，都将value的值-1，只有锁的value值归0时才会真正的释放锁，从而确保锁的可重入性；其中field中存放的是能够区别是否同属一个进程的标识；

Redisson分布式锁原理：

• 如何解决可重入问题：利用hash结构记录线程id和重入次数。
• 如何解决可重试问题：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制。
• 如何解决超时续约问题：利用watchDog，每隔一段时间（releaseTime / 3），重置超时时间。（不设置过期时间，每隔10秒延期30秒）
• 如何解决主从一致性问题：利用Redisson的multiLock，多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功

缺陷：运维成本高、实现复杂

解决主从一致

主从节点不一致性产生的原因：

在主节点中完成增加、删除、修改操作；在从节点中完成查询操作；主节点和从节点之间需要进行主从同步；
虽然主从同步用的时间很少，但是极端情况为主节点刚完成操作就挂了，从节点没来及更新就发生了主从不一致的问题；

解决方式

三个主节点都进行增删改查的操作，从节点只做备份；并且三个主节点操作返回值为一样的才认为是有效的；
主节点挂了，从节点顶上去。

代码实现

@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient(){// 配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");// 创建RedissonClient对象return Redisson.create(config);}@Beanpublic RedissonClient redissonClient2(){// 配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6380");// 创建RedissonClient对象return Redisson.create(config);}@Beanpublic RedissonClient redissonClient3(){// 配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6381");// 创建RedissonClient对象return Redisson.create(config);}
}

测试类：

@SpringBootTest
class RedissonTest {@Resourceprivate RedissonClient redissonClient;@Resourceprivate RedissonClient redissonClient2;@Resourceprivate RedissonClient redissonClient3;private RLock lock;//    @BeforeEach
//    void setUp() {
//        lock = redissonClient.getLock("order");
//    }@BeforeEachvoid setUp() {RLock lock1 = redissonClient.getLock("hihi");RLock lock2 = redissonClient2.getLock("hihi");RLock lock3 = redissonClient3.getLock("hihi");lock = redissonClient.getMultiLock(lock1, lock2, lock3);}@Testvoid method1() throws InterruptedException {// 尝试获取锁boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS);if (!isLock) {log.error("获取锁失败 .... 1");return;}try {log.info("获取锁成功 .... 1");method2();log.info("开始执行业务 ... 1");} finally {log.warn("准备释放锁 .... 1");lock.unlock();}}void method2() {// 尝试获取锁boolean isLock = lock.tryLock();if (!isLock) {log.error("获取锁失败 .... 2");return;}try {log.info("获取锁成功 .... 2");log.info("开始执行业务 ... 2");} finally {log.warn("准备释放锁 .... 2");lock.unlock();}}
}

Redis优化秒杀

关于判断秒杀库存和校验一人一单是从数据库中查询，所以速度很快；
但是减库存和创建订单是增加和修改操作，速度很慢；
我们可以根据卖饭的操作：
1、前台工作人员用来看当前是否库存充足和用户不能多买，能买就打印小票，
2、后厨根据打印的小票来做饭（异步）

现在实现逻辑：
将String类型的 票的库存（value）、Set类型的 票的id（key）：用户id（value） 存放到redis中；用来判断是否能买；
能买就异步执行SQL保存操作；不能买直接返回；很大程度上减少了和数据库之间的通信；

同样使用lua来完成redis中的原子操作，判断库存和购买必须要一块执行。

-- 优惠券id
local voucherId = ARGV[1];
-- 用户id
local userId = ARGV[2];
-- 库存的key
local stockKey = 'seckill:stock:' .. voucherId;
-- 订单key
local orderKey = 'seckill:order:' .. voucherId;-- 判断库存是否充足 get stockKey > 0 ?
local stock = redis.call('GET', stockKey);
if (tonumber(stock) <= 0) then-- 库存不足，返回1return 1;
end-- 库存充足，判断用户是否已经下过单 SISMEMBER orderKey userId
if (redis.call('SISMEMBER', orderKey, userId) == 1) then-- 用户已下单，返回2return 2;
end-- 库存充足，没有下过单，扣库存、下单
redis.call('INCRBY', stockKey, -1);
redis.call('SADD', orderKey, userId);
-- 返回0，标识下单成功
return 0;

下面方法的逻辑为：
创建一个阻塞队列，这个阻塞队列中存储秒杀订单，如果队列中没有订单那么线程阻塞，如果有订单那么线程继续
创建一个实现Runable类的任务方法
定义一个函数并在Service类完成初始化之后，执行本方法，在这个方法中开启线程去执行任务
在Servivce中经过redis检验能够购买的情况下，生成秒杀订单的对象
将秒杀订单的对象存储到数据库中，并且是开启事务的存储，由于存储订单为重新开启一个线程进行存储，因此相关事务方面存在一些坑，而且我感觉他写的有问题，他令成员proxy为同一个，后续他又写成队列的形式，因此打算放弃这部分多线程代理的研究。

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;/*** 当前类初始化完毕就立马执行该方法*/@PostConstructprivate void init() {// 执行线程任务SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}/*** 存储订单的阻塞队列*/private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);/*** 线程池*/private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();/*** 线程任务: 不断从阻塞队列中获取订单*/private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while (true) {// 从阻塞队列中获取订单信息，并创建订单try {VoucherOrder voucherOrder = orderTasks.take();handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理订单异常", e);}}}}/*** 创建订单** @param voucherOrder*/private void handleVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);boolean isLock = lock.tryLock();if (!isLock) {// 索取锁失败，重试或者直接抛异常（这个业务是一人一单，所以直接返回失败信息）log.error("一人只能下一单");return;}try {// 创建订单（使用代理对象调用，是为了确保事务生效）proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}/*** 加载 判断秒杀券库存是否充足 并且 判断用户是否已下单 的Lua脚本*/private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}/*** VoucherOrderServiceImpl类的代理对象* 将代理对象的作用域进行提升，方面子线程取用*/private IVoucherOrderService proxy;/*** 抢购秒杀券** @param voucherId* @return*/@Transactional@Overridepublic Result seckillVoucher(Long voucherId) {// 1、执行Lua脚本，判断用户是否具有秒杀资格Long result = null;try {result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(),ThreadLocalUtls.getUser().getId().toString());} catch (Exception e) {log.error("Lua脚本执行失败");throw new RuntimeException(e);}if (result != null && !result.equals(0L)) {// result为1表示库存不足，result为2表示用户已下单int r = result.intValue();return Result.fail(r == 2 ? "不能重复下单" : "库存不足");}// 2、result为0，用户具有秒杀资格，将订单保存到阻塞队列中，实现异步下单long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);// 创建订单VoucherOrder voucherOrder = new VoucherOrder();voucherOrder.setId(orderId);voucherOrder.setUserId(ThreadLocalUtls.getUser().getId());voucherOrder.setVoucherId(voucherId);// 将订单保存到阻塞队列中orderTasks.add(voucherOrder);// 索取锁成功，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();this.proxy = proxy;return Result.ok();}/*** 创建订单** @param voucherOrder* @return*/@Transactional@Overridepublic void createVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();Long voucherId = voucherOrder.getVoucherId();// 1、判断当前用户是否是第一单int count = this.count(new LambdaQueryWrapper<VoucherOrder>().eq(VoucherOrder::getUserId, userId));if (count >= 1) {// 当前用户不是第一单log.error("当前用户不是第一单");return;}// 2、用户是第一单，可以下单，秒杀券库存数量减一boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>().eq(SeckillVoucher::getVoucherId, voucherId).gt(SeckillVoucher::getStock, 0).setSql("stock = stock -1"));if (!flag) {throw new RuntimeException("秒杀券扣减失败");}// 3、将订单保存到数据库flag = this.save(voucherOrder);if (!flag) {throw new RuntimeException("创建秒杀券订单失败");}}
}

Redis消息队列实现异步秒杀

弊端

前面我们使用 Java 自带的阻塞队列 BlockingQueue 实现消息队列，这种方式存在以下几个严重的弊端：

• 信息可靠性没有保障，BlockingQueue 的消息是存储在内存中的，无法进行持久化，一旦程序宕机或者发生异常，会直接导致消息丢失
• 消息容量有限，BlockingQueue 的容量有限，无法进行有效扩容，一旦达到最大容量限制，就会抛出OOM异常
  所以这里我们可以选择采用其它成熟的的（和之前分布式锁一样）MQ，比如：RabbitMQ、RocketMQ、Kafka等，但是本项目是为了学习Redis而设计的，所以这里我们将要学习如何使用Redis实现一个相对可靠的消息队列（自己实现的肯定没法和别人成熟的产品相比）
  
  感觉消息队列不用redis好，算了跳
  

达人探店

Feed流关注推送

什么是Feed流？

关注推送也叫做Feed流，直译为投喂。为用户持续的提供“沉浸式”的体验，通过无限下拉刷新获取新的信息。Feed流是一种基于用户个性化需求和兴趣的信息流推送方式，常见于社交媒体、新闻应用、音乐应用等互联网平台。Feed流通过算法和用户行为数据分析，动态地将用户感兴趣的内容以流式方式呈现在用户的界面上。

Feed流产品有两种常见模式：

• 时间排序（Timeline）：不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈
  • 优点：信息全面，不会有缺失。并且实现也相对简单
  • 缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低
• 智能排序：利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户
  • 优点：投喂用户感兴趣信息，用户粘度很高，容易沉迷
  • 缺点：如果算法不精准，可能起到反作用
    本例中的个人页面，是基于关注的好友来做Feed流，因此采用Timeline的模式。该模式的实现方案有三种：
    
• 拉模式：也叫做读扩散。博主发送Blog先将Blog存到邮箱中，其他关注博主的用户从他的邮箱中读取
  • 延迟较高，当用户读取数据时才去关注的人里边去读取数据，假设用户关注了大量的用户，那么此时就会拉取海量的内容，对服务器压力巨大。
• 推模式：也叫做写扩散。在推模式中，数据提供方主动将最新的数据推送给终端用户或应用程序。数据提供方会实时地将数据推送到终端用户或应用程序，而无需等待请求。
  • 内存耗费大，假设一个大V写信息，很多人关注他， 就会写很多份数据到粉丝那边去
• 推拉结合：也叫做读写混合，兼具推和拉两种模式的优点。在推拉结合模式中，数据提供方会主动将最新的数据推送给终端用户或应用程序，同时也支持用户通过拉取的方式来获取数据。这样可以实现实时的数据更新，并且用户也具有按需获取数据的能力。推拉模式是一个折中的方案，站在发件人这一段，如果是个普通的人，那么我们采用写扩散的方式，直接把数据写入到他的粉丝中去，因为普通的人他的粉丝关注量比较小，所以这样做没有压力，如果是大V，那么他是直接将数据先写入到一份到发件箱里边去，然后再直接写一份到活跃粉丝收件箱里边去，现在站在收件人这端来看，如果是活跃粉丝，那么大V和普通的人发的都会直接写入到自己收件箱里边来，而如果是普通的粉丝，由于他们上线不是很频繁，所以等他们上线时，再从发件箱里边去拉信息
  

实现

当前项目用户量比较小，所以这里我们选择使用推模式，延迟低、内存占比也没那么大
由于我们需要实现分页查询功能，这里我们可以选择 list 或者 SortedSet，而不能使用Set，因为Set是无需的， list是有索引的，SortedSet 是有序的，那么我们该如何选择呢？
如果我们选择 list 会存在索引漂移现象（这个在Vue中也存在，就是查询0-10个记录，然后新插入一个记录，查询第二页的时候第一个为第一页中显示过的10），从而导致读取重复数据，所以我们不能选择使用 list。

我们可以选择使用滚动分页，我们使用SortedSet，如果使用排名和使用角标是一样的，但是SortedSet可以按照Score排序（Score默认按照时间戳生成，所以是固定的），每次我们可以选择比之前Score较小的，这样就能够实现滚动排序，从而防止出现问题。我们在后端根据前端返回的最小时间戳和偏移个数去redis中查ID，格局list找最小的时间戳和计算最小时间戳相同个数为偏移个数，返回前端。这里有个Bug，需要判断上次时间戳和这次时间戳是否一样，一样的话两个偏移量相加。

    /*** 保存探店笔记** @param blog* @return*/@Overridepublic Result saveBlog(Blog blog) {Long userId = ThreadLocalUtls.getUser().getId();blog.setUserId(userId);// 保存探店笔记boolean isSuccess = this.save(blog);if (!isSuccess){return Result.fail("笔记保存失败");}// 查询笔记作者的所有粉丝List<Follow> follows = followService.list(new LambdaQueryWrapper<Follow>().eq(Follow::getFollowUserId, userId));// 将笔记推送给所有的粉丝for (Follow follow : follows) {// 获取粉丝的idLong id = follow.getUserId();// 推送笔记String key = FEED_KEY + id;stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());}return Result.ok(blog.getId());}