• List item

Redis实战案例（黑马点评）

一、短信登录

tomcat的运行原理：

当用户发起请求时，会访问tomcat注册的端口，任何程序想要运行，都需要有一个线程对当前端口号进行监听，当用户和tomcat连接连接时，那会由监听线程创建socket连接，socket都是成对出现的，用户通过socket像互相传递数据，当tomcat端的socket接受到数据后，此时监听线程会从tomcat的线程池中取出一个线程执行用户请求，在我们的服务部署到tomcat后，线程会找到用户想要访问的工程，然后用这个线程转发到工程中的controller，service，dao中，并且访问对应的DB，在用户执行完请求后，再统一返回，再找到tomcat端的socket，再将数据写回到用户端的socket，完成请求和响应

由此我们可以得知 每个用户其实对应都是去找tomcat线程池中的一个线程来完成工作的， 使用完成后再进行回收，所以在每个用户去访问我们的工程时，我们可以使用threadlocal来做到线程隔离，每个线程操作自己的一份数据

在threadLocal中，无论是他的put方法和他的get方法， 都是先从获得当前用户的线程，然后从线程中取出线程的成员变量map，只要线程不一样，map就不一样，所以可以通过这种方式来做到线程隔离。

基于session实现

短信验证码登录

//发送验证码
@Override
public Result sendCode(String phone, HttpSession session) {if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号码格式错误");}String code = RandomUtil.randomNumbers(6);//保存到sessionsession.setAttribute("code",code);log.debug("验证成功，发从验证码：{}",code);return Result.ok();
}//登录功能
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {String phone = loginForm.getPhone();if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号码格式错误");}Object cacheCode = session.getAttribute("code");String code = loginForm.getCode();if(cacheCode == null || !cacheCode.equals(code)){return Result.fail("验证码错误");}//一致，根据手机号查询用户User user = query().eq("phone", phone).one();if(user == null){//不存在，则创建user =  createUserWithPhone(phone);}//保存用户信息到session中session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));//注意BeanUtil的包位置return Result.ok(token);
}

登录拦截器

public class LoginInterceptor implements HandlerInterceptor {@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {HttpSession session = request.getSession();// 1、判断用户是否存在User user = (User) session.getAttribute("user");if (Objects.isNull(user)){// 用户不存在，直接拦截response.setStatus(401);return false;}// 2、用户存在，则将用户信息保存到ThreadLocal中，方便后续逻辑处理// 比如：方便获取和使用用户信息，session获取用户信息是具有侵入性的ThreadLocalUtls.saveUser(user);return HandlerInterceptor.super.preHandle(request, response, handler);}
}

将自定义的拦截器添加到SpringMVC的拦截器列表中:

@Configuration
public class WebMvcConfig implements WebMvcConfigurer {@Overridepublic void addInterceptors(InterceptorRegistry registry) {// 添加登录拦截器registry.addInterceptor(new LoginInterceptor())// 设置放行请求.excludePathPatterns("/user/code","/user/login","/blog/hot","/shop/**","/shop-type/**","/upload/**","/voucher/**");}
}

基于Redis实现

session共享问题：

每个tomcat中都有一份属于自己的session,用户第一次访问一台tomcat，并且把自己的信息存放到第一台服务器的session中，但是第二次这个用户访问到了第二台tomcat，那么在第二台服务器上，肯定没有第一台服务器存放的session，所以此时整个登录拦截功能就会出现问题。早期的方案是session拷贝，但是每当任意一台服务器的session修改时，都会同步给其他的Tomcat服务器的session

但是这种方案具有两个大问题

1. 每台服务器中都有完整的一份session数据，服务器压力过大。

2. session拷贝数据时，可能会出现延迟

解决办法：我们可以使用Redis实现，因为Redis数据本身就是共享的

整体访问流程：

//发送验证码
@Override
public Result sendCode(String phone, HttpSession session) {if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号码格式错误");}String code = RandomUtil.randomNumbers(6);stringRedisTemplate.opsForValue().set("login:code" + phone,code,2, TimeUnit.MINUTES);log.debug("验证成功，发从验证码：{}",code);return Result.ok();
}//登录
public Result login(LoginFormDTO loginForm, HttpSession session) {String phone = loginForm.getPhone();if (RegexUtils.isPhoneInvalid(phone)) {return Result.fail("手机号码格式错误");}//从redis中获取验证码并校验String cacheCode = stringRedisTemplate.opsForValue().get("login:code" + phone);String code = loginForm.getCode();if(cacheCode == null || !cacheCode.equals(code)){return Result.fail("验证码错误");}//一致，根据手机号查询用户User user = query().eq("phone", phone).one();if(user == null){//不存在，则创建user =  createUserWithPhone(phone);}//保存用户信息到redisString token = UUID.randomUUID().toString();UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);//！！！因为使用的是stringRedisTemplate，map的值必须为String类型Map<String, Object> userMap = BeanUtil.beanToMap(userDTO,new HashMap<>(), CopyOptions.create().setIgnoreNullValue(true)     //忽略空的值，在转换过程中不包括空值的字段.setFieldValueEditor((fieldName,fieldValue) -> fieldValue.toString()));  //修改字段值：将字段值转为字符串stringRedisTemplate.opsForHash().putAll("login:token" + token,userMap);stringRedisTemplate.expire("login:token" + token,30,TimeUnit.MINUTES);return Result.ok(token);
}

token刷新问题

将token刷新等功能放在一个拦截器中确实也可以实现token刷新，但当用户访问了不需要拦截的接口拦截器不会生效，token也就无法刷新。所以需要加一个拦截器拦截所有路径

刷新token拦截器

//拦截一切路径的请求，并刷新token的有效期，即使用户不存在也放行
public class RefreshTokenInterceptor implements HandlerInterceptor {/**当前类RefreshTokenInterceptor没有交给spring管理，不能使用依赖注入（@Autowired）
!!!只能用构造方法方式注入属性，将来由使用者构造对象传入stringRedisTemplate实例**/
private StringRedisTemplate stringRedisTemplate;public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;
}@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {//获取请求头中的token    authorization为前段token名字String token = request.getHeader("authorization");if (StrUtil.isBlank(token)) {return true;}//基于token获取redis中的用户    get是获取单个键值对；entries获得所有键值对Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries("login:token" + token);//判断用户是否存在if(userMap.isEmpty()){return true;}//将Map转为DTO对象UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);UserHolder.saveUser(userDTO);stringRedisTemplate.expire("login:token" + token,30, 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(401);return false;}// 用户存在，直接放行return true;}
}

拦截器配置

@Configuration
public class MvcConfig implements WebMvcConfigurer {@Resourceprivate StringRedisTemplate stringRedisTemplate;@Overridepublic void addInterceptors(InterceptorRegistry registry) {// 登录拦截器registry.addInterceptor(new LoginInterceptor()).excludePathPatterns("/shop/**","/voucher/**","/shop-type/**","/upload/**","/blog/hot","/user/code","/user/login").order(1);// token刷新的拦截器registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);}}

二、商户缓存

    /*** 根据id查询商铺数据*/@Overridepublic Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;// 1、从Redis中查询店铺数据String shopJson = stringRedisTemplate.opsForValue().get(key);Shop shop = null;if (StrUtil.isNotBlank(shopJson)) {shop = JSONUtil.toBean(shopJson, Shop.class);return Result.ok(shop);}// 2、 缓存未命中，从数据库中查询店铺数据shop = this.getById(id);if (Objects.isNull(shop)) {return Result.fail("店铺不存在");}stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));return Result.ok(shop);}

缓存更新问题

常见的缓存更新策略：

• 内存淘汰：利用Redis的内存淘汰机制实现缓存更新，Redis的内存淘汰机制是当Redis发现内存不足时，会根据一定的策略自动淘汰部分数据

• 超时剔除：手动给缓存数据添加TTL，到期后Redis自动删除缓存

• 主动更新：手动编码实现缓存更新，在修改数据库的同时更新缓存

  • 双写方案：人工编码方式，缓存调用者在更新完数据库后再去更新缓存。使用困难，灵活度高。

    • 读取（Read）：当需要读取数据时，首先检查缓存是否存在该数据。如果缓存中存在，直接返回缓存中的数据。如果缓存中不存在，则从底层数据存储（如数据库）中获取数据，并将数据存储到缓存中，以便以后的读取操作可以更快地访问该数据。

    • 写入（Write）：当进行数据写入操作时，首先更新底层数据存储中的数据。然后，根据具体情况，可以选择直接更新缓存中的数据（使缓存与底层数据存储保持同步），或者是简单地将缓存中与修改数据相关的条目标记为无效状态（缓存失效），以便下一次读取时重新加载最新数据

      使用双写方案需要考虑以下几个问题：

      1.是使用更新缓存模式还是使用删除缓存模式？

      • 更新缓存模式：每次更新数据库都更新缓存，无效写操作较多（不推荐使用）
      • 删除缓存模式：更新数据时更新数据库并删除缓存，查询时更新缓存，无效写操作较少（推荐使用）

      2.选择使用删除缓存模式，那么是先操作缓存还是先操作数据库？

      • 先操作缓存：先删缓存，再更新数据库
      • 先操作数据库：先更新数据库，再删缓存（推荐使用）

      3.选择先更新数据库，再删除缓存。那么如何保证缓存与数据库的操作的原子性（同时成功或失败）？

      • 对于单体系统1，将缓存与数据库操作放在同一个事务中（当前项目就是一个单体项目，所以选择这种方式）
      • 对于分布式系统2，利用TCC（Try-Confirm-Cancel）等分布式事务方案

  • 读写穿透方案（Read/Write Through Pattern）：将读取和写入操作首先在缓存中执行，然后再传播到数据存储

  • 读取穿透（Read Through）：当进行读取请求时，首先检查缓存。如果所请求的数据在缓存中找到，直接返回数据。如果缓存中没有找到数据，则将请求转发给数据存储以获取数据。获取到的数据随后存储在缓存中，然后返回给调用者。

  • 写入穿透（Write Through）：当进行写入请求时，首先将数据写入缓存。缓存立即将写操作传播到数据存储，确保缓存和数据存储之间的数据保持一致。这样保证了后续的读取请求从缓存中返回更新后的数据。

  • 写回方案（Write Behind Caching Pattern）：调用者只操作缓存，其他线程去异步处理数据库，实现最终一致

    • 读取（Read）：先检查缓存中是否存在数据，如果不存在，则从底层数据存储中获取数据，并将数据存储到缓存中。
    • 写入（Write）：先更新底层数据存储，然后将待写入的数据放入一个缓存队列中。在适当的时机，通过批量操作或异步处理，将缓存队列中的数据写入底层数据存储

主动更新策略中三种方案的比较：

双写方案 和 读写穿透方案 在写入数据时都会直接更新缓存，以保持缓存和底层数据存储的一致性。而 写回方案 延迟了缓存的更新操作，将数据先放入缓存队列，然后再进行批量或异步写入。
读写穿透方案 和 写回方案 相比，写回方案 具有更高的写入性能，因为它通过批量和异步操作减少了频繁的写入操作。但是 写回方案 带来了数据一致性的考虑，需要确保缓存和底层数据存储在某个时间点上保持一致，而 读写穿透方案 将数据库和缓存整合为一个服务，由服务来维护缓存与数据库的一致性，调用者无需关心数据一致性问题，降低了系统的可维护性，但是实现困难
主动更新策略中三种方案的应用场景：

双写方案 较适用于读多写少的场景，数据的一致性由应用程序主动管理
读写穿透方案 适用于数据实时性要求较高、对一致性要求严格的场景
写回方案 适用于追求写入性能的场景，对数据的实时性要求相对较低、可靠性也相对低
更新策略的应用场景：

对于低一致性需求，可以使用内存淘汰机制。例如店铺类型数据的查询缓存
对于高一致性需求，可以采用主动更新策略，并以超时剔除作为兜底方案。例如店铺详情数据查询的缓存
缓存主动更新策略的实现

最终实现逻辑：根据id查询店铺时，如果缓存未命中，则查询数据库，将数据库结果写入缓存，并设置超时时间。根据id修改店铺时，先修改数据库，再删除缓存

 public Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;String shopJson = stringRedisTemplate.opsForValue().get(key);Shop shop = null;// 判断缓存是否命中if (StrUtil.isNotBlank(shopJson)) {shop = JSONUtil.toBean(shopJson, Shop.class);return Result.ok(shop);}// 缓存未命中，从数据库中查询店铺数据shop = this.getById(id);// 判断数据库是否存在店铺数据if (Objects.isNull(shop)) {return Result.fail("店铺不存在");}// 数据库中存在，重建缓存，并返回店铺数据stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);return Result.ok(shop);
}/*** 更新商铺数据（更新时，更新数据库，删除缓存）*/
@Transactional
@Override
public Result updateShop(Shop shop) {// 参数校验, 略// 1、更新数据库中的店铺数据boolean f = this.updateById(shop);if (!f){// 缓存更新失败，抛出异常，事务回滚throw new RuntimeException("数据库更新失败");}// 2、删除缓存f = stringRedisTemplate.delete(CACHE_SHOP_KEY + shop.getId());if (!f){// 缓存删除失败，抛出异常，事务回滚throw new RuntimeException("缓存删除失败");}return Result.ok();
}

缓存穿透

缓存穿透 ：缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。

解决办法：

• 缓存空对象：实现简单，维护方便，但是会消耗内存可能造成短期的不一致
• 布隆过滤：内存占用较少，但实现复杂存在误判可能

方案一：

/*** 根据id查询商铺数据*/
@Override
public Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;String shopJson = stringRedisTemplate.opsForValue().get(key);Shop shop = null;// 判断缓存是否命中if (StrUtil.isNotBlank(shopJson)) {shop = JSONUtil.toBean(shopJson, Shop.class);return Result.ok(shop);}// 缓存未命中，判断缓存中查询的数据是否是空字符串(isNotBlank把null和空字符串给排除了)if (Objects.nonNull(shopJson)){// 当前数据是空字符串（说明该数据是之前缓存的空对象），直接返回失败信息return Result.fail("店铺不存在");}//  当前数据是null，则从数据库中查询店铺数据shop = this.getById(id);// 判断数据库是否存在店铺数据if (Objects.isNull(shop)) {//  数据库中不存在，缓存空对象（解决缓存穿透），返回失败信息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);
}

缓存穿透的解决方案有哪些：缓存null值、布隆过滤、增强id的复杂度，避免被猜测id规律、做好数据的基础格式校验、加强用户权限校验、做好热点参数的限流。

缓存雪崩

缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

解决方案：

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

缓存穿透

缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

常见的解决方案有两种：

• 互斥锁
• 逻辑过期

互斥锁方案：

操作锁代码

核心思路就是利用redis的setnx方法来表示获取锁，该方法含义是redis中如果没有这个key，则插入成功，返回1，在stringRedisTemplate中返回true， 如果有这个key则插入失败，则返回0，在stringRedisTemplate返回false，我们可以通过true，或者是false，来表示是否有线程成功插入key，成功插入的key的线程我们认为他就是获得到锁的线程。

/获取锁函数      setIfAbsent为setnx
private boolean tryLock(String key){//为防止程序异常，锁获取成功后迟迟没有释放，导致后续无发获取锁，给锁增加有效期Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);//此处返回的是0或1，spring帮我们转换成啦对象结果，我们需要转换为基本类型，直接返回flag会自动拆箱可能导致空指针异常//对象结果：flag为Boolean，而不是具体的boolean，而Boolean的值可以为null，此处将false和null都处理为nullreturn BooleanUtil.isTrue(flag);
}//释放锁函数
private void unlock(String key){stringRedisTemplate.delete(key);
}

业务代码：

//缓存击穿设置互斥锁的解决方案 （利用redis  String类型的setnx的特性判断是否可以获取锁）
public Shop queryWithMutex(Long id){String key = CACHE_SHOP_KEY + id;//从redis查询缓存String shopJson = stringRedisTemplate.opsForValue().get(key);//判断是否命中if(StrUtil.isNotBlank(shopJson)){//命中直接返回Shop shop = JSONUtil.toBean(shopJson,Shop.class);return shop;}//判断命中的是否空值if(shopJson != null){return null;}String lockKey = "lock:shop:" + id;Shop shop = null;try {boolean islock = tryLock(lockKey);if(!islock){Thread.sleep(50);return queryWithMutex(id);}//获取锁成功，根据id查询数据库shop = getById(id);//mysql中也不存在，返回错误if(shop == null){//写入空值stringRedisTemplate.opsForValue().set(key,"",2,TimeUnit.MINUTES);return null;}//存在，写入redisstringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),10, TimeUnit.MINUTES);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {unlock(lockKey);}return shop;
}

逻辑过期方案：

当用户开始查询redis时，判断是否命中，如果没有命中则直接返回空数据，不查询数据库，而一旦命中后，将value取出，判断value中的过期时间是否满足，如果没有过期，则直接返回redis中的数据，如果过期，则在开启独立线程后直接返回之前的数据，独立线程去重构数据，重构完成后释放互斥锁。

//缓存击穿设置逻辑过期的解决方案
public Shop queryWithLogicalExpire(Long id){String key = CACHE_SHOP_KEY + id;//从redis查询缓存String shopJson = stringRedisTemplate.opsForValue().get(key);//判断是否命中if(StrUtil.isBlank(shopJson)){return null;}//存在，把json反序列化为对象RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);JSONObject data = (JSONObject) redisData.getData();//因为使用的是stringRedisTemplate，这里获得的对象为JSONObjectShop shop = JSONUtil.toBean(data,Shop.class);LocalDateTime expireTime = redisData.getExpireTime();//判断是否过期if(expireTime.isAfter(LocalDateTime.now())){//未过期return shop;}//已过期，缓存重建//重建缓存//获取互斥锁String lockey = "lock:shop:" + id;boolean isLock = tryLock(lockey);//判断是否获取成功if(isLock){//TODO 注意：获取锁成功再次判断是否过期做doubleCheck，未过期则无需重建（获取的所可能是上一个线程刚刚释放的锁）CACHE_REBUID_EXECUTOR.submit(()->{try {//成功，开启独立线程，实现重建缓存this.saveShop2Redis(id,20L);} catch (Exception e) {throw new RuntimeException(e);} finally {unlock(lockey);}});}//返回信息return shop;
}//设置逻辑过期时间函数：redisData为Shop+expiredSeconds
private void saveShop2Redis(Long id,Long expiredSeconds) {Shop shop = getById(id);RedisData redisData = new RedisData();redisData.setData(shop);redisData.setExpireTime(LocalDateTime.now().plusSeconds(expiredSeconds));stringRedisTemplate.opsForValue().set("lock:shop:" + id,JSONUtil.toJsonStr(redisData));
}

三、优惠券秒杀

全局唯一id

使用数据库自增ID存在的问题：

• id的规律性太明显
• 受单表数据量的限制

场景分析：如果我们的id具有太明显的规则，用户或者说商业对手很容易猜测出来我们的一些敏感信息，比如商城在一天时间内，卖出了多少单，这明显不合适。

场景分析二：随着我们商城规模越来越大，mysql的单表的容量不宜超过500W，数据量过大之后，我们要进行拆库拆表，但拆分表了之后，他们从逻辑上讲他们是同一张表，所以他们的id是不能一样的， 于是乎我们需要保证id的唯一性。

全局ID生成器，是一种在分布式系统下用来生成全局唯一ID的工具，一般要满足特性：唯一性、高性能、高可用、递增行、安全性

private static final long BEGIN_TIMESTAMP = 1640995200L;  //自定义开始值单位：sprivate static final int COUNT_BITS = 32;private StringRedisTemplate stringRedisTemplate;public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {this.stringRedisTemplate = stringRedisTemplate;}public long nextId(String keyPrefix){//生成时间戳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.自增长 ||  每一天下的单为一个keylong count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);// 3.拼接并返回return timeStamp << COUNT_BITS | count;}

消息队列

基于List结构

Redis的list数据结构是一个双向链表，很容易模拟出队列效果。

可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。需要注意，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

优缺点：

• 优点：利用Redis存储，不受限于JVM内存上限、基于Redis的持久化机制，数据安全性有保证可以满足消息有序性
• 缺点：无法避免消息丢失、只支持单消费者

基于PubSub

PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

SUBSCRIBE channel [channel] ：订阅一个或多个频道
PUBLISH channel msg ：向一个频道发送消息
PSUBSCRIBE pattern[pattern] ：订阅与pattern格式匹配的所有频道

优缺点：

• 优点：采用发布订阅模型，支持多生产、多消费

• 缺点：不支持数据持久化、无法避免消息丢失、消息堆积有上限，超出时数据丢失

基于Stream

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

• 发送消息xadd:

  

• 读取消息：

  

在业务代码可以使用循环使用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下

while(true){//尝试读取队列里面的消息，最多阻塞2秒Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $")if(msg == null){continue;}//处理消息handleMessage(msg)
}

注意：当我们指定起始ID为$时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题

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

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

消费者组

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

• 消息分流：队列中的消息会分流给组内不同的消费者，从而加快消息处理速度
• 消息标识：消费者组会维护一个标识，哪怕消费者组宕机重启，还会从标示之后读取消息确保每条消息被消费
• 消息确认：消费者获取到消息后，消息会处于pending状态，并存日pending-list，当处理完后需要通过XACK确认消息后才从pending-list去除

创建消费者组：

XGROUP CREATE key groupName ID [MKSTREAM]

key：队列名称
groupName：消费者组名称
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命令特点：

• 消息可回溯
• 可以多消费者争抢消息，加快消费速度
• 可以阻塞读取
• 没有消息漏读的风险
• 有消息确认机制，保证消息至少被消费一次

最终版本：Lua脚本+Redission分布式锁+Stream消息队列

-- 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-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then-- 3.2.库存不足，返回1return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then-- 3.3.存在，说明是重复下单，返回2return 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

核心业务代码：

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;//加载lua脚本static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();  //线程池private IVoucherOrderService proxy;@PostConstruct //再类初始化后先开启异步添加订单服务，准备private  void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}/*** 订单秒杀业务最终方案* @param voucherId* @return*/@Overridepublic Result seckillVoucher(Long voucherId) {Long userId = UserHolder.getUser().getId();long orderId = redisIdWorker.nextId("order");// 1.执行lua脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));int r = result.intValue();// 2.判断结果是否为0if (r != 0) {// 2.1.不为0 ，代表没有购买资格return Result.fail(r == 1 ? "库存不足" : "不能重复下单");}// 3.返回订单idreturn Result.ok(orderId);}private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while (true) {try {// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));// 2.判断订单信息是否为空if (list == null || list.isEmpty()) {// 如果为null，说明没有消息，继续下一次循环continue;}// 解析数据MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch (Exception e) {log.error("处理订单异常", e);//处理异常消息handlePendingList();}}}private void handlePendingList() {while (true) {try {// 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1),StreamOffset.create("stream.orders", ReadOffset.from("0")));// 2.判断订单信息是否为空if (list == null || list.isEmpty()) {// 如果为null，说明没有异常消息，结束循环break;}// 解析数据MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch (Exception e) {log.error("处理pendding订单异常", e);try{Thread.sleep(20);}catch(Exception e){e.printStackTrace();}}}}
}}

创建订单：

/将订单信息写入数据库@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();if (count > 0) {return;}// 6.扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0) // where id = ? and stock > 0.update();if (!success) {// 扣减失败return;}save(voucherOrder);}
}

四、附近商户

GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。常见的命令有：

• GEOADD：添加一个地理空间信息，包含：经度（longitude）、纬度（latitude）、值（member）
• GEODIST：计算指定的两个点之间的距离并返回
• GEOHASH：将指定member的坐标转为hash字符串形式并返回
• GEOPOS：返回指定member的坐标
• GEORADIUS：指定圆心、半径，找到该圆内包含的所有member，并按照与圆心之间的距离排序后返回。6.以后已废弃
• GEOSEARCH：在指定范围内搜索member，并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。6.2.新功能
• GEOSEARCHSTORE：与GEOSEARCH功能一致，不过可以把结果存储到一个指定的key。 6.2.新功能

@Overridepublic Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {// 1.判断是否需要根据坐标查询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());}// 2.计算分页参数int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;int end = current * SystemConstants.DEFAULT_PAGE_SIZE;// 3.查询redis、按照距离排序、分页。结果：shopId、distanceString key = SHOP_GEO_KEY + typeId;GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo() // GEOSEARCH key BYLONLAT x y BYRADIUS 10 WITHDISTANCE.search(key,GeoReference.fromCoordinate(x, y),new Distance(5000),RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end));// 4.解析出idif (results == null) {return Result.ok(Collections.emptyList());}List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();if (list.size() <= from) {// 没有下一页了，结束return Result.ok(Collections.emptyList());}// 4.1.截取 from ~ end的部分List<Long> ids = new ArrayList<>(list.size());Map<String, Distance> distanceMap = new HashMap<>(list.size());list.stream().skip(from).forEach(result -> {// 4.2.获取店铺idString shopIdStr = result.getContent().getName();ids.add(Long.valueOf(shopIdStr));// 4.3.获取距离Distance distance = result.getDistance();distanceMap.put(shopIdStr, distance);});// 5.根据id查询ShopString idStr = StrUtil.join(",", ids);List<Shop> shops = query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list();for (Shop shop : shops) {shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());}// 6.返回return Result.ok(shops);}

五、用户签到

用户签到完全可以单独创建一个mysql表存储用户每天签到信息，不过我们可以使用BitMap简化签到记录方式。

我们按月来统计用户签到信息，签到记录为1，未签到则记录为0。把每一个bit位对应当月的每一天，形成了映射关系。用0和1标示业务状态，这种思路就称为位图（BitMap）。这样我们就用极小的空间，来实现了大量数据的表示

Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是 2^32个bit位。

BitMap的操作命令有：

• SETBIT：向指定位置（offset）存入一个0或1
• GETBIT ：获取指定位置（offset）的bit值
• BITCOUNT ：统计BitMap中值为1的bit位的数量
• BITFIELD ：操作（查询、修改、自增）BitMap中bit数组中的指定位置（offset）的值
• BITFIELD_RO ：获取BitMap中bit数组，并以十进制形式返回
• BITOP ：将多个BitMap的结果做位运算（与 、或、异或）
• BITPOS ：查找bit数组中指定范围内第一个0或1出现的位置

签到：

@Override
public Result sign() {// 1.获取当前登录用户Long userId = UserHolder.getUser().getId();// 2.获取日期LocalDateTime now = LocalDateTime.now();// 3.拼接keyString keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));String key = USER_SIGN_KEY + userId + keySuffix;// 4.获取今天是本月的第几天int dayOfMonth = now.getDayOfMonth();// 5.写入Redis SETBIT key offset 1stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true);return Result.ok();
}

签到统计：

@Override
public Result signCount() {// 1.获取当前登录用户Long userId = UserHolder.getUser().getId();// 2.获取日期LocalDateTime now = LocalDateTime.now();// 3.拼接keyString keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));String key = USER_SIGN_KEY + userId + keySuffix;// 4.获取今天是本月的第几天int dayOfMonth = now.getDayOfMonth();// 5.获取本月截止今天为止的所有的签到记录，返回的是一个十进制的数字 BITFIELD sign:5:202203 GET u14 0List<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);}// 6.循环遍历int count = 0;while (true) {// 6.1.让这个数字与1做与运算，得到数字的最后一个bit位  // 判断这个bit位是否为0if ((num & 1) == 0) {// 如果为0，说明未签到，结束break;}else {// 如果不为0，说明已签到，计数器+1count++;}// 把数字右移一位，抛弃最后一个bit位，继续下一个bit位num >>>= 1;}return Result.ok(count);
}

六、好友关注

@Override
public Result follow(Long followUserId, Boolean isFollow) {// 1.获取登录用户Long userId = UserHolder.getUser().getId();String key = "follows:" + userId;// 1.判断到底是关注还是取关if (isFollow) {// 2.关注，新增数据Follow follow = new Follow();follow.setUserId(userId);follow.setFollowUserId(followUserId);boolean isSuccess = save(follow);if (isSuccess) {// 把关注用户的id，放入redis的set集合 sadd userId followerUserIdstringRedisTemplate.opsForSet().add(key, followUserId.toString());}} else {// 3.取关，删除 delete from tb_follow where user_id = ? and follow_user_id = ?boolean isSuccess = remove(new QueryWrapper<Follow>().eq("user_id", userId).eq("follow_user_id", followUserId));if (isSuccess) {// 把关注用户的id从Redis集合中移除stringRedisTemplate.opsForSet().remove(key, followUserId.toString());}}return Result.ok();
}

好友关注-共同关注

@Override
public Result followCommons(Long id) {// 1.获取当前用户Long userId = UserHolder.getUser().getId();String key = "follows:" + userId;// 2.求交集String key2 = "follows:" + id;Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key, key2);if (intersect == null || intersect.isEmpty()) {// 无交集return Result.ok(Collections.emptyList());}// 3.解析id集合List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());// 4.查询用户List<UserDTO> users = userService.listByIds(ids).stream().map(user -> BeanUtil.copyProperties(user, UserDTO.class)).collect(Collectors.toList());return Result.ok(users);
}

七、消息推送

Fead流：当我们关注了用户后，这个用户发了动态，把这些数据推送给用户，这种消息推送的方式叫做Feed流，关注推送也叫做Feed流，直译为投喂。

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

• Timeline：不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈。这种方式信息全面，不会有缺失。并且实现也相对简单，但信息噪音较多，用户不一定感兴趣，内容获取效率低
• 智能排序：利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户这种方式投喂用户感兴趣信息，用户粘度很高，容易沉迷，但如果算法不精准，可能起到反作用

我们本次针对好友的操作，采用的就是Timeline的方式，只需要拿到我们关注用户的信息，然后按照时间排序即可

，因此采用Timeline的模式。该模式的实现方案有三种：

• 拉模式：当张三和李四和王五发了消息后，都会保存在自己的邮箱中，假设赵六要读取信息，那么他会从读取他自己的收件箱，此时系统会从他关注的人群中，把他关注人的信息全部都进行拉取，然后在进行排序
• 推模式：当张三写了一个内容，此时会主动的把张三写的内容发送到他的粉丝收件箱中去，假设此时李四再来读取，就不用再去临时拉取了
• 推拉结合：一个折中的方案，站在发件人这一段，如果是个普通的人，那么我们采用写扩散的方式，直接把数据写入到他的粉丝中去，因为普通的人他的粉丝关注量比较小，所以这样做没有压力，如果是大V，那么他是直接将数据先写入到一份到发件箱里边去，然后再直接写一份到活跃粉丝收件箱里边去，现在站在收件人这端来看，如果是活跃粉丝，那么大V和普通的人发的都会直接写入到自己收件箱里边来，而如果是普通的粉丝，由于他们上线不是很频繁，所以等他们上线时，再从发件箱里边去拉信息。

博客推送

@Override
public Result saveBlog(Blog blog) {// 1.获取登录用户UserDTO user = UserHolder.getUser();blog.setUserId(user.getId());// 2.保存探店笔记boolean isSuccess = save(blog);if(!isSuccess){return Result.fail("新增笔记失败!");}// 3.查询笔记作者的所有粉丝 select * from tb_follow where follow_user_id = ?List<Follow> follows = followService.query().eq("follow_user_id", user.getId()).list();// 4.推送笔记id给所有粉丝for (Follow follow : follows) {// 4.1.获取粉丝idLong userId = follow.getUserId();// 4.2.推送String key = FEED_KEY + userId;stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());}// 5.返回idreturn Result.ok(blog.getId());
}

分页查询收邮箱

我们需要记录每次操作的最后一条，然后从这个位置开始去读取数据

举个例子：我们从t1时刻开始，拿第一页数据，拿到了10~6，然后记录下当前最后一次拿取的记录，就是6，t2时刻发布了新的记录，此时这个11放到最顶上，但是不会影响我们之前记录的6，此时t3时刻来拿第二页，第二页这个时候拿数据，还是从6后一点的5去拿，就拿到了5-1的记录。我们这个地方可以采用sortedSet来做，可以进行范围查询，并且还可以记录当前获取数据时间戳最小值，就可以实现滚动分页了

一、定义出来具体的返回值实体类

@Data
public class ScrollResult {private List<?> list;private Long minTime;private Integer offset;
}

BlogController

注意：RequestParam 表示接受url地址栏传参的注解，当方法上参数的名称和url地址栏不相同时，可以通过RequestParam 来进行指定

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

BlogServiceImpl

@Override
public 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; // 2int os = 1; // 2for (ZSetOperations.TypedTuple<String> tuple : typedTuples) { // 5 4 4 2 2// 4.1.获取idids.add(Long.valueOf(tuple.getValue()));// 4.2.获取分数(时间戳）long time = tuple.getScore().longValue();if(time == minTime){os++;}else{minTime = time;os = 1;}}os = minTime == max ? os : os + offset;// 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) {// 5.1.查询blog有关的用户queryBlogUser(blog);// 5.2.查询blog是否被点赞isBlogLiked(blog);}// 6.封装并返回ScrollResult r = new ScrollResult();r.setList(blogs);r.setOffset(os);r.setMinTime(minTime);return Result.ok(r);
}

八、点赞排行

点赞功能，需判断是否点赞过，未点赞过则点赞数+1，已点赞过则点赞数-1，且需要排序。综合考虑使用SortedSet实现

1. 对于SortedList我们可以使用ZSCORE方法判断用户是否存在
2. Set集合没有提供范围查询，无法获排行榜前几名的数据，SortedList可以使用ZRANGE方法实现范围查询

//点赞
@Override
public Result likeBlog(Long id) {// 1.获取登录用户Long userId = UserHolder.getUser().getId();// 2.判断当前登录用户是否已经点赞String key = BLOG_LIKED_KEY + id;Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());if (score == null) {// 3.如果未点赞，可以点赞// 3.1.数据库点赞数 + 1boolean isSuccess = update().setSql("liked = liked + 1").eq("id", id).update();// 3.2.保存用户到Redis的sortedSet集合  zadd key value scoreif (isSuccess) {stringRedisTemplate.opsForZSet().add(key, userId.toString(), System.currentTimeMillis());}} else {// 4.如果已点赞，取消点赞// 4.1.数据库点赞数 -1boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update();// 4.2.把用户从Redis的sortedSet集合移除if (isSuccess) {stringRedisTemplate.opsForZSet().remove(key, userId.toString());}}return Result.ok();
}private void isBlogLiked(Blog blog) {// 1.获取登录用户UserDTO user = UserHolder.getUser();if (user == null) {// 用户未登录，无需查询是否点赞return;}Long userId = user.getId();// 2.判断当前登录用户是否已经点赞String key = "blog:liked:" + blog.getId();Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());blog.setIsLike(score != null);
}//点赞排行查询
@Override
public Result queryBlogLikes(Long id) {String key = BLOG_LIKED_KEY + id;// 1.查询top5的点赞用户 zrange key 0 4Set<String> top5 = stringRedisTemplate.opsForZSet().range(key, 0, 4);if (top5 == null || top5.isEmpty()) {return Result.ok(Collections.emptyList());}// 2.解析出其中的用户idList<Long> ids = top5.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());// 4.返回return Result.ok(userDTOS);
}

九、UV统计

首先我们搞懂两个概念：

• UV：全称Unique Visitor，也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次。
• PV：全称Page View，也叫页面访问量或点击量，用户每访问网站的一个页面，记录1次PV，用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。

通常来说UV会比PV大很多，所以衡量同一个网站的访问量，我们需要综合考虑很多因素，所以我们只是单纯的把这两个值作为一个参考值

UV统计在服务端做会比较麻烦，因为要判断该用户是否已经统计过了，需要将统计过的用户信息保存。但是如果每个访问的用户都保存到Redis中，数据量会非常恐怖，那怎么处理呢？

Hyperloglog(HLL)是从Loglog算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。相关算法原理大家可以参考：https://juejin.cn/post/6844903785744056333#heading-0
Redis中的HLL是基于string结构实现的，单个HLL的内存永远小于16kb，内存占用低的令人发指！作为代价，其测量结果是概率性的，有小于0.81％的误差。不过对于UV统计来说，这完全可以忽略。

    /*** 测试 HyperLogLog 实现 UV 统计的误差*/@Testpublic void testHyperLogLog() {String[] values = new String[1000];// 批量保存100w条用户记录，每一批1个记录int j = 0;for (int i = 0; i < 1000000; i++) {j = i % 1000;values[j] = "user_" + i;if (j == 999) {// 发送到RedisstringRedisTemplate.opsForHyperLogLog().add("hl2", values);}}// 统计数量Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2");System.out.println("count = " + count);}

ids).last(“ORDER BY FIELD(id,” + idStr + “)”).list()
.stream()
.map(user -> BeanUtil.copyProperties(user, UserDTO.class))
.collect(Collectors.toList());
// 4.返回
return Result.ok(userDTOS);
}

#### ## 九、UV统计首先我们搞懂两个概念：* UV：全称Unique Visitor，也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次。
* PV：全称Page View，也叫页面访问量或点击量，用户每访问网站的一个页面，记录1次PV，用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。通常来说UV会比PV大很多，所以衡量同一个网站的访问量，我们需要综合考虑很多因素，所以我们只是单纯的把这两个值作为一个参考值UV统计在服务端做会比较麻烦，因为要判断该用户是否已经统计过了，需要将统计过的用户信息保存。但是如果每个访问的用户都保存到Redis中，数据量会非常恐怖，那怎么处理呢？Hyperloglog(HLL)是从Loglog算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。相关算法原理大家可以参考：https://juejin.cn/post/6844903785744056333#heading-0
Redis中的HLL是基于string结构实现的，单个HLL的内存**永远小于16kb**，**内存占用低**的令人发指！作为代价，其测量结果是概率性的，**有小于0.81％的误差**。不过对于UV统计来说，这完全可以忽略。```java/*** 测试 HyperLogLog 实现 UV 统计的误差*/@Testpublic void testHyperLogLog() {String[] values = new String[1000];// 批量保存100w条用户记录，每一批1个记录int j = 0;for (int i = 0; i < 1000000; i++) {j = i % 1000;values[j] = "user_" + i;if (j == 999) {// 发送到RedisstringRedisTemplate.opsForHyperLogLog().add("hl2", values);}}// 统计数量Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2");System.out.println("count = " + count);}