Redis的实战篇

1.短信的登入注册

可以使用RandomUtil.randomNumbers(6)生成6位随机数

1.1 使用session来存储验证码

题外话：session与Cookie的关系:

• 当用户登录的时候，服务器在Session中新增一个新记录，并把SessionID返回给客户端（通过HTTP响应的Set-Cookie字段返回）。
• 客户端后续再给服务器发送请求的时候，需要在请求中带上SessionID（通过HTTP请求中的Cookie字段带上）
• 服务器收到请求之后，根据SessionID在Session信息中获取到对应的用户信息，再进行后续操作，找不到则重新创建Session，并把SessionID返回。

1.1.1 验证用户的登入状态

使用拦截器(为什么要使用拦截器？)
1.保证用户的登入状态
2.在集群模式下，可以获得用户的信息，存储在ThreadLocal下
拦截器的使用 类 implements HandlerInterceptor(三个方法), 然后需要在Configuraion中配置addInterceptor

1.1.2 需要注意用户的隐藏信息

返回的数据要有限的，可以使用BeanUtils.copyProperties()来复制对象，然后再返回

1.2.3问题

每个tomcat中都有一份属于自己的session,假设用户第一次访问第一台tomcat，并且把自己的信息存放到第一台服务器的session中，但是第二次这个用户访问到了第二台tomcat，那么在第二台服务器上，肯定没有第一台服务器存放的session，所以此时 整个登录拦截功能就会出现问题。

1.2 使用redis来存储session

使用UUID.randomUUID().toString()来生成一个唯一的token
在去返回给前端，前端在请求的时候，需要在请求头中带上这个token
在经过拦截器的时候，我们可以通过这个token来获取到用户的信息

1.3 优化

有一些不用登入的接口，我们可以不用拦截器，直接放行
可以设置2个拦截器。（解决刷新登录token令牌的问题）

• 第一个拦截器，用来拦截token，把用户的信息存放到ThreadLocal中
• 第二个拦截器，用来判断用户是否登入，如果没有登入，就返回一个错误信息

2.商户查询缓存

缓存(Cache),就是数据交换的缓冲区,俗称的缓存就是缓冲区内的数据,一般从数据库中获取,存储于本地代码
浏览器缓存：主要是存在于浏览器端的缓存

**应用层缓存：**可以分为tomcat本地缓存，比如之前提到的map，或者是使用redis作为缓存

**数据库缓存：**在数据库中有一片空间是 buffer pool，增改查数据都会先加载到mysql的缓存中

2.1使用redis来缓存商户数据

较为常规，使用redis的String类型来存储数据

2.2 缓存更新策略

• 内存淘汰：redis自动进行，当redis内存达到咱们设定的max-memery的时候，会自动触发淘汰机制，淘汰掉一些不重要的数据(可以自己设置策略方式)
• 超时剔除: 设置一个过期时间，当过期时间到了，就会自动删除
• 主动更新：当数据发生变化的时候，我们可以主动去更新缓存
• 问题：数据库与redis出现数据不一致的情况
  • 1.是写少读多的情况，可以使用以把缓存删除，等待再次查询时，将缓存中的数据加载出来
  • 2.删除删除缓存还是更新缓存
  • 3.何保证缓存与数据库的操作的同时成功或失败
• 需要先操作数据库，再删除缓存,防止高并发时候，出现了脏数据

2.3 缓存穿透

防止黑客通过一些特殊的字符，来绕过缓存，直接访问数据库

• 1.缓存空对象

• 就是当数据库中没有这个数据的时候，我们也把这个空对象存放到缓存中，访问时候，先去缓存中查找，如果没有，再去数据库中查找

• 2.布隆过滤器（可能会出现误判）
  解决方案：
  • 缓存null值
  • 布隆过滤
  • 增强id的复杂度，避免被猜测id规律
  • 做好数据的基础格式校验
  • 加强用户权限校验
  • 做好热点参数的限流

2.4缓存雪崩问题及解决思路

解决方案：

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

2.5 缓存击穿问题(热点Key问题)

就是在高并发的情况下，一个热点key过期了，导致大量的请求直接访问数据库，导致数据库压力过大
解决方案：

• 1.使用互斥锁
• 2.使用逻辑过期

2.5.1 使用互斥锁

• 1.在获取缓存的时候，先去获取锁，如果获取不到锁，就等待一段时间，再去获取锁，使用了串行的方式

• 2.获得锁的去查询数据库，然后再去更新缓存
  问题：1.性能不行 2.可能有死锁的情况
  如何实现：1.s使用redis的setIfabsent 来去设置锁

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

2.5.2 使用逻辑过期

什么是逻辑过期：
就是在存储redis的时候，不去设置过期时间，而是在对象的元素设置一个过期时间
在使用redis的时候，先去判断这个过期时间，如果过期了，开启一个线程去更新缓存，
在这个线程去更新缓存的时候，其他的线程去访问缓存，发现缓存过期了，就去返回的是脏数据。
代码实现：

   //使用线程的模式import java.util.concurrent.ExecutorService;private static final ExecutorService executorService = Executors.newFixedThreadPool(10);
//核心代码
boolean isLock = tryLock(lockKey);
// 6.2.判断是否获取锁成功if (isLock){CACHE_REBUILD_EXECUTOR.submit( ()->{try{//重建缓存this.saveShop2Redis(id,20L);}catch (Exception e){throw new RuntimeException(e);}finally {
unlock(lockKey);}});}

2.5.3 使用泛型来去使用工具类

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

    public <R ,ID> R ShopThrough(String keyprefix,Class<R>Type,ID id,Long time,TimeUnit timeUnit,Function<ID,R>dpFallBack){String key = keyprefix+id;String s = stringRedisTemplate.opsForValue().get(key);if(StrUtil.isNotBlank(s)){return BeanUtil.toBean(s,Type);}R r = dpFallBack.apply(id);if(r==null){stringRedisTemplate.opsForValue().set(key,"",time,timeUnit);return null;}this.set(key,r,time,timeUnit);return r;}

根据指定的key查询缓存，并反序列化为指定类型，利用逻辑过期时间解决缓存击穿问题

    public <R, ID> R ShopLogicDelete(String keyprefix, Class<R> Type, ID id, Long time, TimeUnit timeUnit, Function<ID, R> dpFollBack) {String key = keyprefix + id;String s = stringRedisTemplate.opsForValue().get(key);if (StrUtil.isBlank(s)) {return null;}RedisData redisData = BeanUtil.copyProperties(s, RedisData.class);JSONObject data = (JSONObject) redisData.getData();R r = BeanUtil.toBean(data, Type);if (redisData.getExpireTime().isAfter(LocalDateTime.now())) {return r;}boolean lock = onLock(key);if (lock) {CACHE_REBUILD_EXECUTOR.submit(() -> {try {R r1 = dpFollBack.apply(id);this.setWithLogicalExpire(key, r1, time, timeUnit);} catch (Exception e) {throw new RuntimeException(e);} finally {offLock(key);}});}return r;}

3.优惠卷秒杀

3.1 全局唯一ID

为什么使用全局ID:

• 1.id具有太明显的规则，用户或者说商业对手很容易猜测出来我们的一些敏感信息
• 2.mysql的单表的容量不宜超过500W，数据量过大之后，我们要进行拆库拆表，但拆分表了之后，他们从逻辑上讲他们是同一张表，所以他们的id是不能一样的， 于是乎我们需要保证id的唯一性
  如何生成Id:时间戳+序列号,是一个64位的数字，前32位是时间戳，后32位是序列号

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

3.2 实现秒杀下单

3.2.1 超卖问题

注意的是：由于基础的秒杀的代码不同，如使用MyBatisPlus和MyBatis的代码不同，结果会有不同的结果
如：超卖或少卖问题

超卖问题出现的原因：高并发的问题与原子性问题

3.2.2 解决方案

• 1.使用悲观锁: 悲观锁可以实现对于数据的串行化执行，比如syn，和lock都是悲观锁的代表，同时，悲观锁中又可以再细分为公平锁，非公平锁，可重入锁，等等
• 2.使用乐观锁: 乐观锁是一种乐观的思想，他认为数据的读取是不会出现问题的，只有在更新的时候才会出现问题，所以在更新的时候，他会去判断数据是否被修改，如果没有被修改，就会去更新，如果被修改了，就会去重新读取数据，再去更新(cas)

3.2.3 悲观锁

就是加入一个锁，这里就不说了

3.2.4 乐观锁

boolean success = seckillVoucherService.update().setSql("stock= stock -1").eq("voucher_id", voucherId).update().gt("stock",0); 

就是在更新的时候，去判断库存是否大于0，如果大于0，就去更新，如果小于0，就不去更新

3.3 优惠券秒杀-一人一单

还是因为高并发的问题，导致了一个用户可以多次下单，乐观锁比较适合更新数据，而现在是插入数据，所以我们需要使用悲观锁操作

我们使用悲观锁时要考虑控制锁粒度：

• 使用了用户的Id来作为锁的粒度，这样就可以保证一个用户只能下一单
• 但是需要注意的是 Id.toString每一次都是不一样的，所以我们需要使用Id.toString.intern()来去保证是同一个对象

还是有问题的：问题的原因在于当前方法被spring的事务控制，如果你在方法内部加锁，可能会导致当前方法事务还没有提交，但是锁已经释放也会导致问题，所以我们选择将当前方法整体包裹起来，确保事务不会出现问题：

但是要用事务生效，需要去代理来生效IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();

• 在在SpringBoot的启动类上加上@EnableAspectJAutoProxy(exposeProxy = true)来暴露代理对象

3.4 集群环境下的并发问题

在多个tomcat中，使用了集群，每一个集群都会有一个自己jvm,使用syn锁是不行的，因为每一个jvm都会有自己的锁，所以我们需要使用redis来去实现锁

3.4.1 使用分布式锁

• 获取锁：

  • 互斥：确保只能有一个线程获取锁
  • 非阻塞：尝试一次，成功返回true，失败返回false
  • 使用setnx命令去保证原子性

• 释放锁：

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

3.4.2 分布式锁的问题

持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明
解决方案：要去判断锁是否是自己的

• 但是要去保证原子性的，使用lua的脚本来去实现

    -- 这里的 KEYS[1] 就是锁的key，这里的ARGV[1] 就是当前线程标示-- 获取锁中的标示，判断是否与当前线程标示一致if (redis.call('GET', KEYS[1]) == ARGV[1]) then-- 一致，则删除锁return redis.call('DEL', KEYS[1])end-- 不一致，则直接返回return 0 

使用

 StringRedisTemple stringRedisTemplate.execute(
UNLOCK_SCRIPT,
Collections.singletonList(KEY_PREFIX + name),
ID_PREFIX + Thread.currentThread().getId());`

去执行lua脚本

3.5 Redission去实现分布式锁

3.5.1 使用redis的问题:

• 1.重入问题
• 2.不可重试
• 3.超时释放(就是在获取的时候，时间过长，就会自动释放，就是同时有2个线程获取锁。)
• 4.主从一致性

Redission 提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现

3.5.2 使用redission的准备

依赖：

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

配置文件：

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

idea中使用redission:

Lock lock = redisClient.getLock("lock");
boolean success = lock.getLock();

3.5.3 在redisson的可重入锁与看门狗机制

这一部分需要去看源码，这里就不说了

• 看门狗机制就是在获取锁的时候，设置一个超时时间，如果业务未完成，就会去延长超时时间

• 可重入锁就是在获取锁的时候，可以多次获取锁，但是要去释放多次，用一个计数器来去记录

3.5.4 tryLock的一些用法

• 1.尝试获取锁，如果获取不到，就返回false

RLock lock = redissonClient.getLock("myLock");
boolean isLocked = lock.tryLock();

• 2.尝试获取锁，如果获取不到，就等待一段时间

RLock lock = redissonClient.getLock("myLock");
boolean isLocked = lock.tryLock(100, 10, TimeUnit.SECONDS);
//TODO:100是等待时间，10是超时时间（设置看门狗只有waittime,不要超时时间）

3.6 MutiLock锁

此时我们去写命令，写在主机上， 主机会将数据同步给从机，但是假设 在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。

3.7 秒杀优化

串行的方式会导致我们的程序执行的很慢，可以使用异步的方式来去执行

• 1.去判断是否有库存
• 2.去redis使用lua脚本去减少库存,完成秒杀资格判断
• 3.异步去生成订单

1、初步使用阻塞队列

   import jakarta.annotation.PostConstruct;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;private BlockingQueue<VoucherOrder>orderTasks = new LinkedBlockingQueue<>(1024*1024);
private static final ExecutorService executorService = Executors.newFixedThreadPool(10);@PostConstruct
public void init() {executorService.submit(new VoucherOrderHandle());private class VoucherOrderHandle implements Runnable {@Overridepublic void run() {while (true) {try {// 1.获取队列中的订单信息VoucherOrder voucherOrder = orderTasks.take();// 2.创建订单handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("处理订单异常", e);}}}}
}

存在的问题：

• 内存限制问题
• 数据安全问题

3.8 使用消息队列

3.8.1 基于List结构模拟消息队列

优点：

• 利用Redis存储，不受限于JVM内存上限
• 基于Redis的持久化机制，数据安全性有保证
• 可以满足消息有序性

缺点：

• 无法避免消息丢失
• 只支持单消费者

3.8.2 基于Stream的消息队列

1.基于Stream的消息队列

• xadd key * filed value *是自动生成的id
• xrange [count] [blocktime] STREAMS key ID [0是从头开始读取，$是最新的信息] count是读取的数量 blocktime是阻塞时间

STREAM类型消息队列的XREAD命令特点：

• 消息可回溯
• 一个消息可以被多个消费者读取
• 可以阻塞读取
• 有消息漏读的风险

2.基于Stream的消息队列-消费者
创建消费者组：

• XGROUP CREATE key groupname 0 [MKSTREAM] (0是从头开始读取)($是队列的最后的一个信息)
  groupName是消费者组的名字 [MKSTREAM]是创建一个新的stream

给指定的消费者组添加消费者

• XGROUP CREATECONSUMER key groupname consumername

从消费者组读取消息

• XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key …] ID [ID …]
  ID是 > 表示从最新的开始读取，0表示从头开始读取

从消费者组确认消息：

• XACK key group ID [ID ...] key是消费人 ， group 是消费组 ， Id为ID

• 查询未确认的消息： XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

• 实验的过程：

• 1.创建一个线程池来，去执行

• 2.在主线程中判断是否有库存，如果有库存，就去发送消息，执行lua脚本，去减少库存

• 3.副线程去消费消息，去生成订单（注意副线程不要使用LocalThread因为是异步的）

4.达人探店

4.1 点赞的实现

使用Zset的数据结构，因为Zset是有序的，可以根据时间来排序，分数就是时间的大小

4.2 好友关注

• 关注的好友
• 共同的关注

4.3 关注的人的动态（类型微信的朋友圈）,使用feed的流

Feed流的设计:

• 1.拉模式,：比较节约空间，但是延迟大
• 2.推模式：比较节约时间，但是空间大
• 3.混合模式：拉模式+推模式
  使用了推模式,与Feed流的滚动分页，如果不使用滚动分页，就会导致数据量的重复显现
  要注意的是顺序的问题，MyBatiesPlus的分页是根据id来排序的，所以我们要去根据时间来排序,所以要
  加入last(“order by field(id,” + strIds + “)”)来去排序
  需要去注意的是：1.时间的最小值 2.offset 3.分页的大小

    @Overridepublic Result queryBlogOfFollow(long max, Integer offset) {//1。先查询关注的人blogUser user = UserHolder.getUser();if(user == null){return Result.fail("请先登录");}String key = FEED_KEY + user.getId();//3.查询redisSet<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, 0, max, offset, 2);//2.null的判断if(typedTuples == null||typedTuples.isEmpty()){return Result.ok();}// 有Blog的id,和scoreList<Long> blogIds = new ArrayList<>(typedTuples.size());  //防止扩容long minTime = 0 ; //最小时间int oc = 1; //偏移量//3.minTime , offsetfor (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {blogIds.add(Long.valueOf(typedTuple.getValue()));  //获取博文idLong time = typedTuple.getScore().longValue();if(time == minTime) {oc++;}else {minTime = time;oc=1;}}String strIds = StrUtil.join(",", blogIds);//4.查询数据库的blog的数据List<Blog> blogList = query().in("id", blogIds).last("order by field(id," + strIds + ")").list();//5.判断是否为空if(blogList == null||blogList.isEmpty()){return Result.ok();}//5.要带上用户信息for (Blog blog : blogList) {isBlogLike(blog);}ScrollResult scrollResult =new ScrollResult();scrollResult.setList(blogList);scrollResult.setOffset(oc);scrollResult.setMinTime(minTime);return Result.ok(scrollResult);}

5.使用GEO去处理店铺的位置信息

要注意的是：StringRedisData 最好使用3.2的版本

    @Overridepublic Result queryShopType(Integer typeId, Integer current, Double x, Double y) {if(x==null || y==null){// 根据类型分页查询Page<Shop> page = query().eq("type_id", typeId).page(new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE));// 返回数据return Result.ok(page.getRecords());}int start=  (current-1)*SystemConstants.DEFAULT_PAGE_SIZE;int end =  current * SystemConstants.DEFAULT_PAGE_SIZE;// 2.根据redis的geo查询String key = SHOP_GEO_KEY + typeId;GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo().search(key,GeoReference.fromCoordinate(x, y),new Distance(5000),RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end));//查询出来了shop的Id与distanceif(results == null){return Result.ok(Collections.emptyList());}//3.获得idList<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();if (list.size() <= start) {// 没有下一页了，结束return Result.ok(Collections.emptyList());}List<Long>ids= new ArrayList<>();//存放idMap<String,Distance>distanceMap=new HashMap<>();//跳过start个,搜集List<id>与Map<id,distance>list.stream().skip(start).forEach(result->{ids.add(Long.valueOf(result.getContent().getName()));distanceMap.put(result.getContent().getName(),result.getDistance());});String StrId = StrUtil.join(",", ids);//4.需要根据id查询商铺信息(需要去数据库查询，把distance放到shop对象中)List<Shop> shopList = query().in("id", ids).last("Order by field(id," + StrId + ")").list();for (Shop shop : shopList) {shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());}//5.返回数据return Result.ok(shopList);}