跳到主要内容
黑马点评项目实战:Redis 缓存策略与 RabbitMQ 异步秒杀 | 极客日志
Java java 算法
黑马点评项目实战:Redis 缓存策略与 RabbitMQ 异步秒杀 综述由AI生成 黑马点评项目基于 SpringBoot 和 Redis 实现高并发场景下的核心功能。文章详解了使用 Redis 替代 Session 解决集群共享问题,Cache Aside 模式保证数据一致性,以及布隆过滤器、互斥锁、逻辑过期等方案应对缓存穿透、雪崩和击穿。通过 Redisson 分布式锁实现一人一单,利用 Lua 脚本原子性预检秒杀资格。采用 Stream 或 RabbitMQ 消息队列实现异步下单,结合 ZSet、GEO、BitMap 及 HyperLogLog 分别处理点赞排行、附近商户、用户签到和 UV 统计,有效提升系统性能与吞吐量。
安卓系统 发布于 2026/3/25 更新于 2026/5/11 9 浏览项目描述
黑马点评项目是一个 SpringBoot 开发的前后端分离项目,使用了 Redis 集群、Tomcat 集群、MySQL 集群提高服务性能。类似于大众点评,实现了短信登录、商户查询缓存、优惠券秒杀、附近的商户、UV 统计、用户签到、好友关注、达人探店八个部分形成了闭环。其中重点使用了分布式锁实现了一人一单功能、项目中大量使用了 Redis 的知识。
所用技术
SpringBoot+nginx+MySql+Lombok+MyBatis-Plus+Hutool+Redis
使用 Redis 解决了在集群模式下的 Session 共享问题,使用拦截器实现用户的登录校验和权限刷新。
基于 Cache Aside 模式解决数据库与缓存的一致性问题。
使用 Redis 对高频访问的信息进行缓存,降低了数据库查询的压力,解决了缓存穿透、雪崩、击穿问题。
使用 Redis + Lua 脚本实现对用户秒杀资格的预检,同时用乐观锁解决秒杀产生的超卖问题。
使用 Redis 分布式锁解决了在集群模式下一人一单的线程安全问题。
基于 Stream 结构作为消息队列,实现异步秒杀下单。
使用 Redis 的 ZSet 数据结构实现了点赞排行榜功能,使用 Set 集合实现关注、共同关注功能。
项目亮点难点
使用 Redis 解决了在集群模式下的 Session 共享问题,使用拦截器实现了用户的登录校验和权限刷新
为什么用 Redis 替代 Session?
使用 Session 时,根据客户端发送的 session-id 获取 Session,再从 Session 获取数据。由于 Session 共享问题:多台 Tomcat 并不共享 session 存储空间,当请求切换到不同 Tomcat 服务时导致数据丢失的问题。
解决方法:用 Redis 代替 Session 存储 User 信息,注册用户时,会生成一个随机的 Token 作为 Key 值存放用户到 Redis 中。
还有其他解决方法吗?
基于 Cookie 的 Token 机制,不再使用服务器端保存 Session,而是通过客户端保存 Token(如 JWT)。Token 包含用户的认证信息(如用户 ID、权限等),并通过签名验证其完整性和真实性。每次请求,客户端将 Token 放在 Cookie 或 HTTP 头中发送到服务。
说说你的登录流程?
使用 Redis 代替 session 作为缓存,使用 Token 作为唯一的 key 值。
怎么使用拦截器实现这些功能?
系统中设置了两层拦截器:
第一层拦截器是做全局处理,例如获取 Token,查询 Redis 中的用户信息,刷新 Token 有效期等通用操作。
第二层拦截器专注于验证用户登录的逻辑,如果路径需要登录,但用户未登录,则直接拦截请求。
使用两层的原因?
使用拦截器是因为,多个线程都需要获取用户,在想要方法之前统一做些操作,就需要用拦截器,还可以拦截没用登录的用户,但只有一层拦截器不是拦截所有请求,所有有些请求不会刷新 Token 时间,我们就需要再加一层拦截器,拦截所有请求,做到一直刷新。
好处:
职责分离:这种分层设计让每个拦截器的职责更加单一,代码更加清晰、易于维护。
提升性能:如果直接在第一层拦截器处理登录验证,可能会对每个请求都进行不必要的检查。而第二层拦截器仅在'需要登录的路径'中生效,可以避免不必要的性能开销。
灵活性:这种机制方便扩展,不需要修改第一层的全局逻辑。
复用 ThreadLocal 数据:第一层拦截器已经将用户信息保存到 ThreadLocal 中,第二层拦截器可以直接使用这些数据,而不需要重复查询 Redis 或其他数据源。
基于 Cache Aside 模式解决数据库与缓存的一致性问题
怎么保证缓存更新策略的高一致性需求?
我们使用的是 Redisson 实现的读写锁,再读的时候添加共享锁,可以保证读读不互斥,读写互斥。我们更新数据的时候,添加排他锁,它是读写,读读都互斥,这样就能保证在写数据的同时,是不会让其他线程读数据的,避免了脏数据。读方法和写方法是同一把锁。
使用 Redis 对高频访问的信息进行缓存,降低了数据库查询的压力,解决了缓存穿透、雪崩、击穿问题
什么是缓存穿透,怎么解决?
定义: 1.用户请求的 id 在缓存中不存在。2.恶意用户伪造不存在的 id 发起请求。
大量并发去访问一个数据库不存在的数据,由于缓存中没有该数据导致大量并发查询数据库,这个现象叫缓存穿透。缓存穿透可以造成数据库瞬间压力过大,连接数等资源用完,最终数据库拒绝连接不可用。
解决方法:
对请求增加校验机制
eg:字段 id 是长整型,如果发来的不是长整型则直接返回
使用布隆过滤器
为了避免缓存穿透我们需要缓存预热将要查询的课程或商品信息的 id 提前存入布隆过滤器,添加数据时将信息的 id 也存入过滤器,当去查询一个数据时先在布隆过滤器中找一下如果没有到到就说明不存在,此时直接返回。
缓存空值或特殊值(本项目应用)
请求通过了第一步的校验,查询数据库得到的数据不存在,此时我们仍然去缓存数据,缓存一个空值或一个特殊值的数据。但是要注意:如果缓存了空值或特殊值要设置一个短暂的过期时间。 定义: 缓存雪崩是缓存中大量 key 失效后当高并发到来时导致大量请求到数据库,瞬间耗尽数据库资源,导致数据库无法使用。
造成缓存雪崩问题的原因是是大量 key 拥有了相同的过期时间,比如对课程信息设置缓存过期时间为 10 分钟,在大量请求同时查询大量的课程信息时,此时就会有大量的课程存在相同的过期时间,一旦失效将同时失效,造成雪崩问题。
使用同步锁控制查询数据库的线程
synchronized (obj){
对同一类型信息的 key 设置不同的过期时间
通常对一类信息的 key 设置的过期时间是相同的,这里可以在原有固定时间的基础上加上一个随机时间使它们的过期时间都不相同。
缓存预热
不用等到请求到来再去查询数据库存入缓存,可以提前将数据存入缓存。使用缓存预热机制通常有专门的后台程序去将数据库的数据同步到缓存。
定义: 缓存击穿是指大量并发访问同一个热点数据,当热点数据失效后同时去请求数据库,瞬间耗尽数据库资源,导致数据库无法使用。比如某手机新品发布,当缓存失效时有大量并发到来导致同时去访问数据库。
基于互斥锁解决
互斥锁(时间换空间)
优点:内存占用小,一致性高,实现简单
缺点:性能较低,容易出现死锁
这里使用 Redis 中的 setnx 指令实现互斥锁,只有当值不存在时才能进行 set 操作
锁的有效期更具体业务有关,需要灵活变动,一般锁的有效期是业务处理时长 10~20 倍
线程获取锁后,还需要查询缓存(也就是所谓的双检),这样才能够真正有效保障缓存不被击穿
基于逻辑过期方式
逻辑过期(空间换时间)
优点:性能高
缺点:内存占用较大,容易出现脏读
·注意:逻辑过期一定要先进行数据预热,将我们热点数据加载到缓存中
适用场景
商品详情页、排行榜等热点数据场景。
数据更新频率低,但访问量大的场景。
总结:两者相比较,互斥锁更加易于实现,但是容易发生死锁,且锁导致并行变成串行,导致系统性能下降,逻辑过期实现起来相较复杂,且需要耗费额外的内存,但是通过开启子线程重建缓存,使原来的同步阻塞变成异步,提高系统的响应速度,但是容易出现脏读
开启子线程重建缓存的作用在于提高系统的响应速度,避免因缓存击穿导致的数据库压力过大,同时保障系统在高并发场景下的稳定性,但开启子线程重建缓存可能引入数据不一致(脏读)问题。
提高系统响应速度
同步阻塞的缺点:在缓存失效时,传统方案通常会同步查询数据库更新缓存,这会导致用户请求被阻塞,特别是在高并发环境下可能出现大量线程等待,影响系统响应性能。
子线程重建缓存的优势:主线程只需返回缓存中的旧数据,避免阻塞用户请求。重建缓存的任务交由后台线程执行,提高用户体验。
减少数据库压力
缓存击穿问题:当热点数据过期时,多个线程同时访问数据库,可能导致数据库压力骤增,甚至崩溃。
子线程异步重建缓存:将数据库查询集中到一个后台线程中执行,避免多个线程同时查询数据库。即便在缓存击穿的情况下,也不会对数据库造成过大的负载。
提高系统吞吐量
同步更新的瓶颈:如果所有线程都等待缓存更新完成,系统吞吐量会因阻塞而降低。
异步重建的优化:主线程可以快速返回旧数据,提升并发处理能力。数据更新操作与用户请求分离,减少了阻塞等待。
减少热点数据竞争
高并发场景下的竞争:热点数据被大量请求时,多个线程可能同时触发缓存更新逻辑,产生资源竞争。
单子线程更新的效果:后台线程独占更新任务,避免多线程竞争更新缓存。配合分布式锁机制,可以有效减少竞争开销。
提升系统的稳定性
数据库保护:异步更新缓存,减缓数据库的瞬时高并发压力。在极端情况下,即使缓存更新失败,系统仍能通过返回旧数据保持基本的服务能力。
熔断机制结合:子线程的异步更新可以结合熔断、降级等机制,当更新任务失败时,系统可快速响应并记录失败日志以便后续处理。
·适用场景
热点数据:商品详情页、排行榜等访问量极高的场景。
高并发场景:秒杀、抢购活动中,需要频繁访问热点数据。
容忍短暂数据不一致的场景:如排行榜数据的延迟更新对用户体验影响较小。
使用 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 时,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):将多个消费者划分到一个组中,监听同一个队列,具备以下特点
消息分流
队列中的消息会分留给组内的不同消费者,而不是重复消费者,从而加快消息处理的速度
消息标识
消费者会维护一个标识,记录最后一个被处理的消息,哪怕消费者宕机重启,还会从标识之后读取消息,确保每一个消息都会被消费
消息确认
消费者获取消息后,消息处于 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 ){
Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >" );
if (msg ==null ){
try {
handleMessage(msg);}
catch (Exception e){
while (true ){
Object msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0" );
if (msg ==null ){
try {
handleMessage(msg);}
catch (Exception e){
消息可回溯
可以多消费者争抢消息,加快消费速度
可以阻塞读取
没有消息漏读风险
有消息确认机制,保证消息至少被消费一次
改进:使用 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" ;
@Bean("commonExchange")
public DirectExchange commonExchange () {
return new DirectExchange (COMMON_EXCHANGE);
}
@Bean("deadLetterExchange")
public DirectExchange deadLetterExchange () {
return new DirectExchange (DEAD_DEAD_LETTER_EXCHANGE);
}
@Bean("queueC")
public Queue queueC () {
HashMap<String,Object> arguments = new HashMap <>();
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();
}
@Bean("deadLetterQueueD")
public Queue deadLetterQueueD () {
return QueueBuilder.durable(DEAD_LETTER_QUEUE_D).build();
}
@Bean
public Binding bindingQueueCToCommonExchange (@Qualifier("queueC") Queue queueC,
@Qualifier("commonExchange") DirectExchange commonExchange) {
return BindingBuilder.bind(queueC).to(commonExchange).with("CQ" );
}
@Bean
public Binding bindingDeadLetterQueueDToDeadLetterExchange (@Qualifier("deadLetterQueueD") Queue deadLetterQueueD,
@Qualifier("deadLetterExchange") DirectExchange deadLetterExchange) {
return BindingBuilder.bind(deadLetterQueueD).to(deadLetterExchange).with("DLQ" );
}
}
VoucherOrder order = new VoucherOrder ();
order.setId(orderId);
order.setUserId(userId);
order.setVoucherId(voucherId);
String jsonStr = JSONUtil.toJsonStr(order);
try {
rabbitTemplate.convertAndSend("Common" ,"CQ" ,jsonStr );
}catch (Exception e){
log.error("发送 RabbitMQ 消息失败,订单 ID: {}" , orderId, e);
throw new RuntimeException ("发送消息失败" );
}
return Result.ok(orderId);
@Component
@RequiredArgsConstructor
@Slf4j
public class SeckillVoucherListener {
@Resource
SeckillVoucherServiceImpl seckillVoucherService;
@Resource
VoucherOrderServiceImpl voucherOrderService;
@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" )
.eq("voucher_id" , voucherId)
.gt("stock" ,0 )
.update();
}
@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" )
.eq("voucher_id" , voucherId)
.gt("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 会影响其它重要业务,容易影响整个系统的性能,继而降低了用户体验。
唯一性:和 set 类型一样,Zset 中的元素也是唯一的,也就是说,同一个元素在同一个 Zset 中只能出现一次。
排序:Zset 中的元素是有序的,它们按照 score 的值从小到大排列。如果多个元素有相同的 score,那么它们会按照字典序进行排序。
自动更新排序:当你修改 Zset 中的元素的 score 值时,元素的位置会自动按新的 score 值进行调整。
点赞
用 ZSet 中的 add 方法增添,时间戳作为 score(zadd key value score)
@Override
public Result updateLike (Long id) {
Long userId = UserHolder.getUser().getId();
String key = BLOG_LIKED_KEY + id;
Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
if (score==null ){
boolean isSuccess = update().setSql("liked=liked+1" ).eq("id" , id).update();
if (isSuccess){
stringRedisTemplate.opsForZSet().add(key,userId.toString(),System.currentTimeMillis());
}
}else {
boolean isSuccess = update().setSql("liked=liked-1" ).eq("id" , id).update();
if (isSuccess){
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;
if (isFollow){
Follow follow = new Follow ();
follow.setFollowUserId(followUserId);
follow.setUserId(userId);
boolean isSuccess = save(follow);
if (isSuccess){
sadd userId followUserId
stringRedisTemplate.opsForSet().add(key,followUserId.toString());
}
}else {
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());
}
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 数据结构,来存储商户地理座标
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
若任一成员不存在,返回 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;
public Boolean mark (String key,long offset,boolean tag) {
return redisTemplate.opsForValue().setBit(key, offset, tag);
}
判断存在与否
即 getbit key index,如果返回 1,表示存在否则不存在
public Boolean container (String key,long offest) {
return redisTemplate.opsForValue().getBit(key, offest);
}
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 中
public Result sign () {
Long userId = ThreadLocalUtls.getUser().getId();
LocalDateTime now = LocalDateTime.now();
String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM" ));
String key = USER_SIGN_KEY + userId + keySuffix;
int dayOfMonth = now.getDayOfMonth();
stringRedisTemplate.opsForValue().setBit(key, dayOfMonth -1 ,true );
return Result.ok();
}
@Override
public Result signCount () {
Long userId = ThreadLocalUtls.getUser().getId();
LocalDateTime now = LocalDateTime.now();
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 ));
if (result ==null || result.isEmpty()){
return Result.ok(0 );
}
Long num = result.get(0 );
if (num ==null || num==0 ){
return Result.ok(0 );
}
int count = 0 ;
while (true ){
if ((num &1 )==0 ){
break ;
}else {
count++;
}
num >>>=1 ;
}
return Result.ok(count);
}
UV 统计 UV 统计(Unique Visitor Statistics)是用于衡量网站、应用程序或其他在线服务中独立访客数量的关键指标。以下是关于 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 估算值
@Test
public void testHyperLogLog () {
String[] values = new String [1000 ];
int j = 0 ;
for (int i = 0 ; i <1000000 ; i++){
j = i %1000 ;
values[j]="user_" + i;
if (j ==999 ){
stringRedisTemplate.opsForHyperLogLog().add("hl2" , values);
}
}
Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2" );
System.out.println("count = " + count);
}
相关免费在线工具 Keycode 信息 查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具,Keycode 信息在线工具,online
Escape 与 Native 编解码 JavaScript 字符串转义/反转义;Java 风格 \uXXXX(Native2Ascii)编码与解码。 在线工具,Escape 与 Native 编解码在线工具,online
JavaScript / HTML 格式化 使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具,JavaScript / HTML 格式化在线工具,online
JavaScript 压缩与混淆 Terser 压缩、变量名混淆,或 javascript-obfuscator 高强度混淆(体积会增大)。 在线工具,JavaScript 压缩与混淆在线工具,online
加密/解密文本 使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
Gemini 图片去水印 基于开源反向 Alpha 混合算法去除 Gemini/Nano Banana 图片水印,支持批量处理与下载。 在线工具,Gemini 图片去水印在线工具,online