黑马点评项目实战：Redis 缓存策略与 RabbitMQ 异步秒杀

什么是缓存雪崩，怎么解决？

定义： 缓存雪崩是缓存中大量 key 失效后当高并发到来时导致大量请求到数据库，瞬间耗尽数据库资源，导致数据库无法使用。 造成缓存雪崩问题的原因是是大量 key 拥有了相同的过期时间，比如对课程信息设置缓存过期时间为 10 分钟，在大量请求同时查询大量的课程信息时，此时就会有大量的课程存在相同的过期时间，一旦失效将同时失效，造成雪崩问题。

解决方法：

1. 使用同步锁控制查询数据库的线程

   synchronized(obj){ //查询数据库 //存入缓存 }
   

2. 对同一类型信息的 key 设置不同的过期时间 通常对一类信息的 key 设置的过期时间是相同的，这里可以在原有固定时间的基础上加上一个随机时间使它们的过期时间都不相同。

   //设置过期时间 300 秒 redisTemplate.opsForValue().set("course:" + courseId, JSON.toJSONString(coursePublish), 300 + newRandom().nextInt(100), TimeUnit.SECONDS);
   

3. 缓存预热 不用等到请求到来再去查询数据库存入缓存，可以提前将数据存入缓存。使用缓存预热机制通常有专门的后台程序去将数据库的数据同步到缓存。

什么是缓存击穿，怎么解决？

定义： 缓存击穿是指大量并发访问同一个热点数据，当热点数据失效后同时去请求数据库，瞬间耗尽数据库资源，导致数据库无法使用。比如某手机新品发布，当缓存失效时有大量并发到来导致同时去访问数据库。

解决方法：

1. 基于互斥锁解决 互斥锁（时间换空间） 优点：内存占用小，一致性高，实现简单 缺点：性能较低，容易出现死锁 这里使用 Redis 中的 setnx 指令实现互斥锁，只有当值不存在时才能进行 set 操作 锁的有效期更具体业务有关，需要灵活变动，一般锁的有效期是业务处理时长 10~20 倍 线程获取锁后，还需要查询缓存（也就是所谓的双检），这样才能够真正有效保障缓存不被击穿
2. 基于逻辑过期方式 逻辑过期（空间换时间） 优点：性能高 缺点：内存占用较大，容易出现脏读 ·注意：逻辑过期一定要先进行数据预热，将我们热点数据加载到缓存中

适用场景 商品详情页、排行榜等热点数据场景。 数据更新频率低，但访问量大的场景。

总结：两者相比较，互斥锁更加易于实现，但是容易发生死锁，且锁导致并行变成串行，导致系统性能下降，逻辑过期实现起来相较复杂，且需要耗费额外的内存，但是通过开启子线程重建缓存，使原来的同步阻塞变成异步，提高系统的响应速度，但是容易出现脏读

为什么重建子线程，作用是什么？

开启子线程重建缓存的作用在于提高系统的响应速度，避免因缓存击穿导致的数据库压力过大，同时保障系统在高并发场景下的稳定性，但开启子线程重建缓存可能引入数据不一致（脏读）问题。

具体原因：

1. 提高系统响应速度 同步阻塞的缺点：在缓存失效时，传统方案通常会同步查询数据库更新缓存，这会导致用户请求被阻塞，特别是在高并发环境下可能出现大量线程等待，影响系统响应性能。 子线程重建缓存的优势：主线程只需返回缓存中的旧数据，避免阻塞用户请求。重建缓存的任务交由后台线程执行，提高用户体验。
2. 减少数据库压力 缓存击穿问题：当热点数据过期时，多个线程同时访问数据库，可能导致数据库压力骤增，甚至崩溃。 子线程异步重建缓存：将数据库查询集中到一个后台线程中执行，避免多个线程同时查询数据库。即便在缓存击穿的情况下，也不会对数据库造成过大的负载。
3. 提高系统吞吐量 同步更新的瓶颈：如果所有线程都等待缓存更新完成，系统吞吐量会因阻塞而降低。 异步重建的优化：主线程可以快速返回旧数据，提升并发处理能力。数据更新操作与用户请求分离，减少了阻塞等待。
4. 减少热点数据竞争 高并发场景下的竞争：热点数据被大量请求时，多个线程可能同时触发缓存更新逻辑，产生资源竞争。 单子线程更新的效果：后台线程独占更新任务，避免多线程竞争更新缓存。配合分布式锁机制，可以有效减少竞争开销。
5. 提升系统的稳定性 数据库保护：异步更新缓存，减缓数据库的瞬时高并发压力。在极端情况下，即使缓存更新失败，系统仍能通过返回旧数据保持基本的服务能力。 熔断机制结合：子线程的异步更新可以结合熔断、降级等机制，当更新任务失败时，系统可快速响应并记录失败日志以便后续处理。

·适用场景 热点数据：商品详情页、排行榜等访问量极高的场景。 高并发场景：秒杀、抢购活动中，需要频繁访问热点数据。 容忍短暂数据不一致的场景：如排行榜数据的延迟更新对用户体验影响较小。

使用 Redis + Lua 脚本实现对用户秒杀资格的预检，同时用乐观锁解决秒杀产生的超卖问题

为什么是要用 Redis+Lua？

Redis 执行一条命令的时候是具备原子性的，因为 Redis 执行命令是单线程的，不存在线程安全的问题，但当执行多条 Redis 命令时，就不是的了，我们把多条 Redis 指令放到 Lua 脚本中，Redis 会把 Lua 脚本作为一个整体执行，保证了原子性，无需加锁，天然互斥。 我使用 Redis + Lua 实现秒杀资格预检，是因为它能在一次原子操作中完成库存判断、用户去重、扣减库存、记录订单、发消息，避免了传统多命令方式的竞态条件。同时，Lua 脚本减少了网络开销，提升了性能，是高并发场景下的最佳实践。

说说这一套流程？（就是 seckillVoucher 方法）

先获取用户 ID 和订单 ID，再将优惠券 ID，用户 ID 和订单 ID 传给 Lua 脚本执行，进行资格预检，根据 Lua 脚本的返回值（0: 成功，1: 库存不足，2: 重复下单）返回对应的错误信息。将订单信息异步发送到 RabbitMQ 队列，由消费者处理后续逻辑，最后返回订单 ID 给前端。

什么是超卖问题，怎么解决？

超卖问题：并发多线程问题，当线程 1 查询库存后，判断前，又有别的线程来查询，从而造成判断错误，超卖。

解决方式： 悲观锁：添加同步锁，让线程串行执行 优点：简单粗暴 缺点：性能一般 乐观锁：不加锁，再更新时判断是否有其他线程在修改 优点：性能好 缺点：存在成功率低的问题 (该项目在超卖问题中，不在需要判断数据查询时前后是否一致，直接判读库存>0;有的项目里不是库存，只能判断数据有没有变化时，还可以用分段锁，将数据分到 10 个表，同时十个去抢)

说一下乐观锁和悲观锁？

悲观锁：悲观锁总是假设最坏的情况，认为共享资源每次被访问的时候就会出现问题 (比如共享数据被修改)，所以每次在获取资源操作的时候都会上锁，这样其他线程想拿到这个资源就会阻塞直到锁被上一个持有者释放。也就是说，共享资源每次只给一个线程使用，其它线程阻塞，用完后再把资源转让给其它线程。 乐观锁：乐观锁总是假设最好的情况，认为共享资源每次被访问的时候不会出现问题，线程可以不停地执行，无需加锁也无需等待，只是在提交修改的时候去验证对应的资源（也就是数据）是否被其它线程修改了（具体方法可以使用版本号机制或 CAS 算法）。 悲观锁通常多用于写比较多的情况（多写场景，竞争激烈），这样可以避免频繁失败和重试影响性能，悲观锁的开销是固定的。不过，如果乐观锁解决了频繁失败和重试这个问题的话（比如 LongAdder），也是可以考虑使用乐观锁的，要视实际情况而定。 乐观锁通常多用于写比较少的情况（多读场景，竞争较少），这样可以避免频繁加锁影响性能。不过，乐观锁主要针对的对象是单个共享变量（参考 java.util.concurrent.atomic 包下面的原子变量类）。

你使用的什么？

使用的是乐观锁 CAS 算法。CAS 是一个原子操作，底层依赖于一条 CPU 的原子指令。 设计三个参数：

• V:要更新的变量值
• E：预期值
• N：拟入的新值

当且仅当 V 的值等于 E 时，CAS 通过原子方式用新值 N 来更新 V 的值。如果不等，说明已经有其他线程更新了 V，则当前线程放弃更新。

从业务的角度看，只要库存数还有，就能执行这个操作，所以 where 条件设置为 stock>0

使用 Redis 分布式锁解决了在集群模式下一人一单的线程安全问题

为了防止批量刷券，添加逻辑：根据优惠券 id 和用户 id 查询订单，如果不存在，则创建。 在集群模式下，加锁只是对该 JVM 给当前这台服务器的请求的加锁，而集群是多台服务器，所以要使用分布式锁，满足集群模式下多进程可见并且互斥的锁。

Redis 分布式锁实现思路？ 我使用的 Redisson 分布式锁，他能做到可重入，可重试。 可重入:同一线程可以多次获取同一把锁，可以避免死锁，用 hash 结构存储。 大 key 是根据业务设置的，小 key 是线程唯一标识，value 值是当前重入次数。

可重试：Redisson 手动加锁，可以控制锁的失效时间和等待时间，当锁住的一个业务并没有执行完成的时候，Redisson 会引入一个 Watch Dog 看门狗机制。就是说，每隔一段时间就检查当前事务是否还持有锁。如果持有，就增加锁的持有时间。当业务执行完成之后，需要使用释放锁就可以了。还有个好处就是，在高并发下，一个业务有可能会执行很快。客户 1 持有锁的时候，客户 2 来了以后并不会马上拒绝，他会自旋不断尝试获取锁。如果客户 1 释放之后，客户 2 可以立马持有锁，性能也能得到提升。

主从一致性：连锁 (multiLock)-不再有主从节点，都获取成功才能获取锁成功，有一个节点获取锁不成功就获取锁失败 一个宕机了，还有两个节点存活，锁依旧有效，可用性随节点增多而增强。如果想让可用性更强，也可以给多个节点建立主从关系，做主从同步，但不会有主从一致问题，当新线程来新的主节点获取锁，由于另外两个主节点依然有锁，不会出现锁失效问题吗，所以不会获取成功。

基于 Stream 结构作为消息队列，实现异步秒杀下单

为什么用异步秒杀？ 我们用 jmeter 测试，发现高并发下异常率高，吞吐量低，平均耗时高。 整个业务流程是串行执行的，查询优惠券，查询订单，减库存，创建订单这四步都是走的数据库，mysql 本身并发能力就较少，还有读写操作，还加了分布式锁，整个业务耗时长，并发能力弱。

怎么进行优化？ 我们分成两个线程，我们将耗时较短的逻辑判断放到 Redis 中，例如：库存是否充足，是否一人一单这样的操作，只要满足这两条操作，那我们是一定可以下单成功的，不用等数据真的写进数据库，我们直接告诉用户下单成功就好了，将信息引入异步队列记录相关信息，然后后台再开一个线程，后台线程再去慢慢执行队列里的消息，这样我们就能很快的完成下单业务。

• 当用户下单之后，判断库存是否充足，只需要取 Redis 中根据 key 找对应的 value 是否大于 0 即可，如果不充足，则直接结束。如果充足，则在 Redis 中判断用户是否可以下单，如果 set 集合中没有该用户的下单数据，则可以下单，并将 userId 和优惠券存入到 Redis 中，并且返回 0，整个过程需要保证是原子性的，所以我们要用 Lua 来操作，同时由于我们需要在 Redis 中查询优惠券信息，所以在我们新增秒杀优惠券的同时，需要将优惠券信息保存到 Redis 中
• 完成以上逻辑判断时，我们只需要判断当前 Redis 中的返回值是否为 0，如果是 0，则表示可以下单，将信息保存到 queue 中去，然后返回，开一个线程来异步下单，其异步下单可以通过返回订单的 id 来判断是否下单成功

说说 Stream 类型消息队列？ 使用的是消费者组模式（Consumer Group）

• 消费者组 (Consumer Group)：将多个消费者划分到一个组中，监听同一个队列，具备以下特点
  1. 消息分流 队列中的消息会分留给组内的不同消费者，而不是重复消费者，从而加快消息处理的速度
  2. 消息标识 消费者会维护一个标识，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标识之后读取消息，确保每一个消息都会被消费
  3. 消息确认 消费者获取消息后，消息处于 pending 状态，并存入一个 pending-list，当处理完成后，需要通过 XACK 来确认消息，标记消息为已处理，才会从 pending-list 中移除

基本语法：

• key: 队列名称
• groupName: 消费者组名称
• ID: 起始 ID 标识，$代表队列中的最后一个消息，0 代表队列中的第一个消息
• MKSTREAM: 队列不存在时自动创建队列
• group: 消费者组名称
• consumer: 消费者名，如果消费者不存在，会自动创建一个消费者
• count: 本次查询的最大数量
• BLOCK milliseconds: 当前没有消息时的最大等待时间
• NOACK: 无需手动 ACK，获取到消息后自动确认（一般不用，我们都是手动确认）
• STREAMS key: 指定队列名称
• ID: 获取消息的起始 ID
  • >：从下一个未消费的消息开始 (pending-list 中)
  • 其他：根据指定 id 从 pending-list 中获取已消费但未确认的消息，例如 0，是从 pending-list 中的第一个消息开始

从消费者组中读取消息

XREADGROUP GROUP group consumer [COUNT count][BLOCK milliseconds][NOACK] STREAMS key [keys ...] ID [ID ...]

删除消费者组中指定的消费者

XGROUP DELCONSUMER key groupName consumerName 

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

XGROUP CREATECONSUMER key groupName consumerName 

删除指定的消费者组

XGROUP DESTORY key groupName 

创建消费者组

XGROUP CREATE key groupName ID [MKSTREAM]

基本思路：

while(true){ // 尝试监听队列，使用阻塞模式，最大等待时长为 2000ms
Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >");
if(msg ==null){ // 没监听到消息，重试 continue;}
try{//处理消息，完成后要手动确认 ACK，ACK 代码在 handleMessage 中编写
handleMessage(msg);}
catch(Exception e){
while(true){//0 表示从 pending-list 中的第一个消息开始，如果前面都 ACK 了，那么这里就不会监听到消息
Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0");
if(msg ==null){//null 表示没有异常消息，所有消息均已确认，结束循环 break;}
try{//说明有异常消息，再次处理
handleMessage(msg);}
catch(Exception e){//再次出现异常，记录日志，继续循环 log.error(".."); continue;}}}}

XREADGROUP 命令的特点？

1. 消息可回溯
2. 可以多消费者争抢消息，加快消费速度
3. 可以阻塞读取
4. 没有消息漏读风险
5. 有消息确认机制，保证消息至少被消费一次 改进：使用 RabbitMQ 更适合。

1.在 application.yml 中配置 RabbitMQ

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

• 2. 声明队列和交换机
  • 正常队列和交换机的绑定： commonExchange ('Common') → 使用路由键 'CQ' 绑定到 queueC ('CQ')。
  • 死信队列和交换机的绑定： deadLetterExchange ('Dead-letter') → 使用路由键 'DLQ' 绑定到 deadLetterQueueD ('DLQ')。
  • 普通队列到死信交换机的关系（死信机制）： queueC ('CQ') 配置了：
    • 死信交换机为 Dead-letter；
    • 死信路由键为 'DLQ'；
    • TTL 为 10 秒。
  • 当 queueC 中的消息超过 TTL 或触发其它死信条件后，这些消息将被自动发送到 deadLetterExchange，再由 deadLetterExchange 根据 'DLQ' 路由键路由到 deadLetterQueueD。

@Configuration
public class QueueConfig {
    // 普通交换机名称
    public static final String COMMON_EXCHANGE = "Common";
    // 死信交换机名称
    public static final String DEAD_DEAD_LETTER_EXCHANGE = "Dead-letter";
    // 普通队列名称
    public static final String QUEUE_C = "CQ";
    // 死信队列名称
    public static final String DEAD_LETTER_QUEUE_D = "DLQ";

    /**
     * 声明普通交换机
     *
     * @return DirectExchange
     */
    @Bean("commonExchange")
    public DirectExchange commonExchange(){
        return new DirectExchange(COMMON_EXCHANGE);
    }

    /**
     * 声明死信交换机
     *
     * @return DirectExchange
     */
    @Bean("deadLetterExchange")
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_DEAD_LETTER_EXCHANGE);
    }

    /**
     * 声明普通队列 C, 并绑定死信交换机及设置消息 TTL
     *
     * 设置说明：
     * - x-dead-letter-exchange: 配置消息过期后转发的死信交换机名称
     * - x-dead-letter-routing-key: 配置转发到死信交换机时使用的路由键，此处与死信队列绑定时的路由键一致（"DLQ"）
     * - x-message-ttl: 消息存活时间（此处设置为 10000 毫秒，即 10 秒）
     *
     * @return Queue
     */
    @Bean("queueC")
    public Queue queueC(){
        HashMap<String,Object> arguments = new HashMap<>();
        // 消息在队列中存活 10 秒后失效，进入死信队列
        arguments.put("x-message-ttl", 10000);
        // 配置死信交换机
        arguments.put("x-dead-letter-exchange", DEAD_DEAD_LETTER_EXCHANGE);
        // 配置死信路由键，绑定到死信队列时使用
        arguments.put("x-dead-letter-routing-key","DLQ");
        return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
    }

    /**
     * 声明死信队列 D
     *
     * @return Queue
     */
    @Bean("deadLetterQueueD")
    public Queue deadLetterQueueD(){
        return QueueBuilder.durable(DEAD_LETTER_QUEUE_D).build();
    }

    /**
     * 普通队列 C 与普通交换机 Common 绑定
     *
     * 当消息发送到交换机 Common，并使用路由键 "CQ" 时，
     * 消息将被路由到队列 CQ。
     *
     * @param queueC 普通队列
     * @param commonExchange 普通交换机
     * @return Binding
     */
    @Bean
    public Binding bindingQueueCToCommonExchange(@Qualifier("queueC") Queue queueC,
                                                 @Qualifier("commonExchange") DirectExchange commonExchange){
        return BindingBuilder.bind(queueC).to(commonExchange).with("CQ");
    }

    /**
     * 死信队列 D 与死信交换机 Dead-letter 绑定
     *
     * 当普通队列 CQ 中的消息由于 TTL 过期或其他原因被转为死信后，
     * 消息会转发到死信交换机 Dead-letter，并使用路由键 "DLQ"，
     * 从而被路由到死信队列 DLQ。
     *
     * @param deadLetterQueueD 死信队列
     * @param deadLetterExchange 死信交换机
     * @return Binding
     */
    @Bean
    public Binding bindingDeadLetterQueueDToDeadLetterExchange(@Qualifier("deadLetterQueueD") Queue deadLetterQueueD,
                                                               @Qualifier("deadLetterExchange") DirectExchange deadLetterExchange){
        return BindingBuilder.bind(deadLetterQueueD).to(deadLetterExchange).with("DLQ");
    }
}

3.发送者

VoucherOrder order = new VoucherOrder();
order.setId(orderId);
order.setUserId(userId);
order.setVoucherId(voucherId);
// 你可以用 JSON，也可以用序列化
// 增加消息发送的异常处理
//放入 mq
String jsonStr = JSONUtil.toJsonStr(order);
try{ 
    rabbitTemplate.convertAndSend("Common","CQ",jsonStr );
}catch(Exception e){ 
    log.error("发送 RabbitMQ 消息失败，订单 ID: {}", orderId, e);
    throw new RuntimeException("发送消息失败");
}
// 3. 返回订单号给前端（实际下单异步处理）
return Result.ok(orderId);

4.接受者

@Component
@RequiredArgsConstructor
@Slf4j
public class SeckillVoucherListener {
    @Resource
    SeckillVoucherServiceImpl seckillVoucherService;
    @Resource
    VoucherOrderServiceImpl voucherOrderService;

    /**
     * 普通队列消费者：监听队列 "CQ"
     *
     * 消息从普通队列 "CQ" 进入后进行转换处理，保存订单，同时数据库秒杀库存减一
     *
     * @param message RabbitMQ 消息
     * @param channel 消息通道
     * @throws Exception 异常处理
     */
    @RabbitListener(queues = "CQ")
    public void receivedC(Message message, Channel channel) throws Exception{
        String msg = new String(message.getBody());
        log.info("普通队列:");
        VoucherOrder voucherOrder = JSONUtil.toBean(msg, VoucherOrder.class);
        log.info(voucherOrder.toString());
        voucherOrderService.save(voucherOrder);// 保存订单到数据库
        // 秒杀业务：库存减一操作
        Long voucherId = voucherOrder.getVoucherId();
        seckillVoucherService.update()
            .setSql("stock = stock - 1")// set stock = stock - 1
            .eq("voucher_id", voucherId)
            .gt("stock",0)// where voucher_id = ? and stock > 0
            .update();
    }

    /**
     * 死信队列消费者：监听队列 "DLQ"
     *
     * 消息从死信队列 "DLQ" 进入后进行相同的处理，
     * 适用于消息因过期或其它原因进入死信队列时的处理逻辑
     *
     * @param message RabbitMQ 消息
     * @throws Exception 异常处理
     */
    @RabbitListener(queues = "DLQ")
    public void receivedDLQ(Message message) throws Exception{
        log.info("死信队列:");
        String msg = new String(message.getBody());
        VoucherOrder voucherOrder = JSONUtil.toBean(msg, VoucherOrder.class);
        log.info(voucherOrder.toString());
        voucherOrderService.save(voucherOrder);// 保存订单到数据库
        // 秒杀业务：库存减一操作
        Long voucherId = voucherOrder.getVoucherId();
        seckillVoucherService.update()
            .setSql("stock = stock - 1")// set stock = stock - 1
            .eq("voucher_id", voucherId)
            .gt("stock",0)// where voucher_id = ? and stock > 0
            .update();
    }
}

使用 Redis 的 ZSet 数据结构实现了点赞排行榜功能，使用 Set 集合实现关注、共同关注功能

什么是 ZSet? Zset，即有序集合（Sorted Set），是 Redis 提供的一种复杂数据类型。Zset 是 set 的升级版，它在 set 的基础上增加了一个权重参数 score，使得集合中的元素能够按 score 进行有序排列。 在 Zset 中，集合元素的添加、删除和查找的时间复杂度都是 O(1)。这得益于 Redis 使用的是一种叫做跳跃列表（skiplist）的数据结构来实现 Zset。

为什么使用 ZSet 数据结构？ 一人只能点一次赞，对于点赞这种高频变化的数据，如果我们使用 MySQL 是十分不理智的，因为 MySQL 慢、并且并发请求 MySQL 会影响其它重要业务，容易影响整个系统的性能，继而降低了用户体验。

Zset 的主要特性包括：

1. 唯一性：和 set 类型一样，Zset 中的元素也是唯一的，也就是说，同一个元素在同一个 Zset 中只能出现一次。
2. 排序：Zset 中的元素是有序的，它们按照 score 的值从小到大排列。如果多个元素有相同的 score，那么它们会按照字典序进行排序。
3. 自动更新排序：当你修改 Zset 中的元素的 score 值时，元素的位置会自动按新的 score 值进行调整。

点赞 用 ZSet 中的 add 方法增添，时间戳作为 score（zadd key value score)

@Override
public Result updateLike(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.数据库点赞数 +1
        boolean isSuccess = update().setSql("liked=liked+1").eq("id", id).update();
        //3.2.保存用户到 redis 的 set 集合 zadd key value score
        if(isSuccess){ 
            stringRedisTemplate.opsForZSet().add(key,userId.toString(),System.currentTimeMillis());
        }
    }else{
        //4.如果已经点赞，取消点赞
        //4.1.数据库点赞数 -1
        boolean isSuccess = update().setSql("liked=liked-1").eq("id", id).update();
        if(isSuccess){
            //4.2.将用户从 set 集合中移除
            stringRedisTemplate.opsForZSet().remove(key,userId.toString());
        }
    }
    return Result.ok();
}

共同关注 通过 Set 中的 intersect 方法求两个 key 的交集

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

@Override
public Result followCommons(Long id){
    //获取当前用户
    Long userId = UserHolder.getUser().getId();
    String key = "follows:" + userId;
    //求交集
    String key2 = "follows:" + id;
    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());
    //查询用户
    List<UserDTO> userDTOS = userService.listByIds(ids).stream().map(user -> BeanUtil.copyProperties(user,UserDTO.class)).collect(Collectors.toList());
    return Result.ok(userDTOS);
}

附近商铺搜索

我使用的是 Redis 的 GEO 数据结构，来存储商户地理座标

GEO 数据结构简介

• Redis GEO 本质：底层基于 Sorted Set（有序集合）实现，存储每个位置的经纬度信息，并支持快速范围查询。
• 核心能力：
  • 添加位置：存储商户 ID 及其经纬度。
  • 计算距离：获取两个位置间的距离。
  • 范围搜索：查找某中心点半径范围内的所有商户。
  • 排序：按距离升序或降序返回结果。

基本操作

1. GEOADD：添加地理位置 作用：将经纬度与成员（如商户 ID）关联，存储到 GEO 集合中。 语法：

GEOADD key 经度 1 纬度 1 成员 1 [经度 2 纬度 2 成员 2 ...]

GEOADD shops:geo 116.39712839.916527"shop:1001"116.40528539.904987"shop:1002"

说明：

• 经纬度范围为：经度（-180 到 180），纬度（-85.05112878 到 85.05112878）。
• 若成员已存在，会更新其经纬度。
• 返回值为成功添加的成员数量（忽略重复成员的更新）。

2. GEOPOS：获取成员坐标 作用：查询指定成员的经纬度。 语法：

GEOPOS key 成员 1 [成员 2 ...]

示例：

GEOPOS shops:geo "shop:1001"

输出：

1)1)"116.39712721109390259"# 经度
2)"39.91652652951830512"# 纬度

说明：

• 若成员不存在，返回 nil。
• 返回值为数组格式，顺序与查询成员一致。

3. GEODIST：计算两个成员间的距离 作用：返回两个地理位置之间的距离。 语法：

GEODIST key 成员 1 成员 2 [单位]

单位参数：

• m（米，默认）、km（千米）、mi（英里）、ft（英尺）。 示例：

GEODIST shops:geo "shop:1001" "shop:1002" km

输出：

"1.6423"# 单位：千米

说明：

• 若任一成员不存在，返回 nil。
• 使用 Haversine 公式计算球面距离，误差<0.5%。

4. GEORADIUS：根据中心点搜索半径内的成员 作用：以指定经纬度为中心，搜索半径范围内的成员。 语法：

GEORADIUS key 经度 纬度 半径 单位 [WITHDIST][WITHCOORD][ASC|DESC][COUNT 数量]

参数说明：

• WITHDIST：返回成员与中心点的距离。
• WITHCOORD：返回成员的经纬度。
• ASC/DESC：按距离升序/降序排序（默认升序）。
• COUNT：限制返回结果数量。 示例：

GEORADIUS shops:geo 116.40384739.9155265 km WITHDIST ASC COUNT 10

输出：

1)1)"shop:1001"# 成员
2)"0.8521"# 距离（单位：km）
2)1)"shop:1002"
2)"1.6423"

5. GEORADIUSBYMEMBER：根据成员位置搜索 作用：以某个成员的位置为中心，搜索半径范围内的其他成员。 语法：

GEORADIUSBYMEMBER key 成员 半径 单位 [WITHDIST][WITHCOORD][ASC|DESC][COUNT 数量]

示例：

GEORADIUSBYMEMBER shops:geo "shop:1001"2 km WITHCOORD

输出：

1)1)"shop:1001"
2)"0.0000"# 距离（自身）
3)1)"116.39712721109390259"
2)"39.91652652951830512"

用户签到

基于 Redis 中的 BitMap 数据结构实现。

• BitMap 概念 Bitmap，即位图，是一串连续的二进制数组（0 和 1），可以通过偏移量（offset）定位元素。BitMap 通过最小的单位 bit 来进行 0|1 的设置，表示某个元素的值或者状态，时间复杂度为 O(1)。我们将签到记录为 1，为签到记录为 0。 由于 bit 是计算机中最小的单位，使用它进行储存将非常节省空间，特别适合一些数据量大且使用二值统计的场景。 Bitmap 本身是用 String 类型作为底层数据结构实现的一种统计二值状态的数据类型。 String 类型是会保存为二进制的字节数组，所以，Redis 就把字节数组的每个 bit 位利用起来，用来表示一个元素的二值状态，你可以把 Bitmap 看作是一个 bit 数组。

• 基本命令

设置标记 即 setbit，主要是指将某个索引，设置为 1(设置 0 表示抹去标记)

@Autowired
private StringRedisTemplate redisTemplate;
/**
 * 设置标记位
 *
 * @param key
 * @param offset
 * @param tag
 * @return
 */
public Boolean mark(String key,long offset,boolean tag){
    return redisTemplate.opsForValue().setBit(key, offset, tag);
}

判断存在与否 即 getbit key index，如果返回 1，表示存在否则不存在

/**
 * 判断是否标记过
 *
 * @param key
 * @param offest
 * @return
 */
public Boolean container(String key,long offest){
    return redisTemplate.opsForValue().getBit(key, offest);
}

计数 即 bitcount key，统计和

/**
 * 统计计数
 *
 * @param key
 * @return
 */
public long bitCount(String key){
    return redisTemplate.execute(new RedisCallback<Long>(){
        @Override
        public Long doInRedis(RedisConnection redisConnection) throws DataAccessException{
            return redisConnection.bitCount(key.getBytes());
        }
    });
}

项目中我创建了一个记录签到的表，实现签到接口，将当前用户当天签到信息保存到 Redis 中

/**
 * 用户签到
 *
 * @return
 */
public Result sign(){
    // 1. 获取当前登录用户的 ID（从 ThreadLocal 中获取用户信息，确保线程安全）
    Long userId = ThreadLocalUtls.getUser().getId();
    // 2. 获取当前日期时间（使用系统默认时区）
    LocalDateTime now = LocalDateTime.now();
    // 3. 拼接 Redis 键名：格式为 "sign:用户 ID:年月"
    // 示例：用户 1001 在 2023 年 10 月签到 → "sign:1001:202310"
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = USER_SIGN_KEY + userId + keySuffix;
    // 4. 获取今天是本月的第几天（范围 1~31）
    int dayOfMonth = now.getDayOfMonth();
    // 5. 使用 Redis 位图记录签到（偏移量从 0 开始，故需 -1）
    stringRedisTemplate.opsForValue().setBit(key, dayOfMonth -1,true);
    // 6. 返回操作成功结果
    return Result.ok();
}

/**
 * 记录连续签到的天数
 *
 * @return
 */
@Override
public Result signCount(){
    // 1、获取签到记录
    // 获取当前登录用户
    Long userId = ThreadLocalUtls.getUser().getId();
    // 获取日期
    LocalDateTime now = LocalDateTime.now();
    // 拼接 key
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = USER_SIGN_KEY + userId + keySuffix;
    // 获取今天是本月的第几天
    int dayOfMonth = now.getDayOfMonth();
    // 获取本月截止今天为止的所有的签到记录，返回的是一个十进制的数字
    BITFIELD sign:5:202203 GET u14 0
    List<Long> result = stringRedisTemplate.opsForValue().bitField(
        key,
        BitFieldSubCommands.create().get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0));
    // 2、判断签到记录是否存在
    if(result ==null|| result.isEmpty()){
        // 没有任何签到结果
        return Result.ok(0);
    }
    // 3、获取本月的签到数（List<Long>是因为 BitFieldSubCommands 是一个子命令，可能存在多个返回结果，这里我们知识使用了 Get，
    // 可以明确只有一个返回结果，即为本月的签到数，所以这里就可以直接通过 get(0) 来获取）
    Long num = result.get(0);
    if(num ==null|| num==0){
        // 二次判断签到结果是否存在，让代码更加健壮
        return Result.ok(0);
    }
    // 4、循环遍历，获取连续签到的天数（从当前天起始）
    int count =0;
    while(true){
        // 让这个数字与 1 做与运算，得到数字的最后一个 bit 位，并且判断这个 bit 位是否为 0
        if((num &1)==0){
            // 如果为 0，说明未签到，结束
            break;
        }else{
            // 如果不为 0，说明已签到，计数器 +1
            count++;
        }
        // 把数字右移一位，抛弃最后一个 bit 位，继续下一个 bit 位
        num >>>=1;
    }
    return Result.ok(count);
}

UV 统计

UV 统计（Unique Visitor Statistics）是用于衡量网站、应用程序或其他在线服务中独立访客数量的关键指标。以下是关于 UV 统计的详细解析：

1. UV 的定义

• 核心概念：UV（Unique Visitor，独立访客）指在一定时间范围内（通常为一天），访问某网站或页面的不同用户数量。同一用户多次访问仅计为 1 次。
• 与 PV 的区别：
  • PV（Page View，页面浏览量）：统计用户每次访问页面的次数，多次刷新页面会累加。
  • UV 更关注用户身份的唯一性，而 PV 反映页面热度

2.实现方法 使用 HyperLogLog(HLL) 数据结构

• 原理：基于概率算法，通过哈希函数估算集合基数（即唯一元素数量），无需存储完整用户数据，极大节省内存。
• Redis 实现：
  • 内存占用：单个 HLL 结构仅需≤16KB，误差率<0.81%，适合高并发场景。
  • 命令示例：
    • PFADD key user_id：添加用户到 HLL。
    • PFCOUNT key：获取 UV 估算值

模拟用户数据

/**
 * 测试 HyperLogLog 实现 UV 统计的误差
 */
@Test
public 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){
            // 发送到 Redis
            stringRedisTemplate.opsForHyperLogLog().add("hl2", values);
        }
    }
    // 统计数量
    Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2");
    System.out.println("count = " + count);
}