1 缓存问题
1.1 汇总：5个
01.热点key
a.定义
短时间内被频繁访问的键
b.解决
1.二级缓存：通过在本地缓存热点Key，减少对Redis的访问频率，降低单个Redis节点的压力。这种方法有效地分散了请求，适合于有多个应用实例的场景
2.备份热Key：在多个Redis实例上备份热点Key，避免所有请求集中到一个Redis实例上。这种方法通过增加冗余来分散负载，适合于分布式缓存环境
-----------------------------------------------------------------------------------------------------
3.请求分摊：将对热点Key的请求分摊到多个不同的Key上，减少单个Key的访问压力。这种方法通过应用层的逻辑分片来实现，适合于可以对请求进行分片的场景
4.异步更新：在缓存失效时，使用异步方式更新缓存，减少对数据库的直接访问。这种方法通过异步处理来降低数据库的瞬时压力，适合于对实时性要求不高的场景
5.请求合并：合并对同一Key的多个请求，减少对数据库的访问次数。这种方法通过批量处理请求来减少数据库负载，适合于高并发访问的场景
6.热点数据预热：在业务高峰期之前预先加载热点数据到缓存中，避免在高峰期时缓存未命中。这种方法通过提前准备来减少高峰期的压力，适合于可以预测热点数据的场景
7.动态调整缓存策略：根据实时监控的数据动态调整缓存策略，以适应当前的访问模式。这种方法通过灵活调整缓存策略来应对变化的访问模式，适合于访问模式动态变化的场景
8.使用更高性能的存储：将热点数据存储在更高性能的存储介质上，减少访问延迟。这种方法通过提升存储介质的性能来提高访问速度，适合于对性能要求极高的场景
02.缓存穿透
a.定义
某一KEY缓存
查询一个【一定不存在的数据】，由于缓存没有命中（因为数据根本就不存在），请求每次都会穿过缓存去查询数据库
b.解决
1.缓存空值/默认值
2.布隆过滤器
-----------------------------------------------------------------------------------------------------
3.参数校验：在请求到达缓存层之前，对请求参数进行校验，过滤掉明显无效的请求
4.限流：使用限流算法（如令牌桶、漏斗算法）限制请求频率，控制单位时间内的请求数量，防止恶意请求导致的缓存穿透
5.动态调整缓存策略：通过监控请求频率和命中率，动态调整缓存的过期时间或策略
6.使用更高性能的存储：将热点数据存储在更高性能的存储介质上，减少访问延迟
03.缓存击穿
a.定义
这个KEY是真实存在对应的值
【一个或少数几个数据被高频访问，当这些数据在缓存中过期时】，大量请求就会直接到达数据库，导致数据库瞬间压力过大
b.解决
1.加锁更新
2.热点数据不过期
-----------------------------------------------------------------------------------------------------
3.请求合并：在缓存失效时，将对同一Key的多个请求合并为一个请求，减少对数据库的访问次数
4.提前刷新缓存：通过后台任务定期检查缓存的过期时间，并在即将过期时主动更新缓存
5.使用更长的过期时间：为热点数据设置更长的过期时间，减少缓存失效的频率
6.多级缓存：本地缓存（Caffeine） + 分布式缓存（redis）
7.异步更新：当缓存失效时，先返回旧数据，同时异步更新缓存
04.缓存雪崩
a.定义
多个缓存键同时过期或失效
某一个时间点，由于【大量的缓存数据同时过期 或 缓存服务器突然宕机】了，导致所有的请求都落到了数据库上
b.解决
1.缓存高可用：通过 Redis Sentinel 或 Redis Cluster 提供高可用性，确保即使某个缓存节点故障，其他节点仍能继续提供服务，避免因单点故障导致的缓存雪崩
2.随机过期时间：通过为缓存键设置随机过期时间，避免大量缓存键在同一时间过期，从而减少瞬时高负载对数据库的冲击
3.限流与熔断机制：通过限流和熔断机制，控制数据库的请求量，防止在缓存失效时数据库被过载，保护数据库的稳定性
4.给业务添加多级缓存：本地缓存（Caffeine） + 分布式缓存（redis）
5.备份缓存：对关键数据进行备份缓存，确保在主缓存失效时仍能快速访问数据，减少对数据库的依赖
05.缓存预热
a.定义
在系统上线或启动时，提前将一些预定义的数据加载到缓存中
b.解决
1.直接写个缓存刷新页面，上线时手工操作一下
2.数据量不大，可以在项目启动的时候自动进行加载
3.定时刷新缓存
1.2 大key：value值大
00.处理大key的策略
a.分页加载
不要一次性加载整个大 key 的所有元素，而是使用 LRANGE 命令分页加载，每次只获取部分元素
b.分割key
可以将一个大的 List 分割成多个小的 List
例如根据数据的时间戳、ID 范围或其他逻辑进行分割，每个小 List 作为一个独立的 key 存储
c.数据压缩
如果数据是可以压缩的，可以在客户端对数据进行压缩后再存储到 Redis，这样可以减少内存的使用量
d.数据存储策略优化
评估是否所有数据都需要存储在 Redis 中，或者是否可以通过其他方式（如数据库）来存储部分数据
e.使用其他数据结构
如果适用，可以考虑使用其他更内存高效的数据结构，比如 zset（有序集合）
f.定期清理
定期检查并清理不再需要的数据，以释放内存
g.Redis集群
如果单个 Redis 实例的内存不足以处理大量数据，可以考虑使用 Redis 集群来分散数据和负载
h.监控和警报
使用 Redis 的 INFO MEMORY 命令或其他监控工具来监控内存使用情况，并设置警报机制
01.概述
a.定义
大 key 是指 key 对应的 value 数据量很大，而不是 key 本身很大
单个String类型的Key大小达到20KB并且OPS高
单个String达到100KB
集合类型的Key总大小达到1MB
集合类型的Key中元素超过5000个
b.影响
a.客户端超时阻塞
Redis 是单线程处理命令，操作大 key 时耗时较长，可能导致客户端长时间没有响应
b.网络阻塞
大 key 的网络流量较大，频繁访问会导致网络拥堵
c.阻塞工作线程
使用 DEL 删除大 key 时，会阻塞工作线程，影响后续命令处理
d.内存分布不均
在集群中，某些节点可能因大 key 导致内存和查询负载不均
02.如何找到大key
a.使用redis-cli --bigkeys
a.说明
可以使用 redis-cli --bigkeys 命令查找大 key。建议在从节点或低峰期执行，以免影响主节点性能
b.代码
redis-cli -h 127.0.0.1 -p 6379 -a "password" --bigkeys
b.使用SCAN命令
a.说明
通过 SCAN 命令遍历数据库，结合 TYPE 和 MEMORY USAGE 命令获取 key 的类型和内存使用情况
b.代码
SCAN 0 MATCH * COUNT 1000
TYPE keyName
MEMORY USAGE keyName
c.使用RdbTools
RdbTools 是一个第三方工具，可以解析 Redis 的 RDB 文件，帮助找到大 key
03.如何删除大key
a.分批次删除
对于集合类型，可以分批次删除，避免一次性释放大量内存导致阻塞
Hash：使用 HSCAN 和 HDEL
List：使用 LTRIM
Set：使用 SSCAN 和 SREM
ZSet：使用 ZREMRANGEBYRANK
b.异步删除
从 Redis 4.0 开始，可以使用 UNLINK 命令代替 DEL，将删除操作放入异步线程，避免阻塞主线程
UNLINK keyName
04.如何解决大key
a.历史key未使用
a.场景描述
针对这种key场景，其实存在着历史原因，可能是伴随着某个业务下线或者不使用
往往对应实现的缓存操作代码会删除，但是对于缓存数据往往不会做任何处理，久而久之
这种脏数据会一直堆积，占用着资源
b.实例经验
如果确定已经无使用，并且可以确认有持久化数据（如mysql、es等）备份的话，可以直接将对应key删除
b.元素数过多
a.场景描述
针对于Set、HASH这种场景，如果元素数量超过5000就视为大的key
b.实例经验
如果对应value值不大，可以采取平铺的形式
修改代码查询和赋值逻辑：
把原始的hGet的逻辑修改为get获取
把原始hSet的逻辑修改为set赋值
历史数据刷新到新缓存key：
通过hGetAll获取所有元素数据
循环缓存元素数据操作存储新的缓存key和value
-------------------------------------------------------------------------------------------------
public String refreshHistoryData(){
try {
String key = "historyKey";
Map<String, String> redisInfoMap= redisUtils.hGetAll(key);
if (redisInfoMap.isEmpty()){
return "查询缓存无数据";
}
for (Map.Entry<String, String> entry : redisInfoMap.entrySet()) {
String redisVal = entry.getValue();
String filedKey = entry.getKey();
String newDataRedisKey = "newDataKey"+filedKey;
redisUtils.set(newDataRedisKey,redisVal);
}
return "success";
}catch (Exception e){
LOG.error("refreshHistoryData 异常:",e);
}
return "failed";
}
-------------------------------------------------------------------------------------------------
注意:这里一定要先刷历史数据，再上线代码业务逻辑的修改。防止引发缓存雪崩
c.大对象转换存储形式
a.场景描述
复杂的大对象可以尝试将对象分拆成几个key-value，使用mGet和mSet操作对应值或者pipeline的形式
最后拼装成需要返回的大对象
b.实例经验
以系统内订单对象为例：订单对象Order基础属性有几十个，如订单号、金额、时间、类型等
对于每个订单相关信息，可以设置为单独的key，把订单信息和几个相关的关联数据每个按照单独key存储
通过mGet方式获取每个信息之后，最后封装成整体Order对象
-------------------------------------------------------------------------------------------------
public enum CacheKeyConstant {
REDIS_ORDER_BASE_INFO("ORDER_BASE_INFO"),
ORDER_SUB_INFO("ORDER_SUB_INFO"),
ORDER_PRESALE_INFO("ORDER_PRESALE_INFO"),
ORDER_PREMISE_INFO("ORDER_PREMISE_INFO"),
ORDER_INVOICE_INFO("ORDER_INVOICE_INFO"),
ORDER_TRACK_INFO("ORDER_TRACK_INFO"),
ORDER_FEE_INFO("ORDER_FEE_INFO"),
;
private String prefix;
public static final String COMMON_PREFIX = "XXX";
CacheKeyConstant(String prefix){
this.prefix = prefix;
}
public String getPrefix(String subKey) {
if(StringUtil.isNotEmpty(subKey)){
return COMMON_PREFIX + prefix + "_" + subKey;
}
return COMMON_PREFIX + prefix;
}
public String getPrefix() {
return COMMON_PREFIX + prefix;
}
}
-------------------------------------------------------------------------------------------------
缓存存储：
public boolean refreshOrderToCache(Order order){
if(order == null || order.getOrderId() == null){
return ;
}
String orderId = order.getOrderId().toString();
Map<String,String> cacheOrderMap = new HashMap<>(16);
cacheOrderMap.put(CacheKeyConstant.ORDER_BASE_INFO.getPrefix(orderId), JSON.toJSONString(buildBaseOrderVo(order)));
cacheOrderMap.put(CacheKeyConstant.ORDER_SUB_INFO.getPrefix(orderId), JSON.toJSONString(order.getCustomerOrderSubs()));
cacheOrderMap.put(CacheKeyConstant.ORDER_PRESALE_INFO.getPrefix(orderId), JSON.toJSONString(order.getPresaleOrderData()));
cacheOrderMap.put(CacheKeyConstant.ORDER_INVOICE_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderInvoice()));
cacheOrderMap.put(CacheKeyConstant.ORDER_TRACK_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderTrackInfo()));
cacheOrderMap.put(CacheKeyConstant.ORDER_PREMISE_INFO.getPrefix(orderId), JSON.toJSONString( order.getPresaleOrderData()));
cacheOrderMap.put(CacheKeyConstant.ORDER_FEE_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderFeeVo()));
superRedisUtils.mSetString(cacheOrderMap);
}
-------------------------------------------------------------------------------------------------
缓存获取：
public Order getOrderFromCache(String orderId){
if(StringUtils.isBlank(orderId)){
return null;
}
List<String> queryOrderKey = Arrays.asList(CacheKeyConstant.ORDER_BASE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_SUB_INFO.getPrefix(orderId),
CacheKeyConstant.ORDER_PRESALE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_INVOICE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_TRACK_INFO.getPrefix(orderId),
CacheKeyConstant.ORDER_PREMISE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_FEE_INFO.getPrefix(orderId));
List<String> result = redisUtils.mGet(queryOrderKey);
if(CollectionUtils.isEmpty(result)){
return null;
}
String[] resultInfo = result.toArray(new String[0]);
if(StringUtils.isBlank(resultInfo[0])){
return null;
}
BaseOrderVo baseOrderVo = JSON.parseObject(resultInfo[0],BaseOrderVo.class);
Order order = coverBaseOrderVoToOrder(baseOrderVo);
if(StringUtils.isNotBlank(resultInfo[1])){
List<OrderSub> orderSubs =JSON.parseObject(result.get(1), new TypeReference<List<OrderSub>>(){});
order.setCustomerOrderSubs(orderSubs);
}
if(StringUtils.isNotBlank(resultInfo[2])){
PresaleOrderData presaleOrderData = JSON.parseObject(resultInfo[2],PresaleOrderData.class);
order.setPresaleOrderData(presaleOrderData);
}
if(StringUtils.isNotBlank(resultInfo[3])){
OrderInvoice orderInvoice = JSON.parseObject(resultInfo[3],OrderInvoice.class);
order.setOrderInvoice(orderInvoice);
}
if(StringUtils.isNotBlank(resultInfo[5])){
OrderTrackInfo orderTrackInfo = JSON.parseObject(resultInfo[5],OrderTrackInfo.class);
order.setOrderTrackInfo(orderTrackInfo);
}
if(StringUtils.isNotBlank(resultInfo[6])){
List<OrderPremiseInfo> orderPremiseInfos =JSON.parseObject(result.get(6), new TypeReference<List<OrderPremiseInfo>>(){});
order.setPremiseInfos(orderPremiseInfos);
}
if(StringUtils.isNotBlank(resultInfo[7])){
OrderFeeVo orderFeeVo = JSON.parseObject(resultInfo[7],OrderFeeVo.class);
order.setOrderFeeVo(orderFeeVo);
}
return order;
}
-------------------------------------------------------------------------------------------------
注意：获取缓存的结果跟传入的key的顺序保持对应即可
d.压缩存储数据
a.压缩方法结果
a.单个元素时
DefaultOutputStream：压缩前大小446Byte，压缩后大小254Byte，压缩耗时1ms，解压耗时0ms
GzipOutputStream：压缩前大小446Byte，压缩后大小266Byte，压缩耗时1ms，解压耗时1ms
ZlibCompress：压缩前大小446Byte，压缩后大小254Byte，压缩耗时1ms，解压耗时0ms
b.四百个元素集合
DefaultOutputStream：压缩前大小6732Byte，压缩后大小190Byte，压缩耗时2ms，解压耗时0ms
GzipOutputStream：压缩前大小6732Byte，压缩后大小202Byte，压缩耗时1ms，解压耗时1ms
ZlibCompress：压缩前大小6732Byte，压缩后大小190Byte，压缩耗时1ms，解压耗时0ms
c.四万个元素集合时
DefaultOutputStream：压缩前大小640340Byte，压缩后大小1732Byte，压缩耗时37ms，解压耗时2ms
GzipOutputStream：压缩前大小640340Byte，压缩后大小1744Byte，压缩耗时11ms，解压耗时3ms
ZlibCompress：压缩前大小640340Byte，压缩后大小1732Byte，压缩耗时69ms，解压耗时2ms
b.压缩代码样例
a.DefaultOutputStream
public static byte[] compressToByteArray(String text) throws IOException {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
Deflater deflater = new Deflater();
DeflaterOutputStream deflaterOutputStream = new DeflaterOutputStream(outputStream, deflater);
deflaterOutputStream.write(text.getBytes());
deflaterOutputStream.close();
return outputStream.toByteArray();
}
public static String decompressFromByteArray(byte[] bytes) throws IOException {
ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
Inflater inflater = new Inflater();
InflaterInputStream inflaterInputStream = new InflaterInputStream(inputStream, inflater);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int length;
while ((length = inflaterInputStream.read(buffer)) != -1) {
outputStream.write(buffer, 0, length);
}
inflaterInputStream.close();
outputStream.close();
byte[] decompressedData = outputStream.toByteArray();
return new String(decompressedData);
}
b.GZIPOutputStream
public static byte[] compressGzip(String str) {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
GZIPOutputStream gzipOutputStream = null;
try {
gzipOutputStream = new GZIPOutputStream(outputStream);
} catch (IOException e) {
throw new RuntimeException(e);
}
try {
gzipOutputStream.write(str.getBytes("UTF-8"));
} catch (IOException e) {
throw new RuntimeException(e);
}finally {
try {
gzipOutputStream.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
return outputStream.toByteArray();
}
public static String decompressGzip(byte[] compressed) throws IOException {
ByteArrayInputStream inputStream = new ByteArrayInputStream(compressed);
GZIPInputStream gzipInputStream = new GZIPInputStream(inputStream);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
byte[] buffer = new byte[1024];
int length;
while ((length = gzipInputStream.read(buffer)) > 0) {
outputStream.write(buffer, 0, length);
}
gzipInputStream.close();
outputStream.close();
return outputStream.toString("UTF-8");
}
c.ZlibCompress
public byte[] zlibCompress(String message) throws Exception {
String chatacter = "UTF-8";
byte[] input = message.getBytes(chatacter);
BigDecimal bigDecimal = BigDecimal.valueOf(0.25f);
BigDecimal length = BigDecimal.valueOf(input.length);
byte[] output = new byte[input.length + 10 + new Double(Math.ceil(Double.parseDouble(bigDecimal.multiply(length).toString()))).intValue()];
Deflater compresser = new Deflater();
compresser.setInput(input);
compresser.finish();
int compressedDataLength = compresser.deflate(output);
compresser.end();
return Arrays.copyOf(output, compressedDataLength);
}
public static String zlibInfCompress(byte[] data) {
String s = null;
Inflater decompresser = new Inflater();
decompresser.reset();
decompresser.setInput(data);
ByteArrayOutputStream o = new ByteArrayOutputStream(data.length);
try {
byte[] buf = new byte[1024];
while (!decompresser.finished()) {
int i = decompresser.inflate(buf);
o.write(buf, 0, i);
}
s = o.toString("UTF-8");
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
o.close();
} catch (IOException e) {
e.printStackTrace();
}
}
decompresser.end();
return s;
}
1.3 热点key：二级缓存、备份热Key
00.处理热key的策略
a.二级缓存
在本地缓存热点Key，减少对Redis的访问频率
b.备份热Key
在多个Redis实例上备份热点Key，避免所有请求集中到一个Redis实例上
c.请求分摊
将对热点Key的请求分摊到多个不同的Key上
d.异步更新
使用异步方式更新缓存，减少对数据库的直接访问
e.请求合并
合并对同一Key的多个请求，减少对数据库的访问次数
f.热点数据预热
在业务高峰期之前预先加载热点数据到缓存中
g.动态调整缓存策略
将热点数据存储在更高性能的存储介质上
h.使用更高性能的存储
将热点数据存储在更高性能的存储介质上
00.概述
a.定义
在短时间内被频繁访问的键
这种键可能会导致 Redis 服务器的负载过高，尤其是在集群环境中，可能导致某个节点的资源耗尽，影响整体性能
常见场景：热门新闻、热门商品
b.产生的原因
a.热点数据
某些数据具有较高的访问频率，例如热门商品、热门新闻、热门评论等
b.业务高峰期
当处于业务高峰期的时候，某些数据会被频繁访问，例如双11秒杀、整点秒杀等
c.代码逻辑问题
如高频轮询，程序的代码逻辑导致部分Key被频繁访问，例如程序中的高频轮询或者存在代码死循环
c.怎么发现热Key？
a.业务经验预估
根据业务场景预估可能的热 Key
b.客户端收集
在操作 Redis 前加入统计代码
c.Proxy层收集
在 Proxy 层进行统计
d.Redis自带命令
MONITOR：实时抓取命令
--hotkeys：Redis 4.0.3 提供的热点 Key 发现功能
e.抓包评估
监听端口解析数据包
d.自动发现和处理热Key
a.监控热Key
a.工具
使用改写的 Jedis 包和 Hermes-SDK 进行监控
b.方法
每次访问 Redis 时，异步上报 Key 访问事件
b.通知系统处理
a.方法
监控到热 Key 后，通知各业务系统进行本地缓存
b.示例
Hermes 服务端通知业务系统缓存热 Key，后续请求直接从本地缓存获取
01.二级缓存
a.定义
在本地缓存热点Key，减少对Redis的访问频率
b.原理
通过在应用层使用本地缓存（如Ehcache或Caffeine），将热点数据缓存到本地内存中，减少对Redis的请求
c.常用API
本地缓存库的API，如Ehcache或Caffeine
d.使用步骤
1.在应用中集成本地缓存库
2.将热点Key的数据缓存到本地
3.在请求时优先从本地缓存获取数据
e.场景示例
// 使用 Caffeine 作为本地缓存
Cache<String, String> cache = Caffeine.newBuilder()
.expireAfterWrite(10, TimeUnit.MINUTES)
.maximumSize(1000)
.build();
// 缓存数据
cache.put("hotKey", "value");
// 获取数据
String value = cache.getIfPresent("hotKey");
02.备份热Key
a.定义
在多个Redis实例上备份热点Key，避免所有请求集中到一个Redis实例上
b.原理
通过在多个Redis节点上存储相同的热点Key数据，分散请求负载
c.常用API
Redis的基本操作API，如SET和GET
d.使用步骤
1.在多个Redis实例上存储热点Key
2.在请求时随机选择一个Redis实例进行访问
e.场景示例
import redis
import random
# 连接多个 Redis 实例
redis_instances = [redis.Redis(host='localhost', port=6379),
redis.Redis(host='localhost', port=6380)]
# 备份数据到多个实例
for r in redis_instances:
r.set("hotKey", "value")
# 随机选择一个实例获取数据
selected_redis = random.choice(redis_instances)
value = selected_redis.get("hotKey")
03.请求分摊
a.定义
将对热点Key的请求分摊到多个不同的Key上
b.原理
通过在应用层对请求进行分片，将同一个热点Key的请求分散到多个Key上，然后在应用层合并结果
c.常用API
应用层的逻辑实现，无特定API
d.使用步骤
1.在应用层对请求进行分片
2.将请求分散到多个Key上
3.在应用层合并结果
e.场景示例
# 分片请求
def get_data(user_id):
shard_key = f"hotKey:{user_id % 10}"
return redis.get(shard_key)
# 合并结果
results = [get_data(user_id) for user_id in user_ids]
04.异步更新
a.定义
使用异步方式更新缓存，减少对数据库的直接访问
b.原理
在缓存失效时，先返回旧数据，同时异步更新缓存
c.常用API
异步任务库，如Celery或Java的CompletableFuture
d.使用步骤
1.在缓存失效时返回旧数据
2.异步更新缓存
e.场景示例
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=2)
def update_cache(key):
# 模拟数据库查询
new_value = "new_value_from_db"
redis.set(key, new_value)
# 异步更新缓存
executor.submit(update_cache, "hotKey")
05.请求合并
a.定义
合并对同一Key的多个请求，减少对数据库的访问次数
b.原理
在应用层对同一时间段内的请求进行合并，只发送一次数据库查询请求，然后将结果返回给所有请求
c.常用API
应用层的逻辑实现，无特定API
d.使用步骤
1.在应用层合并请求
2.发送一次数据库查询请求
3.将结果返回给所有请求
e.场景示例
# 合并请求
def get_data(key):
if key not in cache:
cache[key] = db_query(key)
return cache[key]
# 模拟数据库查询
def db_query(key):
return "value_from_db"
06.热点数据预热
a.定义
在业务高峰期之前预先加载热点数据到缓存中
b.原理
通过分析历史数据，提前将可能成为热点的数据加载到缓存中，避免在高峰期时缓存未命中
c.常用API
Redis的基本操作API，如SET
d.使用步骤
1.分析历史数据，确定热点数据
2.在高峰期之前将热点数据加载到缓存中
e.场景示例
# 预热缓存
hot_keys = ["key1", "key2", "key3"]
for key in hot_keys:
redis.set(key, "preloaded_value")
07.动态调整缓存策略
a.定义
根据实时监控的数据动态调整缓存策略
b.原理
通过监控热点Key的访问频率，动态调整其缓存过期时间或缓存策略，以适应当前的访问模式
c.常用API
Redis的基本操作API，如EXPIRE
d.使用步骤
1.监控热点Key的访问频率
2.动态调整缓存策略
e.场景示例
# 动态调整过期时间
def adjust_expiry(key, access_count):
if access_count > threshold:
redis.expire(key, new_expiry_time)
08.使用更高性能的存储
a.定义
将热点数据存储在更高性能的存储介质上
b.原理
使用内存数据库或更快的存储介质来存储热点数据，减少访问延迟
c.常用API
内存数据库的基本操作API
d.使用步骤
1.确定热点数据
2.将热点数据存储在高性能存储介质上
e.场景示例
# 使用 Redis 作为高性能存储
redis.set("hotKey", "value")
1.4 缓存穿透：缓存空值/默认值、布隆过滤器、谷鸟过滤器
00.缓存穿透
a.定义
查询一个【一定不存在的数据】，由于缓存没有命中（因为数据根本就不存在），请求每次都会穿过缓存去查询数据库
如果这种查询非常频繁，就会给数据库造成很大的压力
b.方案1：缓存空值/默认值
a.适用场景
数据命中不高，保证一致性
b.维护成本
代码维护简单，但需要过多的缓存空间，可能导致数据不一致
c.解决方案
数据库无法命中之后，把一个空对象或者默认值保存到缓存，之后再访问这个数据，就会从缓存中获取，这样就保护了数据库
d.产生问题
空值做了缓存，意味着缓存层中存了更多的键，需要更多的内存空间（如果是攻击，问题更严重）
比较有效的方法是针对这类数据设置一个较短的过期时间，让其自动剔除
c.方案2：BloomFilter布隆过滤器
a.适用场景
数据命中不高，数据相对固定、实时性低
b.维护成本
代码维护复杂，但缓存空间占用小
c.方法
布隆过滤器（Bloom Filter）是一种空间效率极高的概率型数据结构，用于快速检查一个元素是否存在于一个集合中
布隆过滤器由一个长度为 m 的位数组和 k 个哈希函数组成。因为布隆过滤器占用的内存空间非常小，所以查询效率也非常高
d.缺点
因为是通过哈希函数计算的，所以存在哈希冲突的问题，可能会导致误判
常用的缓存 Redis 默认不支持 BloomFilter 数据结构
d.方案3：参数校验
a.定义
在请求到达缓存层之前，对请求参数进行校验，过滤掉明显无效的请求
b.原理
通过在应用层对请求参数进行合法性检查，避免无效请求进入缓存和数据库，从而减少不必要的数据库查询
c.常用API
无特定API，通常在应用层实现
d.使用步骤
在应用层接收到请求时，首先对请求参数进行校验
如果参数无效，直接返回错误响应
如果参数有效，继续进行缓存查询
e.场景示例
def get_user_data(user_id):
if not isinstance(user_id, int) or user_id <= 0:
return "Invalid user ID"
# 继续查询缓存或数据库
data = cache.get(user_id)
if data is None:
data = db_query(user_id)
cache.set(user_id, data)
return data
e.方案4：限流
a.定义
对请求进行限流，控制单位时间内的请求数量，防止恶意请求导致的缓存穿透
b.原理
使用限流算法（如令牌桶、漏斗算法）限制请求频率，防止短时间内大量无效请求对数据库造成压力
c.常用API
限流库或中间件
d.使用步骤
在应用层或网关层集成限流中间件
设置限流规则，控制请求频率
超过限流阈值的请求直接拒绝或延迟处理
e.场景示例
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=10, period=60)
def get_user_data(user_id):
# 查询缓存或数据库
pass
f.方案5：动态调整缓存策略
a.定义
根据实时监控的数据动态调整缓存策略，适应当前的访问模式
b.原理
通过监控请求频率和命中率，动态调整缓存的过期时间或策略，以提高缓存命中率
c.常用API
缓存库的配置API
d.使用步骤
监控缓存的请求频率和命中率
根据监控数据动态调整缓存策略，如过期时间
观察调整后的效果，持续优化
e.场景示例
def adjust_cache_strategy(key, access_count):
if access_count > threshold:
cache.set_expiry(key, new_expiry_time)
g.方案6：使用更高性能的存储
a.定义
将热点数据存储在更高性能的存储介质上，减少访问延迟
b.原理
使用内存数据库或更快的存储介质来存储热点数据，提高数据访问速度
c.常用API
内存数据库的基本操作API
d.使用步骤
确定热点数据
将热点数据存储在高性能存储介质上
优先从高性能存储中读取数据
e.场景示例
# 使用 Redis 作为高性能存储
redis.set("hotKey", "value")
value = redis.get("hotKey")
01.缓存空值/默认值
a.说明
在数据库无法命中之后，把一个空对象或者默认值保存到缓存，之后再访问这个数据，就会从缓存中获取，这样就保护了数据库
b.代码
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
public class CacheService {
private static final ConcurrentHashMap<String, CacheObject> cache = new ConcurrentHashMap<>();
private static final String NULL_VALUE = "NULL"; // 用于表示空值
public String getData(String key) {
// 从缓存获取数据
CacheObject cacheObject = cache.get(key);
if (cacheObject != null && !cacheObject.isExpired()) {
return NULL_VALUE.equals(cacheObject.getValue()) ? null : cacheObject.getValue();
}
// 缓存不存在或已过期，查询数据库
String value = queryFromDatabase(key);
if (value == null) {
cache.put(key, new CacheObject(NULL_VALUE, System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5))); // 缓存空值
return null; // 返回空
} else {
cache.put(key, new CacheObject(value, System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(30))); // 正常数据缓存30分钟
return value;
}
}
private String queryFromDatabase(String key) {
// 模拟数据库查询
return null; // 假设查询不到数据
}
// 内部类用于存储缓存数据和过期时间
private static class CacheObject {
private String value;
private long expireTime;
public CacheObject(String value, long expireTime) {
this.value = value;
this.expireTime = expireTime;
}
public String getValue() {
return value;
}
public boolean isExpired() {
return System.currentTimeMillis() > expireTime;
}
}
}
02.布隆过滤器
a.概念
a.定义
布隆过滤器是一种用于判断元素是否存在于集合中的概率性数据结构
它非常适合用于处理海量数据的存在性查询，具有高效的空间利用率
b.特点
当查询结果为“存在”时，元素可能存在（存在误判）
当查询结果为“不存在”时，元素一定不存在
c.Redis支持
从 Redis 4.0 开始，通过模块（modules）支持布隆过滤器
bf.add：添加元素
bf.exists：判断某个元素是否存在
bf.madd：添加多个元素
bf.mexists：判断多个元素是否存在
bf.reserve：设置布隆过滤器的准确率
d.场景
垃圾邮件过滤
爬虫里的 URL 去重
判断一个元素在亿级数据中是否存在
b.原理
a.数据结构
一个位数组、多个无偏哈希函数来实现
b.添加元素
使用多个哈希函数计算元素的哈希值
将这些哈希值对应的位数组位置设置为 1
c.查询元素
使用相同的哈希函数计算元素的哈希值
检查这些哈希值对应的位数组位置是否全部为 1
如果全部为 1，则元素可能存在；如果有一个为 0，则元素一定不存在
d.误差
位数组越稀疏，查询准确率越高
随着存储的元素增多，误差也会增大
c.代码
a.示例
127.0.0.1:6379> bf.add user xiaoming
(integer) 1
127.0.0.1:6379> bf.add user xiaohong
(integer) 1
127.0.0.1:6379> bf.add user laowang
(integer) 1
127.0.0.1:6379> bf.exists user laowang
(integer) 1
127.0.0.1:6379> bf.exists user lao
(integer) 0
127.0.0.1:6379> bf.madd user huahua feifei
1) (integer) 1
2) (integer) 1
127.0.0.1:6379> bf.mexists user feifei laomiao
1) (integer) 1
2) (integer) 0
b.准确率设置
a.使用 bf.reserve 可以设置布隆过滤器的准确率和初始大小
127.0.0.1:6379> bf.reserve userlist 0.9 10
b.参数说明
error_rate：允许的错误率。值越低，过滤器占用的空间越大，查询的错误率越低。默认值是 0.01
initial_size：布隆过滤器存储的初始元素大小。实际存储的元素数量超过此值时，准确率会降低。默认值是 100
c.注意事项
bf.reserve 必须在元素添加之前执行，否则会报错
设置较低的错误率和较大的数组大小会增加存储空间，但提高查询准确性
d.开启布隆过滤器
a.背景
在 Redis 中不能直接使用布隆过滤器，
但我们可以通过 Redis 4.0 版本之后提供的 modules（扩展模块）的方式引入，本文提供两种方式的开启方式。
b.方式一：编译方式
a.下载并安装布隆过滤器
git clone https://github.com/RedisLabsModules/redisbloom.git
cd redisbloom
make # 编译redisbloom
-------------------------------------------------------------------------------------------------
编译正常执行完，会在根目录生成一个 redisbloom.so 文件。
b.启动 Redis 服务器
> ./src/redis-server redis.conf --loadmodule ./src/modules/RedisBloom-master/redisbloom.so
其中 --loadmodule 为加载扩展模块的意思，后面跟的是 redisbloom.so 文件的目录。
c.方式二：Docker 方式
a.操作
docker pull redislabs/rebloom # 拉取镜像
docker run -p6379:6379 redislabs/rebloom # 运行容器
b.启动验证
服务启动之后，我们需要判断布隆过滤器是否正常开启，
此时我们只需使用 redis-cli 连接到服务端，输入 bf.add 看有没有命令提示，就可以判断是否正常启动了
e.代码实战
import redis.clients.jedis.Jedis;
import utils.JedisUtils;
import java.util.Arrays;
public class BloomExample {
private static final String _KEY = "userlist";
public static void main(String[] args) {
Jedis jedis = JedisUtils.getJedis();
for (int i = 1; i < 100001; i++) {
bfAdd(jedis, _KEY, "user_" + i);
boolean exists = bfExists(jedis, _KEY, "user_" + (i + 1));
if (exists) {
System.out.println("找到了" + i);
break;
}
}
System.out.println("执行完成");
}
/**
* 添加元素
* @param jedis Redis 客户端
* @param key key
* @param value value
* @return boolean
*/
public static boolean bfAdd(Jedis jedis, String key, String value) {
String luaStr = "return redis.call('bf.add', KEYS[1], KEYS[2])";
Object result = jedis.eval(luaStr, Arrays.asList(key, value),
Arrays.asList());
if (result.equals(1L)) {
return true;
}
return false;
}
/**
* 查询元素是否存在
* @param jedis Redis 客户端
* @param key key
* @param value value
* @return boolean
*/
public static boolean bfExists(Jedis jedis, String key, String value) {
String luaStr = "return redis.call('bf.exists', KEYS[1], KEYS[2])";
Object result = jedis.eval(luaStr, Arrays.asList(key, value),
Arrays.asList());
if (result.equals(1L)) {
return true;
}
return false;
}
}
03.谷鸟过滤器
a.谷鸟过滤器（Guava Bloom Filter）概述
a.定义
谷鸟过滤器是一种基于哈希的概率性数据结构，能够快速判断一个元素是否可能存在于集合中
它可能会误判一个不存在的元素为存在，但不会漏判一个存在的元素
b.原理
哈希函数：使用多个哈希函数将元素映射到一个位数组中
位数组：每个元素通过多个哈希函数映射到位数组的多个位置，将这些位置的位设置为 1
查询：检查元素时，通过相同的哈希函数检查对应位置的位是否都为 1。如果是，则元素可能存在；如果有任意一个位为 0，则元素一定不存在
b.常用 API
a.创建 BloomFilter
BloomFilter.create(Funnel<T> funnel, long expectedInsertions, double fpp)：创建一个布隆过滤器
b.添加元素
put(T item)：将元素添加到过滤器中
c.查询元素
mightContain(T item)：检查元素是否可能存在于过滤器中
c.使用步骤
a.引入 Guava 依赖
在项目中添加 Guava 库的依赖
b.创建 BloomFilter 实例
使用 BloomFilter.create() 方法创建一个布隆过滤器实例
c.添加元素
使用 put() 方法将元素添加到过滤器中
d.查询元素
使用 mightContain() 方法检查元素是否可能存在
d.每个场景对应的代码示例
a.引入 Guava 依赖
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>31.1-jre</version>
</dependency>
b.创建和使用 BloomFilter
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
public class BloomFilterExample {
public static void main(String[] args) {
// 创建一个布隆过滤器，预计插入 500 个整数，误判率为 0.01
BloomFilter<Integer> bloomFilter = BloomFilter.create(
Funnels.integerFunnel(),
500,
0.01);
// 添加元素到过滤器
for (int i = 0; i < 500; i++) {
bloomFilter.put(i);
}
// 查询元素是否可能存在
System.out.println(bloomFilter.mightContain(100)); // 可能输出 true
System.out.println(bloomFilter.mightContain(1000)); // 可能输出 false
}
}
1.5 缓存击穿：加锁更新、手动过期
00.缓存击穿
a.定义
【一个或少数几个数据被高频访问，当这些数据在缓存中过期时】，大量请求就会直接到达数据库，导致数据库瞬间压力过大
b.方案1：加锁更新
a.优点
思路简单，保证数据一致性
b.缺点
增加代码复杂度，存在死锁风险
c.实现思路
当缓存失效时，使用分布式锁（如 Redis 锁）来控制对数据库的访问
第一个获取锁的请求去数据库查询数据并更新缓存，其他请求等待锁释放后从缓存读取数据
c.方案2：手动过期
a.优点
性价比高，用户无需等待
b.缺点
需要额外的逻辑来管理过期时间
c.实现思路
将过期时间存储在缓存的值中，通过异步任务定期刷新缓存，防止缓存过期
d.方案3：请求合并
a.定义
在缓存失效时，将对同一Key的多个请求合并为一个请求，减少对数据库的访问次数
b.原理
在应用层对同一时间段内的请求进行合并，只发送一次数据库查询请求，然后将结果返回给所有请求
c.常用API
无特定API，通常在应用层实现
d.使用步骤
1.在应用层维护一个请求队列
2.当缓存失效时，将请求加入队列
3.只发送一次数据库查询请求
4.将查询结果返回给队列中的所有请求
e.场景示例
import threading
lock = threading.Lock()
cache = {}
db = {"key": "value_from_db"}
def get_data(key):
if key in cache:
return cache[key]
with lock:
if key in cache: # Double-check locking
return cache[key]
# Simulate database query
data = db.get(key)
cache[key] = data
return data
# Simulate multiple requests
threads = [threading.Thread(target=get_data, args=("key",)) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
e.方案4：提前刷新缓存
a.定义
在缓存即将过期时，提前刷新缓存，确保缓存始终有效
b.原理
通过后台任务定期检查缓存的过期时间，并在即将过期时主动更新缓存
c.常用API
定时任务库
d.使用步骤
1.设置缓存的过期时间
2.使用定时任务定期检查缓存的过期时间
3.在即将过期时主动更新缓存
e.场景示例
import time
import threading
cache = {"key": ("value", time.time() + 60)} # Cache with expiry time
def refresh_cache():
while True:
time.sleep(30) # Check every 30 seconds
for key, (value, expiry) in cache.items():
if time.time() > expiry - 10: # Refresh 10 seconds before expiry
# Simulate database query
new_value = "new_value_from_db"
cache[key] = (new_value, time.time() + 60)
# Start the refresh thread
threading.Thread(target=refresh_cache, daemon=True).start()
f.方案5：使用更高性能的存储
a.定义
将热点数据存储在更高性能的存储介质上，减少访问延迟
b.原理
使用内存数据库或更快的存储介质来存储热点数据，提高数据访问速度
c.常用API
内存数据库的基本操作API
d.使用步骤
1.确定热点数据
2.将热点数据存储在高性能存储介质上
3.优先从高性能存储中读取数据
e.场景示例
# 使用 Redis 作为高性能存储
import redis
r = redis.Redis()
r.set("hotKey", "value")
value = r.get("hotKey")
g.方案6：多级缓存
a.定义
结合使用本地缓存和分布式缓存，形成多级缓存体系
b.原理
在本地缓存失效时，优先访问分布式缓存，最后访问数据库
c.常用API
本地缓存库和分布式缓存库的API
d.使用步骤
1.在应用中集成本地缓存和分布式缓存
2.优先从本地缓存获取数据
3.本地缓存失效时，从分布式缓存获取数据
4.分布式缓存失效时，从数据库获取数据
e.场景示例
# 使用 Caffeine 作为本地缓存，Redis 作为分布式缓存
from cachetools import TTLCache
import redis
local_cache = TTLCache(maxsize=100, ttl=300)
r = redis.Redis()
def get_data(key):
if key in local_cache:
return local_cache[key]
value = r.get(key)
if value:
local_cache[key] = value
return value
# Simulate database query
value = "value_from_db"
r.set(key, value)
local_cache[key] = value
return value
h.方案7：异步更新
a.定义
在缓存失效时，先返回旧数据，同时异步更新缓存，减少对数据库的直接访问
b.原理
使用异步任务库，在后台异步更新缓存
c.常用API
异步任务库，如Celery或Java的CompletableFuture
d.使用步骤
1.在缓存失效时返回旧数据
2.使用异步任务更新缓存
e.场景示例
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=2)
cache = {"key": "old_value"}
def update_cache(key):
# Simulate database query
new_value = "new_value_from_db"
cache[key] = new_value
def get_data(key):
if key in cache:
executor.submit(update_cache, key)
return cache[key]
else:
# Simulate database query
value = "value_from_db"
cache[key] = value
return value
# Simulate request
print(get_data("key"))
01.加锁更新
a.思路
当请求查询某个键时，如果缓存中没有数据，对该键加锁
查询数据库并更新缓存，释放锁后其他请求可以从缓存中获取数据
b.代码
import java.util.concurrent.ConcurrentHashMap;
public class CacheWithLock {
private static final ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();
private static final Object lock = new Object(); // 用于本地锁
public static String getData(String key) {
String value = cache.get(key);
// 判断缓存是否为空
if (value == null) {
synchronized (lock) {
// 双重检查，防止重复加载
value = cache.get(key);
if (value == null) {
// 查询数据库
value = queryFromDatabase(key);
// 更新缓存
cache.put(key, value);
}
}
}
return value;
}
private static String queryFromDatabase(String key) {
// 模拟数据库查询
return "data_from_db_for_" + key;
}
}
02.手动过期
a.思路
对于热点数据不过期的策略，缓存不主动设置过期时间，而是定期使用异步任务来更新缓存的数据，保证数据的实时性
b.代码
import java.util.concurrent.*;
public class CacheWithHotData {
private static final ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();
// 初始化热点数据
static {
loadHotData(); // 初始加载热点数据
startUpdateTask(); // 启动定时更新任务
}
public static String getData(String key) {
return cache.get(key);
}
private static void loadHotData() {
// 将热点数据加载到缓存
cache.put("hotKey1", queryFromDatabase("hotKey1"));
cache.put("hotKey2", queryFromDatabase("hotKey2"));
// ... 其他热点数据
}
private static String queryFromDatabase(String key) {
// 模拟数据库查询
return "data_from_db_for_" + key;
}
private static void startUpdateTask() {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
// 定时更新任务
executor.scheduleAtFixedRate(() -> {
for (String key : cache.keySet()) {
String newValue = queryFromDatabase(key); // 查询最新数据
cache.put(key, newValue); // 更新缓存
System.out.println("Updated cache for key: " + key);
}
}, 0, 5, TimeUnit.MINUTES); // 每5分钟更新一次
}
}
1.6 缓存雪崩：给不同的Key的TTL添加随机值、限流与熔断机制、给业务添加多级缓存
00.缓存雪崩
a.定义
某一个时间点，由于【大量的缓存数据同时过期 或 缓存服务器突然宕机】了，导致所有的请求都落到了数据库上
b.方案1：缓存高可用
a.目标
确保缓存服务的稳定性，避免因缓存服务器故障导致的服务中断
b.方法
使用 Redis Sentinel 或 Redis Cluster 实现高可用性
通过主从复制和故障转移机制，确保即使某个节点故障，其他节点仍能提供服务
c.方案2：给不同的Key的TTL添加随机值
a.目标
避免大量缓存键在同一时间过期
b.方法
在设置过期时间时，添加随机值以打散过期时间
通过打散过期时间，减少同一时间大量缓存失效的风险
d.方案3：限流与熔断机制
a.目标
保护数据库，避免因大量请求导致的数据库崩溃
b.方法
设置数据库的请求限流策略，控制每秒请求数
使用熔断机制，在高负载时暂时拒绝部分请求
e.方案4：给业务添加多级缓存
a.目标
在分布式缓存失效时，减少对数据库的直接访问
确保关键数据在缓存失效时仍可快速访问
b.方案
本地缓存（Caffeine） + 分布式缓存（redis）
f.方案5：备份缓存
a.目标
确保关键数据在缓存失效时仍可快速访问
b.方法
对关键数据进行备份缓存，在主缓存不可用时切换到备用缓存
01.缓存高可用
a.定义
通过 Redis Sentinel 或 Redis Cluster 提供高可用性，确保即使某个缓存节点故障，其他节点仍能继续提供服务
b.原理
通过主从复制和故障转移机制，避免因单点故障导致的缓存雪崩
c.常用API
Redis Sentinel 和 Redis Cluster 的配置和管理命令
d.使用步骤
1.部署 Redis Sentinel 或 Redis Cluster
2.配置主从复制和故障转移
3.监控节点状态，确保高可用性
e.场景示例
# Redis Sentinel 配置示例
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 5000
sentinel failover-timeout mymaster 60000
sentinel parallel-syncs mymaster 1
02.随机过期时间
a.定义
通过为缓存键设置随机过期时间，避免大量缓存键在同一时间过期
b.原理
在设置TTL时添加随机值，打散过期时间，减少瞬时高负载对数据库的冲击
c.常用API
Redis的SET命令
d.使用步骤
1.设置基础TTL时间
2.在基础TTL上加一个随机值
3.使用Redis客户端设置Key和TTL
e.场景示例
import java.util.concurrent.ThreadLocalRandom;
public class CacheService {
private static final int BASE_TTL = 300; // 基础TTL时间，单位秒
public void setCache(String key, String value) {
int randomTTL = BASE_TTL + ThreadLocalRandom.current().nextInt(60); // 在基础时间上加0-60秒随机值
RedisClient.set(key, value, randomTTL); // Redis客户端设置key和TTL
}
}
03.限流与熔断机制
a.定义
通过限流和熔断机制，控制数据库的请求量，防止在缓存失效时数据库被过载
b.原理
使用限流算法（如Guava的RateLimiter）限制请求频率，防止短时间内大量请求对数据库造成压力
c.常用API
Guava的RateLimiter
d.使用步骤
1.创建RateLimiter实例，设置每秒允许的请求数
2.在请求处理时，使用RateLimiter进行限流
3.如果请求被限流，执行降级处理
e.场景示例
import com.google.common.util.concurrent.RateLimiter;
public class CacheWithRateLimiter {
private static final RateLimiter rateLimiter = RateLimiter.create(100); // 每秒100个请求
public String getDataWithFallback(String key) {
if (rateLimiter.tryAcquire()) {
String value = RedisClusterService.getCache(key);
if (value == null) {
// 缓存不存在，查询数据库
value = queryFromDatabase(key);
RedisClusterService.setCache(key, value); // 缓存新值
}
return value;
} else {
return fallbackData(); // 降级处理，返回默认数据或友好提示
}
}
private String queryFromDatabase(String key) {
// 数据库查询逻辑
return "db_data_for_" + key;
}
private String fallbackData() {
return "default_value"; // 降级数据
}
}
04.给业务添加多级缓存
a.定义
结合使用本地缓存和分布式缓存，形成多级缓存体系
b.原理
在本地缓存失效时，优先访问分布式缓存，最后访问数据库
c.常用API
Caffeine和Redis的API
d.使用步骤
1.在应用中集成本地缓存（如Caffeine）和分布式缓存（如Redis）
2.优先从本地缓存获取数据
3.本地缓存失效时，从分布式缓存获取数据
4.分布式缓存失效时，从数据库获取数据
e.场景示例
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.util.concurrent.TimeUnit;
public class MultiLevelCache {
private static final Cache<String, String> localCache = Caffeine.newBuilder()
.expireAfterWrite(5, TimeUnit.MINUTES) // 本地缓存过期时间
.maximumSize(1000)
.build();
public String getData(String key) {
// 先查本地缓存
String value = localCache.getIfPresent(key);
if (value == null) {
// 本地缓存不存在，查 Redis 缓存
value = RedisClusterService.getCache(key);
if (value == null) {
// Redis 缓存不存在，查询数据库
value = queryFromDatabase(key);
RedisClusterService.setCache(key, value); // 设置 Redis 缓存
}
localCache.put(key, value); // 更新本地缓存
}
return value;
}
private String queryFromDatabase(String key) {
// 模拟数据库查询
return "db_data_for_" + key;
}
}
05.备份缓存
a.定义
对关键数据进行备份缓存，确保在主缓存失效时仍能快速访问数据
b.原理
在多个Redis节点上存储相同的热点Key数据，分散请求负载
c.常用API
Redis的基本操作API，如SET和GET
d.使用步骤
1.在多个Redis实例上存储热点Key
2.在请求时随机选择一个Redis实例进行访问
e.场景示例
import redis
import random
# 连接多个 Redis 实例
redis_instances = [redis.Redis(host='localhost', port=6379),
redis.Redis(host='localhost', port=6380)]
# 备份数据到多个实例
for r in redis_instances:
r.set("hotKey", "value")
# 随机选择一个实例获取数据
selected_redis = random.choice(redis_instances)
value = selected_redis.get("hotKey")
1.7 缓存预热：手工操作、项目启动自加载、定时刷新缓存
00.缓存预热
a.定义
在系统上线或启动时，提前将一些预定义的数据加载到缓存中
这样可以避免在用户首次请求时，缓存未命中导致的性能问题
通过缓存预热，系统可以在上线后立即提供高效的服务，减少首次访问时的延迟
b.解决方案
1.直接写个缓存刷新页面，上线时手工操作一下
2.数据量不大，可以在项目启动的时候自动进行加载
3.定时刷新缓存
01.直接手动刷新
a.方法
上线时通过手动操作刷新缓存
b.实现
提供一个管理页面或接口，允许管理员在系统上线时手动触发缓存刷新
c.实现示例
public class CacheManager {
public void refreshCache() {
// 手动刷新缓存
List<String> keys = getKeysToPreload();
for (String key : keys) {
String value = queryFromDatabase(key);
RedisClient.set(key, value);
}
}
private List<String> getKeysToPreload() {
// 获取需要预热的Key列表
return Arrays.asList("key1", "key2", "key3");
}
private String queryFromDatabase(String key) {
// 模拟数据库查询
return "db_data_for_" + key;
}
}
02.项目启动时自动加载
a.方法
在项目启动时自动加载必要的数据到缓存
b.实现
在应用启动过程中，执行一段代码，将预定义的数据加载到缓存中
c.实现示例
import javax.annotation.PostConstruct;
public class ApplicationStartup {
@PostConstruct
public void preloadCache() {
// 在项目启动时预加载缓存
List<String> keys = getKeysToPreload();
for (String key : keys) {
String value = queryFromDatabase(key);
RedisClient.set(key, value);
}
}
private List<String> getKeysToPreload() {
// 获取需要预热的Key列表
return Arrays.asList("key1", "key2", "key3");
}
private String queryFromDatabase(String key) {
// 模拟数据库查询
return "db_data_for_" + key;
}
}
03.定时刷新缓存
a.方法
定期刷新缓存，确保缓存中的数据始终是最新的
b.实现
使用定时任务定期更新缓存中的数据
c.实现示例
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheduledCacheRefresher {
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
public void startCacheRefresh() {
scheduler.scheduleAtFixedRate(this::refreshCache, 0, 1, TimeUnit.HOURS);
}
private void refreshCache() {
// 定时刷新缓存
List<String> keys = getKeysToPreload();
for (String key : keys) {
String value = queryFromDatabase(key);
RedisClient.set(key, value);
}
}
private List<String> getKeysToPreload() {
// 获取需要预热的Key列表
return Arrays.asList("key1", "key2", "key3");
}
private String queryFromDatabase(String key) {
// 模拟数据库查询
return "db_data_for_" + key;
}
}
2 多级缓存
2.1 汇总
01.浏览器缓存
它的实现主要依靠 HTTP 协议中的缓存机制
当浏览器第一次请求一个资源时
服务器会将该资源的相关缓存规则（如 Cache-Control、Expires 等）一同返回给客户端
浏览器会根据这些规则来判断是否需要缓存该资源以及该资源的有效期
02.Nginx缓存
在Nginx中配置中开启缓存功能
03.分布式缓存
所有系统调用的中间件都是分布式缓存，如Redis、MemCached等
04.本地缓存
JVM层面，单系统运行期间在内存中产生的缓存，例如Caffeine、Google Guava等
2.2 HTTP缓存
01.概况
a.定义
HTTP 缓存通过在客户端（如浏览器）存储已获取的资源副本，减少重复请求，提高访问速度，优化用户体验
并降低服务器负载。即使在离线状态，已缓存的资源仍可访问
b.缓存适用范围
静态资源（适合缓存）：HTML、CSS、JS、图片、字体等
动态资源（一般不缓存）：如购物车、用户数据、订单信息，因数据实时变化，缓存可能导致信息不一致
c.如何优化缓存？
强缓存（Cache-Control / Expires）：优先使用本地缓存，减少请求
协商缓存（ETag / Last-Modified）：资源更新时才重新下载，避免使用过期数据
动态资源优化：可采用短时缓存 + 协商缓存，兼顾性能与数据实时性
d.静态资源-缓存策略
a.HTML
是否缓存：可以缓存(但建议短时缓存)
推荐缓存策略：Cache-Control: no-cache, must-revalidate,避免加载过期页面
b.CSS/JS
是否缓存：建议缓存
推荐缓存策略：Cache-Control: max-age=31536000, immutable 并使用文件 hash 版本号
c.图片、字体
是否缓存：强烈建议缓存
推荐缓存策略：Cache-Control: max-age=31536000 并使用 CDN
d.API数据
是否缓存：不建议缓存
推荐缓存策略：Cache-Control: no-store 确保数据实时性
e.缓存策略-前端实现
a.Cache-Control
类型：强缓存
控制方式：fetch 头部或 <meta>
适用场景：控制缓存时间、是否使用缓存
b.Expires
类型：强缓存
控制方式：<meta> 或 HTTP 头部
适用场景：指定资源过期时间
c.Last-Modified / If-Modified-Since
类型：协商缓存
控制方式：fetch 请求头
适用场景：资源基于修改时间更新
d.ETag / If-None-Match
类型：协商缓存
控制方式：fetch 请求头
适用场景：资源基于唯一标识符更新
02.静态资源-缓存策略
a.强缓存（Strong Cache）
a.定义
通过在 HTTP 响应头中设置 Cache-Control 和 Expires，让浏览器在缓存有效期内直接使用本地缓存
而不向服务器发送请求，从而提高加载速度和减少服务器压力
b.强缓存的实现方式
a.Cache-Control（优先级高，HTTP/1.1）
a.定义
Cache-Control 是现代浏览器主要使用的缓存控制字段，支持更精细的缓存管理
b.图示
指令 作用
max-age=3600 设置缓存有效时间（秒），这里是1小时
no-cache 每次请求都要向服务器确认是否有新版本 (走协商缓存)
no-store 不缓存，每次都从服务器获取
public 允许任何人（包括CDN）缓存
private 只能被客户端缓存，CDN不能缓存
immutable 资源不会更改，即使F5也不会重新请求
b.Expires（HTTP/1.0，已被 Cache-Control 取代）
a.定义
Expires 指定资源的过期时间（绝对时间） ，浏览器在过期时间前直接使用缓存。
b.问题
依赖客户端时间，如果用户本地时间不准确，可能导致缓存异常
被 Cache-Control 覆盖，现代浏览器一般不使用它
c.示例
Expires: Wed, 21 Oct 2025 07:28:00 GMT
含义：资源在 2025 年 10 月 21 日 07:28:00 之前有效
c.建议
推荐优先使用 Cache-Control，因为它更灵活和可靠！
c.强缓存的特点
不会向服务器发送请求，直接从本地缓存读取，加载速度极快
适用于静态资源（如 CSS、JS、图片、字体等）
不能及时更新，如果资源变更但仍在缓存期内，客户端不会请求最新版本
d.如何解决强缓存更新问题？
a.问题
如果资源内容变更了，但缓存仍然有效，用户可能看不到最新版本
b.解决1：文件名加 Hash
在文件名中添加 hash，确保文件变更时 URL 也变
app.js → app.123abc.js
---------------------------------------------------------------------------------------------
优点：
旧缓存不会影响新文件
适用于前端构建工具（如 Webpack）
c.解决2：版本号参数
在 URL 后加 ?v=20250312
https://example.com/app.js?v=20250312
-------------------------------------------------------------------------------------------------
优点：
服务器可以控制更新策略
可能会被部分 CDN 忽略
e.什么时候用强缓存？
资源类型 适用缓存策略
CSS/JS Cache-Control: max-age=31536000, immutable，并加hash
图片/字体 Cache-Control:max-age=31536000, public，并加hash
API数据 不使用强缓存，改用协商缓存（ETag／Last-Modified）
f.总结
Cache-Control: max-age=xxx 是最常用的强缓存方式，比 Expires 更可靠
强缓存不会请求服务器，适用于静态资源
避免缓存问题，建议使用 hash 或版本号来控制文件更新
b.协商缓存
a.定义
协商缓存（also known as Conditional Caching）是 HTTP 缓存的一种机制，能够有效提升网页的加载速度
减少服务器压力。它的核心是通过使用请求和响应中的特定头部信息来确保资源的有效性
从而避免不必要的资源下载
b.常见的头部
a.Last-Modified / If-Modified-Since
a.说明
Last-Modified 和 If-Modified-Since 是 HTTP 协商缓存机制中非常关键的两个头部字段
它们用来协商是否需要重新从服务器获取资源，避免不必要的重复下载
b.第一次请求
客户端请求资源，服务器返回资源并带有 Last-Modified 字段
c.第二次请求
客户端带上 If-Modified-Since 头部，服务器根据该时间检查资源是否修改
如果资源没有变化，服务器返回 304 Not Modified，客户端继续使用缓存
如果资源有变化，服务器返回新的资源和 Last-Modified 时间，客户端更新缓存
d.注意事项
a.精度限制
Last-Modified 的精度通常只有到秒，而有些服务器的文件系统可能支持毫秒级的修改时间
这样可能会导致某些情况下无法准确判断资源是否修改
b.效率问题
对于频繁更新的资源（比如动态生成的内容）
Last-Modified 和 If-Modified-Since 的机制可能不太适用
可能需要其他方式来管理缓存（比如 ETag）
b.ETag / If-None-Match
a.说明
ETag 和 If-None-Match 是另一种常用的 HTTP 协商缓存机制，它们相比 Last-Modified / If-Modified-Since 更为精确
因为它们可以通过一个独特的标识符来判断资源是否发生了变化
b.第一次请求
客户端请求资源，服务器返回资源并带上 ETag 字段
c.第二次请求
客户端带上 If-None-Match 头部，服务器根据该 ETag 值检查资源是否修改
如果资源没有变化，服务器返回 304 Not Modified，客户端继续使用缓存
如果资源有变化，服务器返回新的资源和 ETag，客户端更新缓存
d.优点与 Last-Modified 的对比
a.精确度更高
ETag 可以比 Last-Modified 更精确地标识资源的变化
因为它是一个自定义的标识符，通常可以精确到文件的每一个微小变化
b.无时间依赖
ETag 不依赖于时间戳，因此它可以避免 Last-Modified 精度不够的情况
尤其是当文件的修改时间没有变化但内容发生变化时，ETag 可以有效地判断出资源是否变化
04.缓存策略-前端实现
a.定义
在前端代码中，可以使用 JavaScript 来检查和控制 HTTP 缓存行为
主要涉及 Cache-Control、Expires、Last-Modified / If-Modified-Since、ETag / If-None-Match 这些 HTTP 头部
b.Cache-Control & Expires
a.说明
Cache-Control 和 Expires 主要用于强缓存（不需要请求服务器，直接使用本地缓存）
b.方式1：通过 <meta> 标签控制缓存
<meta http-equiv="Cache-Control" content="max-age=3600, must-revalidate">
<meta http-equiv="Expires" content="Wed, 21 Oct 2025 07:28:00 GMT">
-------------------------------------------------------------------------------------------------
max-age=3600：设置缓存有效期为 3600 秒（1 小时）
must-revalidate：缓存过期后必须重新向服务器验证资源
Expires：指定资源过期时间（但 Cache-Control 的 max-age 优先级更高）
c.方式2：前端发起请求时手动设置 Cache-Control
在 fetch 或 XMLHttpRequest 请求中可以手动控制 Cache-Control：
fetch('/api/data', {
method: 'GET',
headers: {
'Cache-Control': 'no-cache', // 禁用强缓存，每次都重新请求
}
})
.then(response => response.json())
.then(data => console.log(data));
d.常见的 Cache-Control 值
no-cache：每次请求都向服务器验证资源是否更新（走协商缓存）
no-store：完全不缓存，每次都重新下载资源
max-age=3600：缓存 1 小时
private：只能被浏览器缓存，代理服务器不能缓存
public：浏览器和代理服务器都可以缓存
c.Last-Modified / If-Modified-Since
a.说明
Last-Modified 和 If-Modified-Since 用于协商缓存，前端可以通过 fetch 或 XMLHttpRequest 查看它们
b.方式1：检查 Last-Modified
a.说明
前端可以发送请求，并查看服务器返回的 Last-Modified：
b.代码
fetch('/api/data', {
method: 'GET'
})
.then(response => {
console.log('Last-Modified:', response.headers.get('Last-Modified'));
return response.json();
})
.then(data => console.log(data));
c.方式2：发送 If-Modified-Since
a.说明
客户端手动带上 If-Modified-Since 头部，询问服务器资源是否更新：
b.代码
fetch('/api/data', {
method: 'GET',
headers: {
'If-Modified-Since': 'Wed, 10 Mar 2025 10:00:00 GMT'
}
})
.then(response => {
if (response.status === 304) {
console.log('资源未更新，使用缓存');
} else {
return response.json();
}
})
.then(data => console.log(data));
c.说明
如果资源未修改，服务器返回 304 Not Modified，浏览器使用本地缓存数据
如果资源已修改，服务器返回 200 OK 并带回最新的资源
d.方式3：ETag / If-None-Match
a.说明
ETag 是比 Last-Modified 更精确的缓存校验方式，前端可以查看 ETag，并在下次请求时带上 If-None-Match
b.方式1：获取ETag
fetch('/api/data')
.then(response => {
console.log('ETag:', response.headers.get('ETag'));
return response.json();
})
.then(data => console.log(data));
c.方式2：发送If-None-Match
fetch('/api/data', {
method: 'GET',
headers: {
'If-None-Match': '"abc123"'
}
})
.then(response => {
if (response.status === 304) {
console.log('资源未更新，使用缓存');
} else {
return response.json();
}
})
.then(data => console.log(data));
---------------------------------------------------------------------------------------------
如果 ETag 匹配，服务器返回 304 Not Modified，前端使用缓存数据
如果 ETag 不匹配，服务器返回 200 OK，并带回最新资源
e.方式4：结合 Cache-Control、ETag 和 Last-Modified
a.说明
如果要完整控制缓存，可以结合 Cache-Control、ETag 和 Last-Modified
b.代码
fetch('/api/data', {
method: 'GET',
headers: {
'Cache-Control': 'no-cache', // 先检查服务器是否有更新
'If-None-Match': '"abc123"', // 带上上次缓存的 ETag
'If-Modified-Since': 'Wed, 10 Mar 2025 10:00:00 GMT' // 带上上次的 Last-Modified
}
})
.then(response => {
if (response.status === 304) {
console.log('资源未修改，使用缓存');
} else {
console.log('资源已更新，获取新数据');
return response.json();
}
})
.then(data => console.log(data));
2.3 浏览器缓存
01.开启浏览器缓存
a.说明
可以使用 HttpServletResponse 对象来设置与缓存相关的响应头，以开启浏览器的缓存功能
b.配置 Cache-Control
Cache-Control 是 HTTP/1.1 中用于控制缓存策略的主要方式
它可以设置多个指令，如 max-age(定义资源的最大存活时间，单位秒)、no-cache（要求重新验证）
public（指示可以被任何缓存区缓存）、private（只能被单个用户私有缓存存储）等
c.配置 Expires
设置一个绝对的过期时间，超过这个时间点后浏览器将不再使用缓存的内容而向服务器请求新的资源
d.配置 ETag
ETag（实体标签）一种验证机制，它为每个版本的资源生成一个唯一标识符
当客户端发起请求时，会携带上先前接收到的 ETag，服务器根据 ETag 判断资源是否已更新
若未更新则返回 304 Not Modified 状态码，通知浏览器继续使用本地缓存
e.配置 Last-Modified
指定资源最后修改的时间戳，浏览器下次请求时会带上 If-Modified-Since 头
服务器对比时间戳决定是否返回新内容或发送 304 状态码
2.4 Nginx缓存
01.开启Nginx缓存
a.定义缓存配置
在 Nginx 配置中定义一个缓存路径和配置，通过 proxy_cache_path 指令完成
proxy_cache_path /path/to/cache levels=1:2 keys_zone=my_cache:10m max_size=10g inactive=60m use_temp_path=off;
-----------------------------------------------------------------------------------------------------
其中：
/path/to/cache：这是缓存文件的存放路径
levels=1:2：定义缓存目录的层级结构
keys_zone=my_cache:10m：定义一个名为 my_cache 的共享内存区域，大小为 10MB
max_size=10g：设置缓存的最大大小为 10GB
inactive=60m：如果在 60 分钟内没有被访问，缓存将被清理
use_temp_path=off：避免在文件系统中进行不必要的数据拷贝
b.启用缓存
在 server 或 location 块中，使用 proxy_cache 指令来启用缓存，并指定要使用的 keys zone
c.设置缓存有效期
使用 proxy_cache_valid 指令来设置哪些响应码的缓存时间
d.配置反向代理
确保你已经配置了反向代理，以便 Nginx 可以将请求转发到后端服务器
e.重新加载配置
保存并关闭 Nginx 配置文件后，使用 nginx -s reload 命令重新加载配置，使更改生效
2.5 分布式缓存
01.使用分布式缓存
a.添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
b.配置
application.yml 文件中配置 Redis 的相关信息
c.启动缓存
在 Spring Boot 主类或者配置类上添加 @EnableCaching 注解来启用缓存
d.使用缓存
a.@Cacheable注解来缓存方法的返回值
@Cacheable(value = "users", key = "#id")
public User getUserById(Long id) {
return userRepository.findById(id).orElse(null);
}
b.@CacheEvict注解来删除缓存
@CacheEvict(value = "users", key = "#id")
public void deleteUser(Long id) {
userRepository.deleteById(id);
}
c.@CachePut注解来更新缓存
@CachePut(value = "users", key = "#user.id")
public User updateUser(User user) {
return userRepository.save(user);
}
2.6 本地缓存
01.使用本地缓存
a.添加依赖
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
<version>2.8.8</version>
</dependency>
b.配置
spring.cache.caffeine.spec=initialCapacity=100,maximumSize=500,expireAfterWrite=10m
initialCapacity：初始容器容量
maximumSize：最大容量
expireAfterWrite：写入缓存后 N 长时间后过期
c.自定义 Caffeine 配置类（可选步骤）
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.interceptor.CacheResolver;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class CaffeineCacheConfig extends CachingConfigurerSupport {
@Bean
public CacheManager cacheManager() {
Caffeine<Object, Object> caffeine = Caffeine.newBuilder()
.initialCapacity(100)
.maximumSize(1000)
.expireAfterWrite(10, TimeUnit.SECONDS) // 10 秒后过期
.recordStats(); // 记录缓存统计信息
return new CaffeineCacheManager("default", caffeine::build);
}
@Override
public CacheResolver cacheResolver() {
// 自定义缓存解析器（如果需要）
// ...
return super.cacheResolver();
}
}
d.开启缓存
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
@EnableCaching
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
e.使用注解进行缓存操作
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
@Service
public class UserService {
@Cacheable(value = "users", key = "#id") // 假设我们有一个名为"users"的缓存区域
public User getUserById(Long id) {
// 这里是真实的数据库查询或其他耗时操作
return userRepository.findById(id).orElse(null);
}
@CacheEvict(value = "users", key = "#user.id")
public void updateUser(User user) {
userRepository.save(user);
}
}
2.7 缓存策略：6种
00.汇总
a.Read Through（读取穿透）
在缓存缺失时从主存取数据并更新缓存，适合读多写少场景
b.Cache Aside（缓存旁路）
由应用程序自己管理缓存，提供灵活性，适合复杂业务逻辑
c.Write Through（直写）
实时在缓存和主存同步写数据，保证一致性但写入稍慢
d.Write Around（绕过缓存写）
写操作跳过缓存，减少写入负担，适合写频繁但读取少的情况
e.Write Back（回写）
先写缓存，后续批量写入主存，提升写性能，但最终一致性难以保证。选择应根据性能需求和一致性要求平衡
f.Refresh-ahead（预刷新）
预先载入未来可能会被访问的数据，减少未来请求的响应时间，适合在可预测的访问场景下
01.Read Through
a.定义
Read Through缓存策略是一种同步读取策略，当应用程序需要读取数据时
首先查询缓存，如果缓存中没有所需的数据（即缓存未命中）
缓存系统会自动从底层数据存储（如数据库）中读取数据，并将其存入缓存中，然后返回给应用程序
b.优点
简化应用逻辑：应用程序不需要处理缓存未命中的情况，缓存系统自动处理数据加载
数据一致性：由于缓存系统直接从数据源读取数据，确保了缓存中的数据是最新的
c.缺点
初次访问延迟：如果缓存未命中，读取操作会有一定的延迟，因为需要从底层存储中获取数据
缓存填充开销：每次缓存未命中时，都会导致底层存储的访问，这可能会增加系统的负载
d.适用场景
适用于读操作频繁且读一致性要求较高的场景
在数据更新频率较低的情况下，Read Through可以有效减少应用程序的复杂性
02.Cache Aside
a.定义
Cache Aside（也称为Lazy Loading或Lazy Caching）策略要求应用程序显式地管理缓存
应用程序首先检查缓存，如果未命中，则从底层数据存储中读取数据，并将其放入缓存中供下次使用
b.优点
灵活性高：应用程序可以根据具体需求决定何时加载和更新缓存
缓存命中率高：由于应用程序负责缓存管理，可以更好地优化缓存使用
c.缺点
复杂性增加：应用程序需要处理缓存未命中的逻辑以及缓存的更新和失效
潜在的数据不一致性：如果数据更新后未及时刷新缓存，可能会导致不一致的数据
d.适用场景
适用于读多写少且对读性能要求高的场景
应用程序可以容忍一定程度的数据不一致性
03.Write Through
a.定义
Write Through策略是一种同步写入策略，当应用程序对数据进行更新时
数据会同时写入缓存和底层数据存储，这确保了缓存和数据存储的一致性
b.优点
数据一致性强：由于每次写操作都会更新缓存和数据存储，因此可以保证它们之间的数据一致性
简单的实现：不需要复杂的缓存失效机制
c.缺点
写操作延迟：每次写操作都需要更新底层存储，这可能导致写操作的延迟增加
写入开销大：频繁的写操作可能会导致底层存储的负载增加
d.适用场景
适用于数据一致性要求高且写操作相对较少的场景
在需要确保每次写入操作后的数据一致性时，Write Through是一种有效的策略
04.Write Around
a.定义
Write Around策略是一种变体的写入策略，当数据被更新时，仅更新底层数据存储
而不更新缓存，缓存的数据只有在被读取时才会更新
b.优点
降低写入延迟：避免了每次写操作都更新缓存，从而降低了写入延迟
减轻缓存压力：写操作不会直接影响缓存，可以减少缓存的更新频率
c.缺点
缓存未命中率高：由于写入操作不更新缓存，可能导致后续读取操作未命中缓存
潜在的数据不一致性：如果缓存中的数据在更新后没有及时刷新，可能会导致数据不一致
d.适用场景
适用于写操作频繁且读操作可以容忍一定延迟的场景
在需要减少写操作对缓存影响的情况下，Write Around是一种可行的策略
05.Write Back
a.定义
Write Back策略是一种异步写入策略，当应用程序更新数据时，仅更新缓存
缓存中的数据会在一段时间后（或满足特定条件时）批量写入底层数据存储
b.优点
写操作延迟低：由于写操作仅更新缓存，写入延迟较低
提高系统吞吐量：批量写入可以减少对底层存储的访问次数，提高系统的整体吞吐量
c.缺点
数据一致性风险：由于底层存储更新滞后，可能导致数据不一致
数据丢失风险：如果缓存数据在写入底层存储之前丢失（例如系统故障），可能导致数据丢失
d.适用场景
适用于写操作频繁且对写入性能要求高的场景
在可以接受一定程度的数据延迟和不一致性的情况下，Write Back是一种高效的策略
06.Refresh-ahead（预刷新）
a.定义
Refresh-ahead 是一种缓存预取策略，旨在提高系统的响应速度，尤其是在可预测的访问场景下
b.优点
减小读取延迟：通过提前加载数据降低未来请求的响应时间，特别是减少了缓存未命中的概率
提升性能：由于数据被提前载入，系统在实际请求到达时能立即提供服务，减少瓶颈
c.缺点
资源浪费：如果预测不准确，预载入的数据可能根本不会被访问，这将导致内存和IO资源的浪费
处理复杂性增加：需要进行访问模式的监控与分析，对系统增加了额外的复杂度
d.使用场景
时间序列数据：例如股票行情、传感器读数等具有强时间依赖或者逐步递增的数据流场景
顺序读取：如果系统知道存在将要顺序访问的数据块，可以提前将数据加载到缓存
高延迟系统：例如大规模分布式系统或移动网络应用，提前刷新可以减少等待时间和网络延迟
3 缓存和MySQL一致性
3.1 不一致原因
01.并发问题
在高并发场景下，可能会读取到旧的数据库数据并更新到缓存中
02.事务问题
缓存和数据库的操作不在同一个事务中，可能导致一个操作成功而另一个失败
3.2 更新方法：3种
01.低一致性
a.使用Redis自带的内存淘汰机制
适用于对一致性要求不高的场景，通过自动淘汰机制来管理缓存数据
02.高一致性/实时一致性
a.主动更新，并以超时剔除作为兜底方案
读操作：缓存命中则返回，未命中则查询数据库并写入缓存
写操作：先更新MySQL，再删除Redis，确保操作的原子 推荐，更新数据库比删除缓存慢，先更新数据库可以减少脏数据的可能性
b.消息队列
引入消息队列以确保缓存被删除
c.数据库订阅
结合消息队列以确保缓存被删除
d.延时双删
防止脏数据
e.设置缓存过期时间
作为兜底方案
03.最终一致性
先更新MySQL，通过Binlog异步更新Redis 通过监听数据库的Binlog日志，异步更新缓存
04.不推荐或不属于上述两类的方案
先更新Redis，再更新MySQL 不推荐，因为如果数据库写入失败，会导致数据不一致
先更新MySQL，再更新Redis 不推荐，虽然适用于一致性要求不高的项目，但不属于严格的实时或最终一致性方案
先删除Redis，再更新MySQL 不推荐，因为可能导致缓存中出现脏数据
先删除Redis，再更新MySQL，再删除Redis 不推荐，又复杂
3.3 店铺：先更新MySQL，再删除Redis
00.汇总
查询操作：采用【缓存穿透】策略，【先查缓存，缓存未命中时查数据库并更新缓存】
更新操作：采用【先更新数据库再删除缓存】策略，使用【高一致性/实时一致性】
01.查询店铺逻辑
a.思路
a.从Redis查询缓存
使用店铺ID构建缓存键，从Redis中获取缓存数据
b.判断缓存是否存在
如果缓存存在，直接将数据转换为Shop对象并返回
如果缓存不存在，继续执行数据库查询
c.从数据库查询
根据店铺ID从数据库中查询店铺信息
d.判断数据库查询结果
如果数据库中不存在该店铺，返回错误信息
如果存在，将结果写入Redis缓存，并设置超时时间（例如30分钟）
e.返回结果
返回查询到的店铺信息
b.代码
@Override
public Result queryById(Long id) {
String key = "cache:shop:" + id;
// 1. 从Redis查询商铺缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
// 2. 判断是否存在
if (Strutil.isNotBlank(shopJson)) {
// 3. 存在，直接返回
Shop shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
// 4. 不存在，根据ID查询数据库
Shop shop = getById(id);
// 5. 不存在，返回错误
if (shop == null) {
return Result.fail("店铺不存在！");
}
// 6. 存在，写入Redis
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonstr(shop), 30L, TimeUnit.MINUTES);
// 7. 返回
return Result.ok(shop);
}
02.修改店铺逻辑
a.思路
a.检查店铺ID
确保店铺ID不为空，否则返回错误信息
b.更新数据库
使用事务更新数据库中的店铺信息
c.删除缓存
删除Redis中对应店铺ID的缓存，以确保下次查询时获取最新数据
d.返回结果
返回更新成功的信息
b.代码
@Override
@Transactional
public Result update(Shop shop) {
Long id = shop.getId();
if (id == null) {
return Result.fail("店铺id不能为空");
}
// 1. 更新数据库
updateById(shop);
// 2. 删除缓存
stringRedisTemplate.delete("cache:shop:" + id);
return Result.ok();
}
4 本地缓存和分布式缓存一致
4.1 汇总：3个
01.MQ广播模式
当数据修改时向MQ发送消息，节点监听并消费消息，删除本地缓存，达到最终一致性
02.Redis发布/订阅
如果使用Redis作为辅助，可以利用其发布/订阅功能，实现跨实例的缓存更新通知
03.Canal + MQ
订阅Mysql的Binlog日志，再操作缓存；当发生变化时向MQ发送消息，进而也实现数据一致性
4.2 本地缓存
01.常见信息
a.为什么使用本地缓存
本地缓存基于本地环境的内存，访问速度非常快，对于一些变更频率低、实时性要求低的数据，可以放在本地缓存中，提升访问速度
使用本地缓存能够减少和Redis类的远程缓存间的数据交互，减少网络I/O开销，降低这一过程中在网络通信上的耗时
b.设计一个本地缓存
存储，并可以读、写;
原子操作(线程安全)，如ConcurrentHashMap
可以设置缓存的最大限制;
超过最大限制有对应淘汰策略，如LRU、LFU
过期时间淘汰，如定时、懒式、定期;
持久化
统计监控
c.如何提高本地缓存命中率
合理设置缓存粒度
数据预热和懒加载
数据压缩和序列化优化
缓存穿透和击穿防护
动态调整缓存策略
使用合适的缓存策略：默认采用基于访问频率的策略
02.方案选型
a.使用ConcurrentHashMap实现本地缓存
线程安全的ConcurrentHashMap，但是要实现缓存，还需要考虑淘汰、最大限制、缓存过期时间淘汰等等功能
优点：实现简单，不需要引入第三方包，比较适合一些简单的业务场景
缺点：如果需要更多的特性，需要定制化开发，成本会比较高，并且稳定性和可靠性也难以保障
b.Ehcache
类型：本地缓存
数据一致性：应用逻辑维护
性能：高
易用性：高
功能特性：提供丰富的本地缓存策略，支持离线和堆外缓存
适用场景：单应用，需要快速访问且数据量不是分布式共享的场景
c.Caffeine
类型：本地缓存
数据一致性：应用逻辑维护
性能：非常高
易用性：高
功能特性：基于权重的驱逐策略，高性能
适用场景：高性能要求的场景，如高频读写的小数据集
d.Spring Cache
类型：缓存抽象层
数据一致性：依赖底层实现
性能：依赖底层实现
易用性：高
功能特性：缓存技术无关的编程模型，易于切换底层缓存实现
适用场景：需要灵活切换或结合多种缓存技术的场景
e.Redis
类型：分布式缓存
数据一致性：高中至高
性能：高
易用性：中
功能特性：支持多种数据结构，持久化，主从复制和哨兵系统
适用场景：需要数据共享，高并发，分布式场景
f.J2Cache
类型：结合本地与分布式
数据一致性：较高
性能：高中至高
易用性：中
功能特性：一级缓存与二级缓存结合，支持 Redis、Ehcache 等多种二级缓存后端
适用场景：需要结合本地快速访问和分布式数据共享的场景
g.Memcached
类型：分布式缓存
数据一致性：中
性能：高
易用性：高
功能特性：简单的键值存储，不支持持久化
适用场景：高速缓存临时数据，如会话存储，不需要持久化的场景
h.Guava Cache
类型：本地缓存
数据一致性：应用逻辑维护
性能：高
易用性：高
功能特性：易于使用的本地缓存，支持多种缓存过期策略
适用场景：适用于单个应用内需要快速访问的数据缓存，不需分布式数据共享的场景
4.3 本地缓存：Caffeine
00.汇总
概况
数据加载
驱除策略
缓存统计
核心代码
Caffeine本地缓存
Caffeine作为SpringCache实现
Redis+Caffeine实现两级缓存
01.概况
a.Caffeine是什么？
Caffeine是一个基于Java8开发的提供了近乎最佳命中率的高性能的缓存库，也是SpringBoot内置的本地缓存实现
b.Caffeine提供的特性
自动加载条目到缓存中，可选异步方式
可以基于大小剔除
可以设置过期时间，时间可以从上次访问或上次写入开始计算
异步刷新
keys自动包装在弱引用中
values自动包装在弱引用或软引用中
条目剔除通知
缓存访问统计
c.核心类和参数
核心工具类：Caffeine是创建高性能缓存的基类
核心参数：
maximumSize：缓存最大
maximumWeight：缓存最大权重，权重和最大值不能同时设
initialCapacity：缓存初始容
expireAfterWriteNanos：在写入多少纳秒没更新后过
expireAfterAccessNanos：在访问多少纳秒没更新后过
refreshAfterWriteNanos：写入多少纳秒没更新后更
d.@Cacheable、@CachePut、@CacheEvict
a.@Cacheable
a.说明
用于声明一个方法的返回值是可以被缓存的
当方法被调用时，Spring Cache 会先检查缓存中是否存在相应的数据
如果存在，则直接返回缓存中的数据，避免重复执行方法；如果不存在，则执行方法并将返回值存入缓存中
b.代码
@Cacheable(value = "users", key = "#id")
public User getUserById(String id) {
// 模拟从数据库中获取用户信息
System.out.println("Fetching user from database: " + id);
return new User(id, "User Name " + id);
}
b.@CachePut
a.说明
用于更新缓存中的数据
与 @Cacheable 不同，@CachePut 注解的方法总是会执行，并将返回值更新到缓存中
无论缓存中是否存在相应的数据，该方法都会执行，并将新的数据存入缓存中（如果缓存中已存在数据，则覆盖它）
b.代码
@CachePut(value = "users", key = "#user.id")
public User updateUser(User user) {
// 模拟更新数据库中的用户信息
System.out.println("Updating user in database: " + user.getId());
// 假设更新成功
return user;
}
c.@CacheEvict
a.说明
用于删除缓存中的数据。当方法被调用时，指定的缓存项将被删除。这可以用于清除旧数据或使缓存项失效
b.代码
@CacheEvict(value = "users", key = "#id")
public void deleteUser(String id) {
// 模拟从数据库中删除用户信息
System.out.println("Deleting user from database: " + id);
}
// 清除整个缓存，而不仅仅是特定的条目
@CacheEvict(value = "users", allEntries = true)
public void clearAllUsersCache() {
System.out.println("Clearing all users cache");
}
e.Caffeine中的时间轮
a.定义
在 Caffeine 中，时间轮用于管理缓存项的过期
Caffeine 通过维护一个时间轮来跟踪每个缓存项的过期时间，并在适当的时候清除过期的缓存项
b.原理
时间槽：时间轮由多个时间槽组成，每个槽代表一个固定的时间间隔
指针：时间轮有一个指针，指向当前的时间槽。指针随着时间的推移而移动
任务调度：当指针移动到一个新的时间槽时，时间轮会检查该槽中的缓存项，并清除已过期的项
c.优势
高效性：时间轮可以在常数时间内管理大量的过期任务，适合高并发场景
低开销：相比于传统的定时器，时间轮的内存和计算开销较低
d.使用
虽然 Caffeine 的时间轮机制是内部实现细节，用户不需要直接与时间轮交互
但可以通过配置缓存的过期策略来间接使用时间轮
e.示例
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.util.concurrent.TimeUnit;
public class CaffeineExample {
public static void main(String[] args) {
// 创建一个 Caffeine 缓存实例，设置过期时间为 5 分钟
Cache<String, String> cache = Caffeine.newBuilder()
.expireAfterWrite(5, TimeUnit.MINUTES)
.maximumSize(100)
.build();
// 添加缓存项
cache.put("key1", "value1");
// 获取缓存项
String value = cache.getIfPresent("key1");
System.out.println("Cached Value: " + value);
// 等待超过过期时间后，缓存项将被自动清除
}
}
02.数据加载
a.Caffeine提供的缓存添加策略
c.手动加载
public static void demo() {
Cache<String, String> cache =
Caffeine.newBuilder()
.expireAfterAccess(Duration.ofMinutes(1))
.maximumSize(100)
.recordStats()
.build();
// 插入数据
cache.put("a", "a");
// 查询某个key，如果没有返回空
String a = cache.getIfPresent("a");
System.out.println(a);
// 查找缓存，如果缓存不存在则生成缓存元素, 如果无法生成则返回null
String b = cache.get("b", k -> {
System.out.println("begin query ..." + Thread.currentThread().getName());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("end query ...");
return UUID.randomUUID().toString();
});
System.out.println(b);
// 移除一个缓存元素
cache.invalidate("a");
}
b.自动加载
public static void demo() {
LoadingCache<String, String> loadingCache = Caffeine.newBuilder()
.maximumSize(100)
.expireAfterWrite(10, TimeUnit.MINUTES)
.build(new CacheLoader() {
@Nullable
@Override
public Object load(@NonNull Object key) throws Exception {
return createExpensiveValue();
}
@Override
public @NonNull Map loadAll(@NonNull Iterable keys) throws Exception {
if (keys == null) {
return Collections.emptyMap();
}
Map<String, String> map = new HashMap<>();
for (Object key : keys) {
map.put((String) key, createExpensiveValue());
}
return map;
}
});
// 查找缓存，如果缓存不存在则生成缓存元素, 如果无法生成则返回null
String a = loadingCache.get("a");
System.out.println(a);
// 批量查找缓存，如果缓存不存在则生成缓存元素
Set<String> keys = new HashSet<>();
keys.add("a");
keys.add("b");
Map<String, String> allValues = loadingCache.getAll(keys);
System.out.println(allValues);
}
private static String createExpensiveValue() {
{
System.out.println("begin query ..." + Thread.currentThread().getName());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("end query ...");
return UUID.randomUUID().toString();
}
}
c.手动异步加载
public static void demo() throws ExecutionException, InterruptedException {
AsyncCache<String,String> asyncCache = Caffeine.newBuilder()
.maximumSize(100)
.buildAsync();
// 添加或者更新一个缓存元素
asyncCache.put("a",CompletableFuture.completedFuture("a"));
// 查找一个缓存元素， 没有查找到的时候返回null
CompletableFuture<String> a = asyncCache.getIfPresent("a");
System.out.println(a.get());
// 查找缓存元素，如果不存在，则异步生成
CompletableFuture<String> completableFuture = asyncCache.get("b", k ->createExpensiveValue("b"));
System.out.println(completableFuture.get());
// 移除一个缓存元素
asyncCache.synchronous().invalidate("a");
System.out.println(asyncCache.getIfPresent("a"));
}
private static String createExpensiveValue(String key) {
{
System.out.println("begin query ..." + Thread.currentThread().getName());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("end query ...");
return UUID.randomUUID().toString();
}
}
d.自动异步加载
public static void demo() throws ExecutionException, InterruptedException {
AsyncLoadingCache<String, String> cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterWrite(10, TimeUnit.MINUTES)
// 你可以选择: 去异步的封装一段同步操作来生成缓存元素
//.buildAsync(key -> createExpensiveValue(key));
// 你也可以选择: 构建一个异步缓存元素操作并返回一个future
.buildAsync((key, executor) ->createExpensiveValueAsync(key, executor));
// 查找缓存元素，如果其不存在，将会异步进行生成
CompletableFuture<String> a = cache.get("a");
System.out.println(a.get());
// 批量查找缓存元素，如果其不存在，将会异步进行生成
Set<String> keys = new HashSet<>();
keys.add("a");
keys.add("b");
CompletableFuture<Map<String, String>> values = cache.getAll(keys);
System.out.println(values.get());
}
private static String createExpensiveValue(String key) {
{
System.out.println("begin query ..." + Thread.currentThread().getName());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
System.out.println("end query ...");
return UUID.randomUUID().toString();
}
}
private static CompletableFuture<String> createExpensiveValueAsync(String key, Executor executor) {
{
System.out.println("begin query ..." + Thread.currentThread().getName());
try {
Thread.sleep(1000);
executor.execute(()-> System.out.println("async create value...."));
} catch (InterruptedException e) {
}
System.out.println("end query ...");
return CompletableFuture.completedFuture(UUID.randomUUID().toString());
}
}
03.驱除策略
a.基于容量
// 基于缓存容量大小，缓存中个数进行驱逐
Cache<String, String> cache =
Caffeine.newBuilder()
.maximumSize(100)
.recordStats()
.build();
// 基于缓存的权重进行驱逐
AsyncCache<String,String> asyncCache = Caffeine.newBuilder()
.maximumWeight(10)
.buildAsync();
b.基于时间
// 基于固定时间
Cache<Object, Object> cache =
Caffeine.newBuilder()
//距离上次访问后一分钟删除
.expireAfterAccess(Duration.ofMinutes(1))
.recordStats()
.build();
Cache<Object, Object> cache =
Caffeine.newBuilder()
// 距离上次写入一分钟后删除
.expireAfterWrite(Duration.ofMinutes(1))
.recordStats()
.build();
// 基于不同的过期驱逐策略
Cache<String, String> expire =
Caffeine.newBuilder()
.expireAfter(new Expiry<String, String>() {
@Override
public long expireAfterCreate(@NonNull String key, @NonNull String value, long currentTime) {
return LocalDateTime.now().plusMinutes(5).getSecond();
}
@Override
public long expireAfterUpdate(@NonNull String key, @NonNull String value, long currentTime, @NonNegative long currentDuration) {
return currentDuration;
}
@Override
public long expireAfterRead(@NonNull String key, @NonNull String value, long currentTime, @NonNegative long currentDuration) {
return currentDuration;
}
})
.recordStats()
.build();
-----------------------------------------------------------------------------------------------------
Caffeine提供了三种方法进行基于时间的驱逐:
expireAfterAccess(long, TimeUnit): 一个值在最近一次访问后，一段时间没访问时被淘汰
expireAfterWrite(long, TimeUnit): 一个值在初次创建或最近一次更新后，一段时间后被淘汰
expireAfter(Expiry): 一个值将会在指定的时间后被认定为过期项
c.基于引用
// 当key和缓存元素都不再存在其他强引用的时候驱逐
LoadingCache<Object, Object> weak = Caffeine.newBuilder()
.weakKeys()
.weakValues()
.build(k ->createExpensiveValue());
// 当进行GC的时候进行驱逐
LoadingCache<Object, Object> soft = Caffeine.newBuilder()
.softValues()
.build(k ->createExpensiveValue());
-----------------------------------------------------------------------------------------------------
weakKeys：使用弱引用存储key时，当没有其他的强引用时，则会被垃圾回收器回收
weakValues：使用弱引用存储value时，当没有其他的强引用时，则会被垃圾回收器回收
softValues：使用软引用存储key时，当没有其他的强引用时，内存不足时会被回收
d.手动移除
Cache<Object, Object> cache =
Caffeine.newBuilder()
.expireAfterWrite(Duration.ofMinutes(1))
.recordStats()
.build();
// 单个删除
cache.invalidate("a");
// 批量删除
Set<String> keys = new HashSet<>();
keys.add("a");
keys.add("b");
cache.invalidateAll(keys);
// 失效所有key
cache.invalidateAll();
-----------------------------------------------------------------------------------------------------
任何时候都可以手动删除，不用等到驱逐策略生效
e.移除监听器
Cache<Object, Object> cache =
Caffeine.newBuilder()
.expireAfterWrite(Duration.ofMinutes(1))
.recordStats()
.evictionListener(new RemovalListener<Object, Object>() {
@Override
public void onRemoval(@Nullable Object key, @Nullable Object value, @NonNull RemovalCause cause) {
System.out.println("element evict cause" + cause.name());
}
})
.removalListener(new RemovalListener<Object, Object>() {
@Override
public void onRemoval(@Nullable Object key, @Nullable Object value, @NonNull RemovalCause cause) {
System.out.println("element removed cause" + cause.name());
}
}).build();
-----------------------------------------------------------------------------------------------------
你可以为你的缓存通过Caffeine.removalListener(RemovalListener)方法定义一个移除监听器在一个元素被移除的时候进行相应的操作。这些操作是使用 Executor异步执行的，其中默认的 Executor 实现是 ForkJoinPool.commonPool()并且可以通过覆盖Caffeine.executor(Executor)方法自定义线程池的实现。
注意：Caffeine.evictionListener(RemovalListener)。这个监听器将在 RemovalCause.wasEvicted()为 true 的时候被触发。
f.驱逐原因汇总
EXPLICIT：如果原因是这个，那么意味着数据被我们手动的remove掉
REPLACED：就是替换了，也就是put数据的时候旧的数据被覆盖导致的移
COLLECTED：这个有歧义点，其实就是收集，也就是垃圾回收导致的，一般是用弱引用或者软引用会导致这个情
EXPIRED：数据过期，无需解释的原因
SIZE：个数超过限制导致的移
04.缓存统计
a.缓存访问统计
CacheStats stats = cache.stats();
System.out.println("stats.hitCount():"+stats.hitCount());//命中次数
System.out.println("stats.hitRate():"+stats.hitRate());//缓存命中率
System.out.println("stats.missCount():"+stats.missCount());//未命中次数
System.out.println("stats.missRate():"+stats.missRate());//未命中率
System.out.println("stats.loadSuccessCount():"+stats.loadSuccessCount());//加载成功的次数
System.out.println("stats.loadFailureCount():"+stats.loadFailureCount());//加载失败的次数,返回null
System.out.println("stats.loadFailureRate():"+stats.loadFailureRate());//加载失败的百分比
System.out.println("stats.totalLoadTime():"+stats.totalLoadTime());//总加载时间,单位ns
System.out.println("stats.evictionCount():"+stats.evictionCount());//驱逐次数
System.out.println("stats.evictionWeight():"+stats.evictionWeight());//驱逐的weight值总和
System.out.println("stats.requestCount():"+stats.requestCount());//请求次数
System.out.println("stats.averageLoadPenalty():"+stats.averageLoadPenalty());//单次load平均耗时
-----------------------------------------------------------------------------------------------------
Caffeine通过使用Caffeine.recordStats()方法可以打开数据收集功能，可以帮助优化缓存使用
05.核心代码
a.自定义CacheManager多级缓存实现
public class RedisCaffeineCacheManager implements CacheManager{
private ConcurrentMap<String, Cache> cacheMap = new ConcurrentHashMap<>();
private CacheConfigProperties cacheConfigProperties;
private RedisTemplate<Object, Object> stringKeyRedisTemplate;
// 是否动态根据cacheName创建Cache的实现
private boolean dynamic;
// 当前节点存储的缓存对象集合名称
private Set<String> cacheNames;
// 当前节点id
private Object serverId;
/*
* @Description：获取指定名称的缓存对象
* 如果 cacheMap 中已存在，则直接返回；
* 如果 cacheMap 中不存在且不允许动态创建缓存，则返回 null；
* 如果不存在且允许动态创建缓存，则调用 createCache 方法创建缓存并放入 cacheMap 中
* @Param：[name]
* @Return org.springframework.cache.Cache
*/
@Override
public Cache getCache(String name) {
Cache cache = cacheMap.get(name);
if (cache != null) {
return cache;
}
if (!dynamic && !cacheNames.contains(name)) {
return null;
}
cache = createCache(name);
Cache oldCache = cacheMap.putIfAbsent(name, cache);
log.debug("create cache instance, the cache name is : {}", name);
return oldCache == null ? cache : oldCache;
}
}
b.多级缓存查询实现
/**
* @Author：CSNZ
* @Description：自定义的缓存实现类，结合了 Redis 和 Caffeine 两种缓存机制的优点
* @Version：1.0
**/
@Slf4j
@Getter
public class RedisCaffeineCache extends AbstractValueAdaptingCache implements Cache<Object, Object> {
private final String name; // 缓存名称，例如 externalApiData
private final Cache<Object, Object> caffeineCache;
private final RedisTemplate<Object, Object> stringKeyRedisTemplate;
private final String cachePrefix;
private final String getKeyPrefix;
private final Duration defaultExpiration;
private final Duration defaultNullValuesExpiration;
private final Map<String, Duration> expires;
private final String topic;
private final Object serverId;
private final Map<String, ReentrantLock> keyLockMap = new ConcurrentHashMap<>();
/**
* 检查L1或L2缓存中是否存在键，不存在则返回 null
* @param key
* @return
*/
@Override
protected Object lookup(Object key) {
// 根据前缀拼接 key
Object cacheKey = getKey(key);
// 从 L1 中查找此 key
Object value = getCaffeineValue(key);
if (Objects.nonNull(value)) {
log.debug("get cache from caffeine, the key is : {}", cacheKey);
return value;
}
// L1 中查无此key，改从 L2 中查找
value = getRedisValue(key);
if (value != null) {
log.debug("get cache from redis and put in caffeine, the key is : {}", cacheKey);
setCaffeineValue(key, value);
}
return value;
}
}
06.Caffeine本地缓存
a.定义
Caffeine是一个高性能的Java缓存库，提供了接近理论最优的缓存性能
它是Guava Cache的增强版，采用了更先进的缓存算法，适用于需要高效缓存管理的场景
b.原理
Caffeine使用了一种结合LRU（Least Recently Used）和LFU（Least Frequently Used）优点的算法：W-TinyLFU
这种算法在缓存命中率和性能上都有显著的提升
Caffeine通过在内存中存储数据来实现快速访问，并提供多种缓存策略以适应不同的使用场景
c.常用API
Caffeine.newBuilder()：创建缓存构建器
maximumSize(long size)：设置缓存的最大容量
expireAfterWrite(long duration, TimeUnit unit)：设置写入后多久过期
expireAfterAccess(long duration, TimeUnit unit)：设置访问后多久过期
refreshAfterWrite(long duration, TimeUnit unit)：设置写入后多久刷新
build()：构建缓存实例
d.使用步骤
1.创建缓存构建器：使用Caffeine.newBuilder()初始化缓存构建器
2.配置缓存策略：根据需求设置缓存的最大容量、过期策略等
3.构建缓存实例：调用build()方法创建缓存实例
4.使用缓存：通过get(), put(), invalidate()等方法操作缓存
e.每个场景对应的代码示例
e.基本缓存配置
Cache<String, Object> cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterWrite(10, TimeUnit.MINUTES)
.build();
b.带刷新功能的缓存
Cache<String, Object> cache = Caffeine.newBuilder()
.maximumSize(10_000)
.expireAfterWrite(10, TimeUnit.MINUTES)
.refreshAfterWrite(5, TimeUnit.MINUTES)
.build(new CacheLoader<String, Object>() {
@Override
public Object load(String key) throws Exception {
return fetchDataFromDatabase(key);
}
});
3. 使用缓存
// 存储数据
cache.put("key1", "value1");
// 获取数据
String value = cache.getIfPresent("key1");
// 删除数据
cache.invalidate("key1");
f.场景应用
1.短期数据缓存：适用于需要快速访问但不需要持久化的数据，如会话信息、临时计算结果
2.热点数据缓存：适用于访问频率高的数据，通过Caffeine的高效算法提高缓存命中率
3.数据预热：在应用启动时预加载常用数据，减少首次访问延迟
07.Caffeine作为SpringCache实现
a.定义
Caffeine可以作为Spring Cache的实现，为Spring应用提供高性能的本地缓存支持
通过Spring的缓存抽象，开发者可以轻松地将Caffeine集成到Spring应用中，实现缓存的自动管理
b.常用API
1.CaffeineCacheManager：Spring提供的Caffeine缓存管理器
2.@Cacheable：用于标记方法，表示其返回值应被缓存
3.@CachePut：用于标记方法，表示其返回值应更新缓存
4.@CacheEvict：用于标记方法，表示应从缓存中移除某个值
5.@EnableCaching：启用Spring的缓存支持
c.使用步骤
1.添加依赖：在项目的pom.xml中添加Caffeine和Spring Cache的依赖
2.配置缓存管理器：在Spring配置类中定义CaffeineCacheManager
3.启用缓存：使用@EnableCaching注解启用Spring的缓存支持
4.使用缓存注解：在需要缓存的方法上使用@Cacheable、@CachePut、@CacheEvict等注解
d.每个场景对应的代码示例
a.添加依赖
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-cache</artifactId>
</dependency>
b.配置缓存管理器
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cache.caffeine.CaffeineCacheManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableCaching
public class CacheConfig {
@Bean
public CaffeineCacheManager cacheManager() {
CaffeineCacheManager cacheManager = new CaffeineCacheManager();
cacheManager.setCaffeine(Caffeine.newBuilder()
.maximumSize(1000)
.expireAfterWrite(10, TimeUnit.MINUTES));
return cacheManager;
}
}
c.使用缓存注解
在需要缓存的方法上使用@Cacheable、@CachePut、@CacheEvict等注解：
-------------------------------------------------------------------------------------------------
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
@Service
public class UserService {
@Cacheable("users")
public User getUserById(String userId) {
// 模拟数据库查询
return new User(userId, "John Doe");
}
@CachePut(value = "users", key = "#user.id")
public User updateUser(User user) {
// 更新用户信息
return user;
}
@CacheEvict(value = "users", key = "#userId")
public void deleteUser(String userId) {
// 删除用户
}
}
e.场景应用
1.用户信息缓存：通过@Cacheable注解缓存用户信息，减少数据库查询次数
2.配置数据缓存：通过@CachePut注解更新缓存中的配置数据，确保数据一致性
3.临时数据缓存：通过@CacheEvict注解移除过期或无效的临时数据
08.Redis+Caffeine实现两级缓存
a.说明
一种是常规的方式
一种的基于注解的方式
b.流程
查询先走一级缓存
一级缓存不存在查询二级缓存，然后写入一级缓存
二级缓存不存在，查询MySQL然后写入二级缓存，再写入一级缓存
c.MySQL表
a.表结构
CREATE TABLE `t_estimated_arrival_date` (
`id` int(11) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键id',
`warehouse_id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '货仓id',
`warehouse` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '发货仓',
`city` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '签收城市',
`delivery_date` date NULL DEFAULT NULL COMMENT '发货时间',
`estimated_arrival_date` date NULL DEFAULT NULL COMMENT '预计到货日期',
PRIMARY KEY (`id`) USING BTREE,
UNIQUE INDEX `uk_warehouse_id_city_delivery_date`(`warehouse_id`, `city`, `delivery_date`) USING BTREE
) ENGINE = InnoDB COMMENT = '预计到货时间表(具体到day：T, T+1,近90天到货时间众数)' ROW_FORMAT = Dynamic;
b.数据插入
INSERT INTO `t_estimated_arrival_date` VALUES (9, '6', '湖熟正常仓', '兰州市', '2024-07-08', '2024-07-10');
INSERT INTO `t_estimated_arrival_date` VALUES (10, '6', '湖熟正常仓', '兰州市', '2024-07-09', '2024-07-11');
INSERT INTO `t_estimated_arrival_date` VALUES (11, '6', '湖熟正常仓', '兴安盟', '2024-07-08', '2024-07-11');
INSERT INTO `t_estimated_arrival_date` VALUES (12, '6', '湖熟正常仓', '兴安盟', '2024-07-09', '2024-07-12');
INSERT INTO `t_estimated_arrival_date` VALUES (13, '6', '湖熟正常仓', '其他', '2024-07-08', '2024-07-19');
INSERT INTO `t_estimated_arrival_date` VALUES (14, '6', '湖熟正常仓', '其他', '2024-07-09', '2024-07-20');
INSERT INTO `t_estimated_arrival_date` VALUES (15, '6', '湖熟正常仓', '内江市', '2024-07-08', '2024-07-10');
INSERT INTO `t_estimated_arrival_date` VALUES (16, '6', '湖熟正常仓', '内江市', '2024-07-09', '2024-07-11');
INSERT INTO `t_estimated_arrival_date` VALUES (17, '6', '湖熟正常仓', '凉山彝族自治州', '2024-07-08', '2024-07-11');
INSERT INTO `t_estimated_arrival_date` VALUES (18, '6', '湖熟正常仓', '凉山彝族自治州', '2024-07-09', '2024-07-12');
INSERT INTO `t_estimated_arrival_date` VALUES (19, '6', '湖熟正常仓', '包头市', '2024-07-08', '2024-07-11');
INSERT INTO `t_estimated_arrival_date` VALUES (20, '6', '湖熟正常仓', '包头市', '2024-07-09', '2024-07-12');
INSERT INTO `t_estimated_arrival_date` VALUES (21, '6', '湖熟正常仓', '北京城区', '2024-07-08', '2024-07-10');
INSERT INTO `t_estimated_arrival_date` VALUES (22, '6', '湖熟正常仓', '北京城区', '2024-07-09', '2024-07-11');
d.pom.xml
a.依赖配置
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!--redis连接池-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-cache</artifactId>
</dependency>
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.28</version>
</dependency>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
<version>3.3.1</version>
</dependency>
e.application.yml
server:
port: 9001
spring:
application:
name: springboot-redis
datasource:
name: demo
url: jdbc:mysql://localhost:3306/test?userUnicode=true&&characterEncoding=utf8&allowMultiQueries=true&useSSL=false
driver-class-name: com.mysql.cj.jdbc.Driver
username:
password:
# mybatis相关配置
mybatis-plus:
mapper-locations: classpath:mapper/*.xml
configuration:
cache-enabled: true
use-generated-keys: true
default-executor-type: REUSE
use-actual-param-name: true
redis:
host: 192.168.117.73
port: 6379
password: root
logging:
level:
com.itender.redis.mapper: debug
f.配置类
a.RedisConfig
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(connectionFactory);
Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper mapper = new ObjectMapper();
mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
mapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
serializer.setObjectMapper(mapper);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(serializer);
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(serializer);
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}
b.CaffeineConfig
@Configuration
public class CaffeineConfig {
@Bean
public Cache<String, Object> caffeineCache() {
return Caffeine.newBuilder()
.initialCapacity(128)
.maximumSize(1024)
.expireAfterWrite(60, TimeUnit.SECONDS)
.build();
}
@Bean
public CacheManager cacheManager(){
CaffeineCacheManager cacheManager=new CaffeineCacheManager();
cacheManager.setCaffeine(Caffeine.newBuilder()
.initialCapacity(128)
.maximumSize(1024)
.expireAfterWrite(60, TimeUnit.SECONDS));
return cacheManager;
}
}
g.Mapper
@Mapper
public interface EstimatedArrivalDateMapper extends BaseMapper<EstimatedArrivalDateEntity> {
}
h.Service
a.接口定义
public interface DoubleCacheService {
EstimatedArrivalDateEntity getEstimatedArrivalDateCommon(EstimatedArrivalDateEntity request);
EstimatedArrivalDateEntity getEstimatedArrivalDate(EstimatedArrivalDateEntity request);
}
b.实现类
@Slf4j
@Service
public class DoubleCacheServiceImpl implements DoubleCacheService {
@Resource
private Cache<String, Object> caffeineCache;
@Resource
private RedisTemplate<String, Object> redisTemplate;
@Resource
private EstimatedArrivalDateMapper estimatedArrivalDateMapper;
@Override
public EstimatedArrivalDateEntity getEstimatedArrivalDateCommon(EstimatedArrivalDateEntity request) {
String key = request.getDeliveryDate() + RedisConstants.COLON + request.getWarehouseId() + RedisConstants.COLON + request.getCity();
log.info("Cache key: {}", key);
Object value = caffeineCache.getIfPresent(key);
if (Objects.nonNull(value)) {
log.info("get from caffeine");
return EstimatedArrivalDateEntity.builder().estimatedArrivalDate(value.toString()).build();
}
value = redisTemplate.opsForValue().get(key);
if (Objects.nonNull(value)) {
log.info("get from redis");
caffeineCache.put(key, value);
return EstimatedArrivalDateEntity.builder().estimatedArrivalDate(value.toString()).build();
}
log.info("get from mysql");
DateTime deliveryDate = DateUtil.parse(request.getDeliveryDate(), "yyyy-MM-dd");
EstimatedArrivalDateEntity estimatedArrivalDateEntity = estimatedArrivalDateMapper.selectOne(new QueryWrapper<EstimatedArrivalDateEntity>()
.eq("delivery_date", deliveryDate)
.eq("warehouse_id", request.getWarehouseId())
.eq("city", request.getCity())
);
redisTemplate.opsForValue().set(key, estimatedArrivalDateEntity.getEstimatedArrivalDate(), 120, TimeUnit.SECONDS);
caffeineCache.put(key, estimatedArrivalDateEntity.getEstimatedArrivalDate());
return EstimatedArrivalDateEntity.builder().estimatedArrivalDate(estimatedArrivalDateEntity.getEstimatedArrivalDate()).build();
}
@DoubleCache(cacheName = "estimatedArrivalDate", key = {"#request.deliveryDate", "#request.warehouseId", "#request.city"},
type = DoubleCache.CacheType.FULL)
@Override
public EstimatedArrivalDateEntity getEstimatedArrivalDate(EstimatedArrivalDateEntity request) {
DateTime deliveryDate = DateUtil.parse(request.getDeliveryDate(), "yyyy-MM-dd");
EstimatedArrivalDateEntity estimatedArrivalDateEntity = estimatedArrivalDateMapper.selectOne(new QueryWrapper<EstimatedArrivalDateEntity>()
.eq("delivery_date", deliveryDate)
.eq("warehouse_id", request.getWarehouseId())
.eq("city", request.getCity())
);
return EstimatedArrivalDateEntity.builder().estimatedArrivalDate(estimatedArrivalDateEntity.getEstimatedArrivalDate()).build();
}
}
i.Annotitions
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DoubleCache {
String cacheName();
String[] key();
long expireTime() default 120;
CacheType type() default CacheType.FULL;
enum CacheType {
FULL,
PUT,
DELETE
}
}
j.Aspect
@Slf4j
@Component
@Aspect
public class DoubleCacheAspect {
@Resource
private Cache<String, Object> caffeineCache;
@Resource
private RedisTemplate<String, Object> redisTemplate;
@Pointcut("@annotation(com.itender.redis.annotation.DoubleCache)")
public void doubleCachePointcut() {
}
@Around("doubleCachePointcut()")
public Object doAround(ProceedingJoinPoint point) throws Throwable {
MethodSignature signature = (MethodSignature) point.getSignature();
Method method = signature.getMethod();
String[] paramNames = signature.getParameterNames();
Object[] args = point.getArgs();
TreeMap<String, Object> treeMap = new TreeMap<>();
for (int i = 0; i < paramNames.length; i++) {
treeMap.put(paramNames[i], args[i]);
}
DoubleCache annotation = method.getAnnotation(DoubleCache.class);
String elResult = DoubleCacheUtil.arrayParse(Lists.newArrayList(annotation.key()), treeMap);
String realKey = annotation.cacheName() + RedisConstants.COLON + elResult;
if (annotation.type() == DoubleCache.CacheType.PUT) {
Object object = point.proceed();
redisTemplate.opsForValue().set(realKey, object, annotation.expireTime(), TimeUnit.SECONDS);
caffeineCache.put(realKey, object);
return object;
} else if (annotation.type() == DoubleCache.CacheType.DELETE) {
redisTemplate.delete(realKey);
caffeineCache.invalidate(realKey);
return point.proceed();
}
Object caffeineCacheObj = caffeineCache.getIfPresent(realKey);
if (Objects.nonNull(caffeineCacheObj)) {
log.info("get data from caffeine");
return caffeineCacheObj;
}
Object redisCache = redisTemplate.opsForValue().get(realKey);
if (Objects.nonNull(redisCache)) {
log.info("get data from redis");
caffeineCache.put(realKey, redisCache);
return redisCache;
}
log.info("get data from database");
Object object = point.proceed();
if (Objects.nonNull(object)) {
log.info("get data from database write to cache: {}", object);
redisTemplate.opsForValue().set(realKey, object, annotation.expireTime(), TimeUnit.SECONDS);
caffeineCache.put(realKey, object);
}
return object;
}
}
k.Controller
@RestController
@RequestMapping("/doubleCache")
public class DoubleCacheController {
@Resource
private DoubleCacheService doubleCacheService;
@PostMapping("/common")
public EstimatedArrivalDateEntity getEstimatedArrivalDateCommon(@RequestBody EstimatedArrivalDateEntity estimatedArrivalDate) {
return doubleCacheService.getEstimatedArrivalDateCommon(estimatedArrivalDate);
}
@PostMapping("/annotation")
public EstimatedArrivalDateEntity getEstimatedArrivalDate(@RequestBody EstimatedArrivalDateEntity estimatedArrivalDate) {
return doubleCacheService.getEstimatedArrivalDate(estimatedArrivalDate);
}
}
4.4 [1]MQ广播模式
01.MQ广播模式
a.定义
消息队列是一种异步通信机制，用于在分布式系统中传递消息。通过广播模式，可以在数据修改时通知多个节点更新缓存
b.原理
当数据发生变化时，向消息队列发送消息。各节点监听并消费消息，删除或更新本地缓存，以达到最终一致性
c.常用API
Producer：用于发送消息
Consumer：用于接收和处理消息
Topic：消息的主题，用于分类和过滤
d.使用步骤
1.配置消息队列：设置消息队列的主题和消费者
2.发送消息：在数据修改时，向消息队列发送更新通知
3.消费消息：各节点监听消息队列，接收到消息后更新缓存
e.代码示例
// 发送消息
Producer producer = new Producer();
producer.send(new Message("cache-update-topic", "key1"));
// 消费消息
Consumer consumer = new Consumer("cache-update-topic");
consumer.setMessageListener(message -> {
// 删除本地缓存
localCache.invalidate(message.getBody());
});
02.代码示例
a.思路
当店铺数据修改时，向消息队列（MQ）发送更新通知。各节点监听MQ，接收到消息后，删除本地缓存，确保缓存一致性
b.实现步骤
1.发送消息：在更新店铺信息后，向MQ发送消息
2.监听消息：各节点订阅MQ，接收到消息后删除本地缓存
c.代码示例
// 更新店铺信息并发送MQ消息
@Override
@Transactional
public Result update(Shop shop) {
Long id = shop.getId();
if (id == null) {
return Result.fail("店铺id不能为空");
}
// 1. 更新数据库
updateById(shop);
// 2. 删除Redis缓存
stringRedisTemplate.delete("cache:shop:" + id);
// 3. 发送MQ消息
producer.send(new Message("cache-update-topic", "shop:" + id));
return Result.ok();
}
// 消费MQ消息并删除本地缓存
public void onMessage(Message message) {
String key = message.getBody();
localCache.invalidate(key);
}
4.5 [2]Redis发布/订阅
01.Redis发布/订阅
a.定义
Redis的发布/订阅机制允许消息在不同实例之间传递，用于通知缓存更新
b.原理
通过Redis的发布/订阅功能，当缓存数据发生变化时，发布消息通知其他实例更新缓存
c.常用API
publish(channel, message)：发布消息到指定频道
subscribe(channel, listener)：订阅指定频道，监听消息
d.使用步骤
1.设置频道：定义用于发布和订阅的频道
2.发布消息：在数据修改时，发布更新通知到频道
3.订阅消息：各实例订阅频道，接收到消息后更新缓存
e.代码示例
// 发布消息
redisTemplate.convertAndSend("cache-update-channel", "key1");
// 订阅消息
redisTemplate.subscribe((message, pattern) -> {
// 删除本地缓存
localCache.invalidate(message.toString());
}, "cache-update-channel");
02.代码示例
a.思路
使用Redis的发布/订阅机制，当店铺数据修改时，发布更新通知。各实例订阅频道，接收到消息后，删除本地缓存
b.实现步骤
1.发布消息：在更新店铺信息后，发布消息到Redis频道
2.订阅消息：各实例订阅频道，接收到消息后删除本地缓存
c.代码示例
// 更新店铺信息并发布Redis消息
@Override
@Transactional
public Result update(Shop shop) {
Long id = shop.getId();
if (id == null) {
return Result.fail("店铺id不能为空");
}
// 1. 更新数据库
updateById(shop);
// 2. 删除Redis缓存
stringRedisTemplate.delete("cache:shop:" + id);
// 3. 发布Redis消息
redisTemplate.convertAndSend("cache-update-channel", "shop:" + id);
return Result.ok();
}
// 订阅Redis消息并删除本地缓存
redisTemplate.subscribe((message, pattern) -> {
String key = message.toString();
localCache.invalidate(key);
}, "cache-update-channel");
4.6 [3]Canal + MQ
01.Canal + MQ
a.定义
Canal是一个用于解析MySQL Binlog的工具，结合消息队列，可以实现数据库变更的实时通知
b.原理
通过订阅MySQL的Binlog日志，Canal解析数据变更并发送消息到消息队列，通知各节点更新缓存
c.常用API
CanalConnector：用于连接和订阅Binlog
Message：表示从Binlog解析的消息
MQ Producer/Consumer：用于发送和接收消息
d.使用步骤
1.配置Canal：连接MySQL并订阅Binlog
2.解析Binlog：Canal解析数据变更并生成消息
3.发送消息：将解析的消息发送到消息队列
4.消费消息：各节点监听消息队列，接收到消息后更新缓存
e.代码示例
// Canal订阅Binlog
CanalConnector connector = CanalConnectors.newSingleConnector(...);
connector.connect();
connector.subscribe("database.table");
// 解析并发送消息
Message message = connector.getWithoutAck(...);
producer.send(new Message("cache-update-topic", message.getData()));
// 消费消息
Consumer consumer = new Consumer("cache-update-topic");
consumer.setMessageListener(message -> {
// 删除本地缓存
localCache.invalidate(message.getBody());
});
02.代码示例
a.思路
使用Canal订阅MySQL的Binlog日志，解析数据变更。将变更信息发送到MQ，各节点接收到消息后，删除本地缓存
b.实现步骤
1.订阅Binlog：使用Canal订阅MySQL的Binlog日志
2.发送消息：解析变更后，将消息发送到MQ
3.消费消息：各节点监听MQ，接收到消息后删除本地缓存
c.代码示例
// Canal订阅Binlog并发送MQ消息
CanalConnector connector = CanalConnectors.newSingleConnector(...);
connector.connect();
connector.subscribe("database.table");
Message message = connector.getWithoutAck(...);
producer.send(new Message("cache-update-topic", message.getData()));
// 消费MQ消息并删除本地缓存
public void onMessage(Message message) {
String key = message.getBody();
localCache.invalidate(key);
}
4.7 [4]J2Cache
01.为什么选择J2Cache
a.背景
J2Cache 是一个高性能双层缓存框架，旨在解决常见的缓存问题
它结合了内存缓存和集中式缓存的优点，提供了一种高效的缓存解决方案
b.关键问题
使用内存缓存时，应用重启后缓存数据丢失，可能导致缓存雪崩
在内存缓存模式下，多个应用节点无法共享缓存数据
集中式缓存（如Redis）在大量数据读取时可能因网络带宽限制而导致数据读取速度下降
c.解决方案
J2Cache 采用双层缓存模式：利用内存缓存作为一级缓存，Redis 作为二级缓存。通过这种结构，减少对二级缓存的访问次数，降低Redis服务器的数据流量压力
J2Cache 通过 Redis Pub/Sub 和 JGroups 提供了节点间数据同步解决方案，确保缓存数据的一致性
02.J2Cache介绍
a.概述
J2Cache 是由开源社区 OSChina 开发的一个高性能双层缓存框架，专为解决企业级Java应用中的缓存问题而设计
b.两级缓存架构
第一级（L1）缓存：位于应用进程内，使用内存缓存技术（支持Ehcache 2.x、Ehcache 3.x和Caffeine等）
第二级（L2）缓存：采用外部缓存存储，如Redis或Memcached，适用于分布式环境中的数据共享和持久化
c.缓存协作机制
在J2Cache的架构中，两级缓存协同工作，以实现数据的高效访问和同步
当应用请求数据时，先在L1缓存中查找；如果L1缓存未命中，则查询L2缓存
在L2缓存找到数据后，该数据将被回填到L1缓存，以便未来的访问可以直接从L1缓存中获得
03.J2Cache 组播
a.缓存同步策略
Redis Pub/Sub机制（不推荐）：基于Redis的发布/订阅功能实现缓存之间的数据同步
JGroups广播机制（推荐）：使用JGroups的广播机制进行缓存同步，特别是在配置为TCP模式时
b.原理介绍
J2Cache的目标是确保所有集群节点中的缓存数据保持同步
当缓存数据在一个节点上更新时，该变更会通过选定的同步机制广播到所有其他节点
04.使用方法及实际示例
a.引用 Maven
<properties>
<j2cache.version>2.8.0-release</j2cache.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>net.oschina.j2cache</groupId>
<artifactId>j2cache-spring-boot2-starter</artifactId>
<version>${j2cache.version}</version>
</dependency>
<dependency>
<groupId>net.oschina.j2cache</groupId>
<artifactId>j2cache-core</artifactId>
<version>${j2cache.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</dependencyManagement>
b.准备配置
redis:
ip: 10.41.170.160
port: 6380
database: 1
password: test@6380
j2cache:
open-spring-cache: true
cache-clean-mode: passive
allow-null-values: true
redis-client: lettuce
l2-cache-open: true
broadcast: net.oschina.j2cache.cache.support.redis.SpringRedisPubSubPolicy
L1:
provider_class: caffeine
L2:
provider_class: net.oschina.j2cache.cache.support.redis.SpringRedisProvider
config_section: lettuce
sync_ttl_to_redis: true
default_cache_null_object: false
serialization: com.iflytek.sd.common.config.J2cacheJacksonSerializer
caffeine:
properties: /caffeine.properties
lettuce:
mode: single
namespace:
storage: generic
channel: j2cache
scheme: redis
hosts: ${redis.ip}:${redis.port}
password: ${redis.password}
database: ${redis.database}
maxTotal: 2000
maxIdle: 1600
minIdle: 1000
timeout: 10000
c.添加配置，caffeine.properties文件
#########################################
# Caffeine configuration
# [name] = size, xxxx[s|m|h|d]
#########################################
default=2000, 2h
hn=50, 2h
hb=100, 30s
d.使用实例
/**
* 清理指定缓存
*/
@Override
public String deleTest(Long id) {
if (exists(id)) {
cacheChannel.evict("test", getCacheKey(id.toString()));
}
return "evict success:" + id;
}
/**
* 新增test
*/
@Override
@Async
public void insertTest(Test test) {
int row = testMapper.insertTest(test);
if (row > 0) {
cacheChannel.set("test", getCacheKey(String.valueOf(test.getId())), test.getText());
}
}
/**
* 检测存在哪级缓存
*/
@Override
public int check(Long id) {
return cacheChannel.check("test", getCacheKey(id.toString()));
}
/**
* 检测缓存数据是否存在
*/
@Override
public boolean exists(Long id) {
return cacheChannel.exists("test", getCacheKey(id.toString()));
}
/**
* 清理指定区域的缓存
*/
@Override
public String deleTest(String region) {
cacheChannel.clear(region);
return "clear cache success";
}
4.8 [5]xxl-cache
01.XXL-CACHE 概述
a.定义
XXL-CACHE 是由许雪里团队开发的一款轻量级分布式多级缓存框架
旨在提供高效的缓存解决方案，支持多级缓存策略，以提高系统性能和数据访问速度
b.原理
XXL-CACHE 通过结合本地缓存和分布式缓存，提供多级缓存机制，它通常包括以下几个层次：
一级缓存：通常是本地内存缓存，用于快速访问最近使用的数据
二级缓存：通常是分布式缓存，如 Redis，用于共享数据和持久化存储
缓存更新策略：通过配置缓存的过期时间和更新策略，确保数据的一致性和有效性
02.常用 API
a.描述
XXL-CACHE 提供了一些简单易用的 API，用于缓存的设置、获取和删除操作
b.示例
set(String key, Object value, long expireTime)：设置缓存数据及其过期时间
get(String key)：获取缓存数据
remove(String key)：删除缓存数据
03.使用步骤
a.引入依赖
在项目中添加 XXL-CACHE 的依赖
b.配置缓存
根据需求配置本地缓存和分布式缓存的策略
c.使用缓存 API
通过 XXL-CACHE 提供的 API 进行缓存操作
04.每个场景对应的代码示例
a.引入依赖
<dependency>
<groupId>com.xuxueli</groupId>
<artifactId>xxl-cache</artifactId>
<version>1.0.0</version>
</dependency>
b.配置缓存
# xxl-cache.properties
xxl.cache.level1.provider_class = caffeine
xxl.cache.level2.provider_class = redis
# Redis 配置
redis.hosts = 127.0.0.1:6379
redis.password =
redis.database = 0
c.使用缓存 API
import com.xuxueli.cache.XxlCache;
public class XxlCacheExample {
public static void main(String[] args) {
XxlCache cache = new XxlCache();
// 设置缓存
cache.set("key1", "value1", 3600); // 缓存1小时
// 获取缓存
String value = (String) cache.get("key1");
System.out.println("Cached Value: " + value);
// 删除缓存
cache.remove("key1");
}
}
4.9 [6]JetCache
01.JetCache简介
a.说明
JetCache 是一个 Java 缓存抽象框架，为不同缓存解决方案提供统一使用方式，
其注解功能比 Spring Cache 更强大。它支持原生 TTL、两级缓存、分布式环境下自动刷新，还能通过代码操作缓存实例。
-----------------------------------------------------------------------------------------------------
当前有 RedisCache、TairCache（未开源）、CaffeineCache 和 LinkedHashMapCache 四种实现。
其具备多种特性，如通过统一缓存 API 操作缓存、支持带 TTL 和两级缓存的注解式方法缓存、
可创建及配置缓存实例、自动收集缓存访问统计信息、自定义键生成和值序列化策略、支持多种缓存键和值转换器、
分布式缓存自动刷新和分布式锁、异步访问、更新后使本地缓存失效以及 Spring Boot支持等。
b.环境要求
JDK1.8
Spring Framework4.0.8+ (optional, with annotation support)，jetcache 2.7 need 5.2.4+
Spring Boot 1.1.9+ (optional), jetcache 2.7 need 2.2.5+
02.使用案例
a.依赖
<dependency>
<groupId>com.alicp.jetcache</groupId>
<artifactId>jetcache-starter-redis</artifactId>
<version>2.7.8</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>4.4.3</version>
</dependency>
-----------------------------------------------------------------------------------------------------
jetcache-starter-redis默认使用的是Jedis客户端。
因为jetcache-starter-redis 2.7.8和Spring Boot 2.6.13有冲突，需要引入Jedis 4.x版本，否则就是出现部分类找不到的现象
b.配置文件
jetcache.statIntervalMinutes=15
jetcache.areaInCacheName=false
jetcache.local.default.type=linkedhashmap
jetcache.local.default.keyConvertor=jackson
jetcache.local.default.limit=100
jetcache.remote.default.type=redis
jetcache.remote.default.keyConvertor=jackson
jetcache.remote.default.valueEncoder=java
jetcache.remote.default.valueDecoder=java
jetcache.remote.default.poolConfig.minIdle=5
jetcache.remote.default.poolConfig.maxIdle=20
jetcache.remote.default.poolConfig.maxTotal=50
jetcache.remote.default.cluster[0]=127.0.0.1:7000
jetcache.remote.default.cluster[1]=127.0.0.1:7001
jetcache.remote.default.cluster[2]=127.0.0.1:7002
c.注解使用
a.注解可以加在接口上，也可以加在类上。案例以在类上使用为例
# 启动类的注解
@EnableMethodCache(basePackages = "xx.xx.xx")
b.常用注解
@Cached
@CacheUpdate
@CacheInvalidate
@CacheRefresh
c.@Cached 和 @CacheRefresh
@Cached(expire = 3600, cacheType = CacheType.BOTH, name = "cache_key_", key = "#userId")
@CacheRefresh(refresh = 10)
@RequestMapping("test01")
public String test01(String userId) {
System.out.println(LocalDateTime.now().format(FORMAT) + " userId=" + userId);
return "test01" + Instant.now().toEpochMilli();
}
-------------------------------------------------------------------------------------------------
代码注意事项：
CacheType.BOTH表示同时使用本地缓存和远程缓存，这里可以指定任意一个缓存使用。
属性中name和key的拼接组成Redis中的key
key使用的是EL表达式，通过#取属性的值
@CacheRefresh的refresh指每隔多长时间刷新一次缓存
-------------------------------------------------------------------------------------------------
缓存更新效果：再次请求缓存已经更新
d.@CacheUpdate
@CacheUpdate(name = "cache_key_", key = "#userId", value = "#val")
@RequestMapping("test03")
public String test03(String userId, String val) {
return "缓存更新成功！";
}
-------------------------------------------------------------------------------------------------
这里要说明的更新的value的值也是通过EL表达式从参数中获取的，无法指定方法中的数据作为缓存的结果。
e.@CacheInvalidate
@CacheInvalidate(name = "cache_key_", key = "#userId")
@RequestMapping("test04")
public String test04(String userId) {
return "缓存失效！";
}
-------------------------------------------------------------------------------------------------
这里值得注意的是，这里的缓存失效，会将远程的缓存失效以及本机的缓存失效，其他节点的本地缓存并不能同步失效。
本地缓存可以同步失效，需要配置广播通道broadcastChannel以及@Cached开启syncLocal = true属性。
d.手动API接管
@Autowired
CacheManager cacheManager;
@RequestMapping("test02")
public String test02(String userId) {
Cache<String, String> cacheKey = cacheManager.getCache("cache_key_");
if (cacheKey == null) {
return "cache_key_ 不存在";
}
if (cacheKey instanceof MultiLevelCache cache) {
Cache[] caches = cache.caches();
Arrays.stream(caches).forEach(item -> {
if (item instanceof LinkedHashMapCache) {
System.out.println("LinkedHashMapCache:" + item.get(userId));
}else if (item instanceof RedisCache) {
System.out.println("RedisCache:" + item.get(userId));
}
});
}
return cacheKey.get(userId);
}
e.命中率统计
JetCache自带了统计功能，每隔一段时间就会输出统计数据。通过jetcache.statIntervalMinutes配置控制。
03.注意事项
a.汇总1
因为结合本地缓存和远程缓存，那么框架就存在这两者的优势和劣势
b.汇总2
使用的时候也需要注意：
缓存防止大Key的出现
本地缓存需要增加容量限制
结合TTL和主动失效
多节点本地缓存不同步的问题
使用时Spring Boot、Jedis、JetCache版本兼容问题
4.10 缓存数据自动刷新
01.概述
该缓存工具类、主要解决了缓存数据自动刷新的问题
在很多场景下、我们需要将一些代价昂贵的操作（例如数据库查询、远程接口调用或者复杂计算）的结果缓存起来
避免频繁重复执行、但同时又需要保证缓存数据不会长时间过时
02.CacheUtils提供了两种构建LoadingCache的方法
a.异步刷新缓存
通过asyncReloading、实现了当缓存数据达到设定的过期时间后、自动在后台线程池中刷新数据
这样、如果一个线程请求数据时、发现缓存即将生效或者正在刷新
它可以先返回旧数据、而不必等待新数据加载完毕、从而提升系统响应速度
b.同步刷新缓存
当缓存过期时，新请求会阻塞等待最新数据加载完成、然后返回最新数据
这种方式适用于缓存数据与当前线程上下文（比如ThreadLocal）的关联性较强的场景
03.代码示例
a.工具类
public class CacheUtils {
/**
* 构建异步刷新的 LoadingCache 对象
*
* 注意：如果你的缓存和 ThreadLocal 有关系，要么自己处理 ThreadLocal 的传递，要么使用 {@link #buildCache(Duration, CacheLoader)} 方法
*
* 或者简单理解：
* 1、和“人”相关的，使用 {@link #buildCache(Duration, CacheLoader)} 方法
* 2、和“全局”、“系统”相关的，使用当前缓存方法
*
* @param duration 过期时间
* @param loader CacheLoader 对象
* @return LoadingCache 对象
*/
public static <K, V> LoadingCache<K, V> buildAsyncReloadingCache(Duration duration, CacheLoader<K, V> loader) {
return CacheBuilder.newBuilder()
// 只阻塞当前数据加载线程，其他线程返回旧值
.refreshAfterWrite(duration)
// 通过 asyncReloading 实现全异步加载，包括 refreshAfterWrite 被阻塞的加载线程
.build(CacheLoader.asyncReloading(loader, Executors.newCachedThreadPool())); // TODO 芋艿：可能要思考下，未来要不要做成可配置
}
/**
* 构建同步刷新的 LoadingCache 对象
*
* @param duration 过期时间
* @param loader CacheLoader 对象
* @return LoadingCache 对象
*/
public static <K, V> LoadingCache<K, V> buildCache(Duration duration, CacheLoader<K, V> loader) {
return CacheBuilder.newBuilder().refreshAfterWrite(duration).build(loader);
}
}
b.测试类
假设我们有一个场景：从数据库中查询用户信息，但查询操作较慢，我们希望将用户数据缓存起来，并且定时刷新数据
-----------------------------------------------------------------------------------------------------
package util.cache;
import com.byd.common.util.cache.CacheUtils;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import lombok.SneakyThrows;
import org.junit.jupiter.api.Test;
import java.time.Duration;
public class CacheTest {
/**
* 1.模拟一个耗时的数据库查询方法
*/
@SneakyThrows
public static String queryFromDB(String key) {
Thread.sleep(2000);
return "Key:"+key;
}
@Test
@SneakyThrows
public void test() {
// 使用异步刷新缓存、设置缓存过期时间为10秒
LoadingCache<String, String> userCache = CacheUtils.buildAsyncReloadingCache(Duration.ofSeconds(10), new CacheLoader<>() {
@Override
public String load(String key) throws Exception {
System.out.println("load key:" + key);
return queryFromDB(key);
}
});
// 第一次获取数据、会触发 load 方法、延迟2秒
System.out.println("User 1:"+userCache.get("user1"));
// 后续的调用、如果在10秒内，不会再次加载
System.out.println("User 1:"+userCache.get("user1"));
// 等待 12 秒、让缓存过期并触发异步刷新
Thread.sleep(12000);
// 再次获取数据、仍然返回旧数据，同时在后台刷新
System.out.println("User 1 (after 12s):"+userCache.get("user1"));
// 等待 3 秒、让后台刷新完成
Thread.sleep(3000);
// 再次获取数据、返回新数据
System.out.println("User 1 (refreshed):"+userCache.get("user1"));
}
}
-----------------------------------------------------------------------------------------------------
初次加载：调用 userCache.get(1) 时，由于缓存中没有对应数据，会调用 load 方法，从数据库中查询用户信息，此时会有大约 2 秒的延迟
缓存命中：在缓存有效期内（10 秒），多次调用 get 方法，直接返回缓存中的数据
异步刷新：当缓存超过 10 秒后，下一次请求会先返回旧数据，并在后台异步刷新最新数据。这样不会阻塞请求线程
数据更新：等待后台刷新完成后，后续的请求将返回最新的用户信息
5 限流：单位时间的请求数量
5.1 汇总：4个
01.自定义注解和AOP
自定义注解结合AOP（面向切面编程）是一种优雅的方式来防止重复提交
通过在方法上添加自定义注解，并在AOP切面中拦截请求，可以有效地防止重复提交
02.令牌桶算法：string结构
使用 Redis 的 SET 和 GET 命令维护令牌桶，控制令牌的生成和消耗
SET bucket_key tokens_count
GET bucket_key
03.滑动窗口限流：zset结构
使用 Redis 的 ZADD 和 ZRANGEBYSCORE 命令记录和检查请求时间戳，以计算在滑动时间窗口内的请求数
ZADD rate_limiter key timestamp
ZRANGEBYSCORE rate_limiter key (current_time - window_size, current_time)
04.漏斗算法：string结构
使用 Redis 的 SET 和 GET 命令来控制请求的流入速率
SET funnel_key timestamp
GET funnel_key
5.2 防抖节流：2个
00.汇总
a.防抖
控制事件的【触发次数】
只关心最终状态：当事件频繁触发时，只有最后一次触发后 delay 毫秒才执行，适用于需要最终结果的场景（如搜索框输入、表单验证）
b.节流
限制事件的【执行次数】
关心中间状态：以固定速率执行事件处理，适用于需要频繁处理但限制执行频率的场景（如滚动监听、窗口调整）
c.避免接口重复请求
防抖
节流
请求锁定（加laoding状态）
axios.CancelToken取消重复请求
01.防抖、节流
a.防抖
a.定义
确保某个函数在一段时间内之执行一次，但执行函数之后等待规定的时间内没有触发才会执行当前函数
b.手搓一个防抖函数
a.代码
function debounce(fn,delay){
let timer;
return function(){
const _this = this;
const args = arguments;
if(timer){
clearTimeout(timer);
}
timer = setTimeout(()=>{
fn.apply(context, args);
},delay)
}
}
b.说明
先创建一个定时器变量timer
每次时间触发先判断是否存在定时器，若有则清楚之前的定时器
若没有定时器，则新建定时器并存入到定时器变量中
定时器时间到，执行回到运行fn函数
c.手搓一个防抖函数，可通过immediate判断是否立即执行
a.代码
// immediate为true时为立即执行,false为非立即执行
function debounce(func, wait, immediate) {
let timer;
return function () {
const _this = this;
const args = arguments;
if (timer) clearTimeout(timer);
if (immediate) {
const callNow = !timer;
timer = setTimeout(() => {
timer = null;
}, wait)
if (callNow) func.apply(_this, args)
}
else {
timer = setTimeout(() => {
func.apply(context, args)
}, wait);
}
}
}
b.非立即执行
定义：首次触发不立即执行，等待指定的延迟时间（delay）后执行
适用场景：用户输入完成后执行，例如搜索框、表单验证等场景
资源消耗：减少高频操作的资源消耗，适合需要降低频率的操作
用户体验：更适合不需要立刻反应的场景，用户输入完成后再进行处理
c.立即执行
定义：首次触发立即执行，后续触发在指定的延迟时间内不再执行
适用场景：用户点击、首次操作时需要立即响应，例如按钮点击等场景
资源消耗：适合需要立即响应的操作，如按钮防抖、API请求
用户体验：用户点击时，立即给予反馈，更直观的响应
d.总结
非立即执行：适用于需要在用户操作完成后进行处理的场景，能够有效减少资源消耗，提升性能
立即执行：适用于需要在用户操作时立即给予反馈的场景，能够提升用户体验，提供及时响应
d.使用场景
a.电梯开门
电梯门在有人进出时会不断触发关闭倒计时
但只有在最后一次有人进入后的一段时间内无人再进出，电梯门才会真正关闭
b.输入搜索
在搜索框输入内容时，用户每次输入都会触发搜索请求
但搜索请求应该在用户停止输入一段时间后再执行，避免频繁请求服务器
b.节流
a.定义
某个函数在短时间内多次触发，会按照固定的时间间隔执行，而不会让他高频触发
b.手搓一个节流函数
a.代码
// 定时器版本
function throttle(func,wait){
let timer;
return function (){
const _this = this;
const args = arguments;
if(!timer){
timer = setTimeopunt(()=>{
fn.apply(_this, args);
timer = null;
},wait)
}
}
}
// 时间戳版本
function throttleWithTimestamp(fn, delay) {
let lastTime = 0;
return function () {
const _this = this;
const args = arguments;
const now = Date.now();
if (now - lastTime >= delay) {
fn.apply(_this, args); // 立即执行
lastTime = now;
}
};
}
b.说明
先创建一个定时器变量timer
事件触发都要判断是否存在定时器，若存在则不开启新的定时器
若没有新的定时器，则开始定时器并存至timer变量中
时间到，执行定时器回调，即执行func函数并清空timer
c.时间戳版
触发时机：立即执行，之后 delay 时间内不执行
适用场景：鼠标移动、滚动事件（想要第一下生效）
d.定时器版
触发时机：不立即执行，delay 后执行
适用场景：防止高频触发但不丢失最后一次执行
c.使用场景
a.公交车发车
即使乘客不断到达，公交车也会按照固定的时间间隔发车，而不会每到一个人就立刻发车
b.MOBA游戏技能
技能冷却之时不能进行技能的施放，当技能冷却转好之后才可以进行技能的释放
02.避免接口重复请求：防抖、节流、请求锁定、axios.CancelToken取消重复请求
a.防抖
a.代码
<template>
<div>
<button @click="debouncedFetchData">请求</button>
</div>
</template>
b.代码
<script setup>
import { ref } from 'vue';
import axios from 'axios';
const timeoutId = ref(null);
function debounce(fn, delay) {
return function(...args) {
if (timeoutId.value) clearTimeout(timeoutId.value);
timeoutId.value = setTimeout(() => {
fn(...args);
}, delay);
};
}
function fetchData() {
axios.get('http://api/gcshi) // 使用示例API
.then(response => {
console.log(response.data);
})
}
const debouncedFetchData = debounce(fetchData, 300);
</script>
b.节流
a.代码
<template>
<div>
<button @click="throttledFetchData">请求</button>
</div>
</template>
b.代码
<script setup>
import { ref } from 'vue';
import axios from 'axios';
const lastCall = ref(0);
function throttle(fn, delay) {
return function(...args) {
const now = new Date().getTime();
if (now - lastCall.value < delay) return;
lastCall.value = now;
fn(...args);
};
}
function fetchData() {
axios.get('http://api/gcshi') //
.then(response => {
console.log(response.data);
})
}
const throttledFetchData = throttle(fetchData, 1000);
</script>
c.请求锁定（加laoding状态）
a.代码
<template>
<div>
<button @click="fetchData">请求</button>
</div>
</template>
<script setup>
import { ref } from 'vue';
import axios from 'axios';
const laoding = ref(false);
function fetchData() {
// 接口请求中，直接返回，避免重复请求
if(laoding.value) return
laoding.value = true
axios.get('http://api/gcshi') //
.then(response => {
laoding.value = fasle
})
}
const throttledFetchData = throttle(fetchData, 1000);
</script>
b.说明
请求锁定非常好理解，设置一个laoding状态，如果第一个接口处于laoding中，那么，我们不执行任何逻辑！
这种方式简单粗暴，十分好用！
但是也有弊端，比如我搜索A后，接口请求中；但我此时突然想搜B，就不会生效了，因为请求A还没响应！
因此，请求锁定这种方式无法取消原先的请求，只能等待一个请求执行完才能继续请求。
d.axios.CancelToken取消重复请求
a.说明
axios其实内置了一个取消重复请求的方法：axios.CancelToken
我们可以利用axios.CancelToken来取消重复的请求
b.代码
<template>
<div>
<button @click="fetchData">请求</button>
</div>
</template>
<script setup>
import { ref } from 'vue';
import axios from 'axios';
let cancelTokenSource = null;
function fetchData() {
if (cancelTokenSource) {
cancelTokenSource.cancel('取消上次请求');
cancelTokenSource = null;
}
cancelTokenSource = axios.CancelToken.source();
axios.get('http://api/gcshi',{cancelToken: cancelTokenSource.token}) //
.then(response => {
laoding.value = fasle
})
}
</script>
5.3 限流算法：5+20=25
00.总结
a.分类1
固定窗口计数器
滑动窗口日志
滑动窗口计数器
漏桶
令牌桶
b.分类2
自适应限流
计数器算法
响应速率限制
队列算法
优先级队列
突发限流
并发数限流
渐进式限流
预测性限流
分布式限流
客户端限流
优雅降级
熔断器
时间窗口限流
峰值限流
QoS
动态限流
流量整形
反向代理限流
负载均衡器限流
01.分类1
a.固定窗口计数器
a.原理
维护一个计数器，将单位时间段作为一个窗口，记录该窗口内接收请求的次数
当请求次数少于限流阀值时，允许访问并增加计数器
当请求次数超过限流阀值时，拒绝访问
时间窗口结束后，计数器清零
b.示例
假设单位时间是1秒，限流阀值为3。在单位时间1秒内，每来一个请求,计数器就加1
如果计数器累加的次数超过限流阀值3，后续的请求全部拒绝。等到1s结束后，计数器清0，重新开始计数
c.临界问题
存在临界问题，即在窗口边界时可能出现短时间内请求数超过阀值的情况
假设限流阀值为5个请求，单位时间窗口是1s
如果我们在单位时间内的前0.8-1s和1-1.2s，分别并发5个请求。虽然都没有超过阀值
但是如果算0.8-1.2s,则并发数高达10，已经超过单位时间1s不超过5阀值的定义啦
b.滑动窗口计数器
a.原理
将单位时间周期分为多个小周期，分别记录每个小周期内的请求次数，并根据时间滑动删除过期的小周期
每个小周期有独立的计数器，时间窗口滑动时更新计数
解决了固定窗口的临界问题，通过更精细的时间划分实现更平滑的限流
它将单位时间周期分为n个小周期，分别记录每个小周期内接口的访问次数，并且根据时间滑动删除过期的小周期
b.示例
假设单位时间还是1s，滑动窗口算法把它划分为5个小周期，也就是滑动窗口（单位时间）被划分为5个小格子
每格表示0.2s。每过0.2s，时间窗口就会往右滑动一格。然后呢，每个小周期，都有自己独立的计数器
如果请求是0.83s到达的，0.8~1.0s对应的计数器就会加1
c.如何解决临界问题
假设我们1s内的限流阀值还是5个请求，0.8~1.0s内（比如0.9s的时候）来了5个请求，落在黄色格子里
时间过了1.0s这个点之后，又来5个请求，落在紫色格子里
如果是固定窗口算法，是不会被限流的，但是滑动窗口的话，每过一个小周期，它会右移一个小格
过了1.0s这个点后，会右移一小格，当前的单位时间段是0.2~1.2s，这个区域的请求已经超过限定的5了
已触发限流啦，实际上，紫色格子的请求都被拒绝啦
TIPS: 当滑动窗口的格子周期划分的越多，那么滑动窗口的滚动就越平滑，限流的统计就会越精确
d.存在问题
虽然解决了临界问题，但在达到限流后，仍会直接拒绝请求，可能导致用户体验不佳
c.漏桶
a.原理
模拟注水漏水过程，以任意速率流入水（请求），以固定速率流出水（处理请求）
桶的容量表示系统能处理的最大请求数，超过容量的请求被丢弃
保证了请求处理的恒定速率，适合平稳流量控制
b.示例
流入的水滴，可以看作是访问系统的请求，这个流入速率是不确定的
桶的容量一般表示系统所能处理的请求数
如果桶的容量满了，就达到限流的阀值，就会丢弃水滴（拒绝请求）
流出的水滴，是恒定过滤的，对应服务按照固定的速率处理请求
c.问题
面对突发流量时，处理速率不变，可能导致请求积压，无法快速响应
d.令牌桶（Guava的RateLimiter限流组件）
a.原理
面对突发流量的时候，我们可以使用令牌桶算法限流
b.令牌桶算法原理
有一个令牌管理员，根据限流大小，定速往令牌桶里放令牌
如果令牌数量满了，超过令牌桶容量的限制，那就丢弃
系统在接受到一个用户请求时，都会先去令牌桶要一个令牌。如果拿到令牌，那么就处理这个请求的业务逻辑
如果拿不到令牌，就直接拒绝这个请求
c.问题
如果令牌发放的策略正确，这个系统即不会被拖垮，也能提高机器的利用率
Guava的RateLimiter限流组件，就是基于令牌桶算法实现的
02.分类2
a.自适应限流（Adaptive Throttling）
a.原理
根据系统的当前负载情况动态调整限流阈值，通常结合系统的响应时间、错误率等指标。
b.优点
可以根据系统实时状况自适应调整，更加智能。
c.缺点
算法复杂，需要综合多个指标和系统反馈。
b.计数器算法（Counter）
a.原理
在一个给定的时间窗口内计数，并在达到限制阈值时拒绝服务。它不考虑请求的到达时间间隔，只关注总数。
b.优点
简单易实现。
c.缺点
可能会在时间窗口切换时出现请求的瞬间峰值。
c.响应速率限制（Rate Limiting）
a.原理
与令牌桶类似，但通常更关注于限制给定时间内的请求速率，而不是单次请求。可以通过HTTP头或其他机制向客户端传达剩余请求配额和重置时间。
b.优点
可以在API等级别精细地控制请求的速率。
c.缺点
需要客户端理解和遵守限流信息，否则可能导致服务不可用。
d.队列算法（Queuing）
a.原理
通过将到达的请求放入队列中进行缓冲，然后以固定的速率处理队列中的请求。如果队列满了，新的请求将被拒绝或排队等待。
b.优点
可以平滑处理请求，避免突发流量导致的服务崩溃。
c.缺点
增加了请求的延迟时间，队列的管理也可能成为系统的瓶颈。
e.优先级队列（Priority Queueing）
a.原理
基于请求的优先级将它们放入不同的队列。高优先级的请求会被优先处理，而低优先级的请求可能在高负载时被延迟或丢弃。
b.优点
允许对不同类型的流量进行差别化管理。
c.缺点
需要一个公平且有效的方式来确定请求的优先级。
f.突发限流（Burst Control）
a.原理
在正常的限流基础上，允许在短时间内处理超过正常速率的请求，但是超出部分会被记录并在后续时间内“偿还”。
b.优点
允许一定程度的突发流量，同时保持长期的流量控制。
c.缺点
实现相对复杂，需要跟踪和计算“借贷”情况。
g.并发数限流（Concurrency Limiting）
a.原理
限制同时处理的请求的数量。当达到并发限制时，新的请求可以被拒绝或排队。
b.优点
可以直接控制系统的并发负载，防止过载。
c.缺点
可能不适用于处理时间波动较大的请求。
h.渐进式限流（Progressive Throttling）
a.原理
根据系统的当前负载逐步增加限流的强度。例如，随着CPU使用率的提高，限流阈值会逐渐降低。
b.优点
能够根据系统负载动态调整限流策略，避免突然的流量削峰。
c.缺点
需要准确的系统负载指标，实现起来相对复杂。
i.预测性限流（Predictive Throttling）
a.原理
通过历史数据和趋势分析预测未来流量，并据此调整限流规则，以提前防止系统过载。
b.优点
可以提前对潜在的流量高峰做出反应，避免系统崩溃。
c.缺点
需要复杂的数据分析和预测模型，且对未来事件的预测可能不准确。
j.分布式限流（Distributed Rate Limiting）
a.原理
在分布式系统中，限流策略需要在多个节点之间同步。这通常通过中央存储（如Redis）或分布式协调服务（如ZooKeeper）来实现。
b.优点
可以在分布式环境中统一限流策略，确保整个系统的流量控制。
c.缺点
实现复杂，需要处理分布式环境中的同步和一致性问题。
k.客户端限流（Client-Side Throttling）
a.原理
限流逻辑部署在客户端，客户端根据服务器提供的限流信息（如HTTP响应头中的RateLimit字段）自行控制请求发送的速率。
b.优点
减轻服务器端的压力，分散限流逻辑。
c.缺点
依赖客户端遵守限流规则，客户端实现可能不一致。
l.优雅降级（Graceful Degradation）
a.原理
在系统负载过高时，主动降低服务质量（如关闭某些非核心功能），以保证核心功能的正常运行。
b.优点
能够在系统资源紧张时保障关键业务的稳定性。
c.缺点
降级策略可能会影响用户体验，需要仔细设计。
m.熔断器（Circuit Breaker）
a.原理
当服务失败率超过一定阈值时，熔断器会暂时中断服务，防止故障扩散。在一段时间后，熔断器会尝试恢复服务。
b.优点
可以防止系统雪崩，快速响应系统故障。
c.缺点
并非严格意义上的限流，但与限流策略结合使用可以提高系统的鲁棒性。
n.时间窗口限流（Time Window Rate Limiting）
a.原理
在一个滚动的时间窗口内对请求进行计数，与固定窗口类似，但是窗口在时间轴上是连续滚动的，而不是固定切换。
b.优点
比固定窗口限流更平滑，避免了固定窗口切换时的流量峰值问题。
c.缺点
实现相对复杂，需要精确的时间管理和同步。
o.峰值限流（Peak Rate Limiting）
a.原理
设置一个峰值速率限制，当请求速率超过峰值时，超出部分的请求将被限制。
b.优点
能够限制流量的峰值，防止短时间内的过载。
c.缺点
可能导致请求在达到峰值时突然被限制，用户体验受影响。
p.QoS（Quality of Service）限流
a.原理
根据服务质量要求对请求进行分类和优先级排序，确保高优先级的请求得到满足，而低优先级的请求在资源紧张时被限制。
b.优点
可以实现差异化服务，保证关键业务的流量需求。
c.缺点
需要对请求进行分类和评估，增加了系统的复杂性。
q.动态限流（Dynamic Throttling）
a.原理
根据实时流量状况或系统负载动态调整限流阈值，可以是基于算法的自动调整，也可以是基于人工实时监控的手动调整。
b.优点
灵活性高，能够适应流量波动和系统状态变化。
c.缺点
可能需要复杂的监控系统和自动化工具支持。
r.流量整形（Traffic Shaping）
a.原理
通过控制数据包的发送速率和时机来调整流量的传输，常用于网络层面的流量管理。
b.优点
可以平滑网络流量，减少拥塞和延迟。
c.缺点
实现通常依赖于网络设备或协议层的支持。
s.反向代理限流
a.原理
在反向代理层面（如Nginx、HAProxy）实现限流，通过配置规则来控制后端服务的流量。
b.优点
部署简单，不需要修改应用逻辑，可以集中管理。
c.缺点
可能无法处理复杂的业务逻辑，限流粒度有限。
t.负载均衡器限流
a.原理
在负载均衡器（如AWS ELB、Google Cloud Load Balancer）中实现限流，根据负载均衡器的策略对流量进行分配和控制。
b.优点
可以在多个服务器间分散流量，避免单点过载。
c.缺点
限流逻辑通常较为简单，可能无法满足复杂的业务需求。
5.4 防抖节流、限流幂等性
00.汇总
a.防抖
控制事件的【触发次数】
b.节流
限制事件的【执行次数】
限流：主要用于后端，，保护系统资源和稳定性
c.限流
限制控制的【请求速率】
节流：主要用于前端，，优化性能
e.幂等性与防抖、节流、限流不同
确保操作结果的一致性，防止重复执行带来的副作用
01.前端
a.防抖（Debounce）
a.定义
防抖是一种在事件触发后等待一段时间，如果在等待时间内事件没有再次触发，则执行事件处理函数
如果事件在等待时间内再次触发，则重新计时
b.应用场景
防止在短时间内多次触发同一事件，例如搜索框输入、窗口调整大小等
c.目的
减少高频事件的触发次数，优化性能
d.代码示例
function debounce(func, wait) {
let timeout;
return function(...args) {
clearTimeout(timeout);
timeout = setTimeout(() => func.apply(this, args), wait);
};
}
const handleResize = debounce(() => {
console.log('Window resized');
}, 300);
window.addEventListener('resize', handleResize);
b.节流（Throttle）
a.定义
节流是一种在一定时间间隔内只允许事件触发一次的方法
即使在时间间隔内事件多次触发，也只执行一次事件处理函数
b.应用场景
限制高频事件的执行次数，例如滚动事件、鼠标移动事件等
c.目的
控制事件触发频率，优化性能
d.代码示例
function throttle(func, limit) {
let inThrottle;
return function(...args) {
if (!inThrottle) {
func.apply(this, args);
inThrottle = true;
setTimeout(() => inThrottle = false, limit);
}
};
}
const handleScroll = throttle(() => {
console.log('Scrolled');
}, 1000);
window.addEventListener('scroll', handleScroll);
02.后端
a.限流（Rate Limiting）
a.定义
限流是一种控制请求速率的技术，用于限制在一定时间窗口内允许的请求数量
b.应用场景
防止服务过载、保护后端资源、避免滥用API等
c.目的
确保系统稳定性和可用性，防止资源耗尽
d.代码示例（使用Spring Boot和Bucket4j）
@RestController
public class RateLimitController {
private final Bucket bucket;
public RateLimitController() {
Bandwidth limit = Bandwidth.classic(10, Refill.greedy(10, Duration.ofMinutes(1)));
this.bucket = Bucket4j.builder().addLimit(limit).build();
}
@GetMapping("/api/resource")
public ResponseEntity<String> getResource() {
if (bucket.tryConsume(1)) {
return ResponseEntity.ok("Resource accessed");
} else {
return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS).body("Too many requests");
}
}
}
b.幂等性（Idempotency）
a.定义
幂等性是指某个操作无论执行多少次，产生的效果都是相同的
b.应用场景
确保在网络重试、重复请求等情况下，操作结果的一致性，例如支付操作、订单创建等
c.目的
保证操作的安全性和一致性，防止重复执行带来的副作用
d.代码示例（使用Spring Boot）
@RestController
public class IdempotentController {
private final Set<String> processedRequests = ConcurrentHashMap.newKeySet();
@PostMapping("/api/order")
public ResponseEntity<String> createOrder(@RequestBody OrderRequest request) {
if (processedRequests.contains(request.getId())) {
return ResponseEntity.status(HttpStatus.CONFLICT).body("Duplicate request");
}
processedRequests.add(request.getId());
// Process the order
return ResponseEntity.ok("Order created");
}
}
5.5 [1]自定义注解+AOP：过期时间
00.限流
a.核心概念
限流：控制请求速率，防止系统过载，保护后端资源
目的：确保系统稳定性和可用性，防止资源耗尽
过期时间：设置过期时间来控制请求速率，确保限流规则在一定时间后自动重置。
b.实现方式
自定义注解：定义一个注解用于标识需要限流的方法
AOP切面：在方法执行前检查请求速率，决定是否允许执行方法
c.核心不同点
限流策略：限流通常涉及计数器、令牌桶、漏桶等算法，用于计算请求速率
实时性：限流需要实时监控请求速率，并在达到限流阈值时立即拒绝请求
01.自定义注解
注解+AOP[string]，设置Redis键并配置过期时间
02.原理
a.定义自定义注解
通过注解标识需要防重复提交的接口
b.实现切面逻辑
使用AOP拦截带有该注解的方法，利用Redis进行去重判断
c.获取请求上下文
确保能够获取请求头中的token
d.提取注解参数
生成Redis键
e.判断Redis键是否存在
决定是否执行目标方法
f.异常处理
确保异常时删除Redis键
03.常用API
a.RequestContextHolder
获取当前请求上下文
b.RedisTemplate
操作Redis数据
c.ProceedingJoinPoint
控制目标方法的执行
d.DigestUtils.sha1DigestAsHex
生成方法签名的SHA1哈希
04.使用步骤
a.定义自定义注解
创建PreventDuplication注解
b.实现切面逻辑
创建PreventDuplicationAspect类，定义切点和环绕通知
c.在控制器中使用注解
将注解应用到需要防重复提交的接口方法上
05.代码示例
a.自定义注解
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface PreventDuplication {
/**
* 防重复操作限时标记数值（存储redis限时标记数值）
*/
String value() default "value" ;
/**
* 防重复操作过期时间（借助redis实现限时控制）
*/
long expireSeconds() default 10;
}
b.自定义切面
@Aspect
@Component
public class PreventDuplicationAspect {
@Autowired
private RedisTemplate redisTemplate;
/**
* 定义切点
*/
@Pointcut("@annotation(com.example.idempotent.idempotent.annotation.PreventDuplication)")
public void preventDuplication() {
}
/**
* 环绕通知 （可以控制目标方法前中后期执行操作，目标方法执行前后分别执行一些代码）
*/
@Around("preventDuplication()")
public Object before(ProceedingJoinPoint joinPoint) throws Exception {
ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder
.getRequestAttributes();
HttpServletRequest request = attributes.getRequest();
Assert.notNull(request, "request cannot be null.");
//获取执行方法
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
//获取防重复提交注解
PreventDuplication annotation = method.getAnnotation(PreventDuplication.class);
// 获取token以及方法标记，生成redisKey和redisValue
String token = request.getHeader(IdempotentConstant.TOKEN);
String redisKey = IdempotentConstant.PREVENT_DUPLICATION_PREFIX
.concat(token)
.concat(getMethodSign(method, joinPoint.getArgs()));
String redisValue = redisKey.concat(annotation.value()).concat("submit duplication");
if (!redisTemplate.hasKey(redisKey)) {
//设置防重复操作限时标记（前置通知）
redisTemplate.opsForValue()
.set(redisKey, redisValue, annotation.expireSeconds(), TimeUnit.SECONDS);
try {
//正常执行方法并返回
//ProceedingJoinPoint类型参数可以决定是否执行目标方法，且环绕通知必须要有返回值，返回值即为目标方法的返回值
return joinPoint.proceed();
} catch (Throwable throwable) {
//确保方法执行异常实时释放限时标记(异常后置通知)
redisTemplate.delete(redisKey);
throw new RuntimeException(throwable);
}
} else {
throw new RuntimeException("请勿重复提交");
}
}
/**
* 生成方法标记：采用数字签名算法SHA1对方法签名字符串加签
*/
private String getMethodSign(Method method, Object... args) {
StringBuilder sb = new StringBuilder(method.toString());
for (Object arg : args) {
sb.append(toString(arg));
}
return DigestUtils.sha1DigestAsHex(sb.toString());
}
private String toString(Object arg) {
if (Objects.isNull(arg)) {
return "null";
}
if (arg instanceof Number) {
return arg.toString();
}
return JSONObject.toJSONString(arg);
}
}
-----------------------------------------------------------------------------------------------------
public interface IdempotentConstant {
String TOKEN = "token";
String PREVENT_DUPLICATION_PREFIX = "PREVENT_DUPLICATION_PREFIX:";
}
c.controller实现
@Slf4j
@RestController
@RequestMapping("/web")
public class IdempotentController {
@PostMapping("/sayNoDuplication")
@PreventDuplication(expireSeconds = 8)
public String sayNoDuplication(@RequestParam("requestNum") String requestNum) {
log.info("sayNoDuplicatin requestNum：{}", requestNum);
return "sayNoDuplicatin".concat(requestNum);
}
}
5.6 [2]令牌桶算法：string结构
00.思路
a.定义自定义注解
创建一个注解（如 @RateLimiter），用于标识需要限流的接口
并配置限流参数（如限流标识、每秒生成的令牌数、令牌桶容量等）
b.实现切面逻辑
使用 AOP 拦截带有 @RateLimiter 注解的方法，在方法执行前进行限流判断。
如果获取到令牌，则允许方法执行；否则，拒绝请求并抛出异常
1.获取注解参数：从 @RateLimiter 注解中获取 key、rate 和 capacity
2.初始化令牌桶：使用 Redisson 的 getRateLimiter 获取或创建一个分布式令牌桶，并设置其速率和容量
3.获取令牌：尝试从令牌桶中获取一个令牌，如果获取成功，则允许方法执行；否则，抛出异常，拒绝请求
4.继续执行方法：如果成功获取到令牌，调用 proceed() 执行原方法
c.配置Redisson客户端
通过配置文件或代码方式，连接到 Redis 服务器，利用 Redisson 提供的分布式令牌桶实现限流
d.应用注解到接口
在需要限流的接口方法上添加 @RateLimiter 注解，指定相应的限流参数
01.定义
令牌桶算法：通过Redisson的AtomicLong对象维护令牌桶，控制令牌的生成和消耗
02.原理
初始化令牌桶：设置令牌桶的初始令牌数量
请求令牌：每次请求消耗一个令牌
生成令牌：定期生成新的令牌
03.常用API
RAtomicLong：Redisson提供的原子计数器
getAndDecrement()：获取并减少令牌数量
addAndGet()：增加令牌数量
04.使用步骤
获取Redisson客户端：初始化Redisson客户端
获取AtomicLong对象：通过Redisson客户端获取AtomicLong对象
请求令牌：调用getAndDecrement()方法消耗令牌
生成令牌：定期调用addAndGet()方法增加令牌
05.代码示例
a.引入依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.20.0</version>
</dependency>
b.创建限流注解
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RateLimiter {
String key() default ""; // 限流的唯一标识
long rate() default 1; // 每秒生成的令牌数
long capacity() default 5; // 令牌桶的容量
}
c.实现限流逻辑
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.redisson.api.RateIntervalUnit;
import org.redisson.api.RateType;
import org.redisson.api.RateLimiter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Aspect
@Component
public class RateLimiterAspect {
@Autowired
private RedissonClient redissonClient;
@Around("@annotation(rateLimiter)")
public Object around(ProceedingJoinPoint joinPoint, RateLimiter rateLimiter) throws Throwable {
String key = rateLimiter.key();
long rate = rateLimiter.rate();
long capacity = rateLimiter.capacity();
// 初始化令牌桶
RateLimiter limiter = redissonClient.getRateLimiter(key);
limiter.trySetRate(RateType.OVERALL, rate, 1, RateIntervalUnit.SECONDS);
limiter.setRate(RateType.OVERALL, rate, capacity, RateIntervalUnit.SECONDS);
// 尝试获取令牌
boolean acquired = limiter.tryAcquire(1);
if (!acquired) {
throw new RuntimeException("请求过于频繁，请稍后再试！");
}
// 继续执行方法
return joinPoint.proceed();
}
}
d.配置 RedissonClient
redisson:
address: redis://127.0.0.1:6379
e.使用限流注解
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/api")
public class TestController {
@RateLimiter(key = "testKey", rate = 2, capacity = 5)
@GetMapping("/test")
public String test() {
return "请求成功";
}
}
-----------------------------------------------------------------------------------------------------
@RateLimiter 注解：用于定义限流策略，包括 key（限流标识）、rate（每秒生成的令牌数）、capacity（令牌桶的最大容量）
RateLimiterAspect 切面：实现限流逻辑，判断是否获取令牌，未获取则抛出异常
RedissonClient：负责连接 Redis 并操作令牌桶
5.7 [3]滑动窗口限流：zset结构+lua脚本
00.思路
滑动窗口限流的基本思路是根据给定的时间窗口内的请求数量来判断是否超出限制
滑动窗口限流将时间划分为更小的窗口，每次请求都会记录时间戳，并清理掉超出窗口时间的请求
Redis + Lua脚本组合可以帮助我们实现这种限流算法，Lua脚本的优势在于它在Redis内具有原子性
01.定义
滑动窗口限流：使用Redisson的RScoredSortedSet记录和检查请求时间戳，以计算在滑动时间窗口内的请求数
02.原理
记录请求时间戳：每次请求记录时间戳
检查请求数：在滑动时间窗口内计算请求数
限流判断：根据请求数判断是否限流
03.常用API
RScoredSortedSet：Redisson提供的有序集合
add()：添加请求时间戳
count()：获取滑动窗口内的请求数
04.使用步骤
获取Redisson客户端：初始化Redisson客户端
获取RScoredSortedSet对象：通过Redisson客户端获取RScoredSortedSet对象
记录请求时间戳：调用add()方法记录请求时间戳
检查请求数：调用count()方法获取请求数
限流判断：根据请求数判断是否限流
05.代码实现
a.Lua脚本限流算法逻辑
1.获取当前时间戳
2.计算滑动窗口的起始时间戳
3.删除小于起始时间戳的记录，确保滑动窗口内的数据不会累积
4.统计滑动窗口内的请求数量
5.如果请求数未超出限流阈值，则记录当前请求，并返回通过；否则，返回限流提示
b.Lua脚本实现
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
-- 删除过期的记录
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
-- 统计当前窗口内的请求数
local count = redis.call('ZCARD', key)
if count < limit then
-- 插入当前请求
redis.call('ZADD', key, now, now)
-- 设置key的过期时间为窗口时间，以防key长期存在
redis.call('EXPIRE', key, window)
return 1 -- 表示请求通过
else
return 0 -- 表示限流
end
c.java注解
a.创建注解@RateLimit
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RateLimit {
int limit(); // 每个时间窗口允许的最大请求数
int window(); // 时间窗口大小（秒）
}
b.将滑动窗口限流算法的核心逻辑封装到一个服务类中，该类会通过Lua脚本在Redis中执行限流操作
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
@Service
public class RateLimiterService {
private static final String SCRIPT = "local key = KEYS[1]\n" +
"local limit = tonumber(ARGV[1])\n" +
"local window = tonumber(ARGV[2])\n" +
"local now = tonumber(ARGV[3])\n" +
"redis.call('ZREMRANGEBYSCORE', key, 0, now - window)\n" +
"local count = redis.call('ZCARD', key)\n" +
"if count < limit then\n" +
" redis.call('ZADD', key, now, now)\n" +
" redis.call('EXPIRE', key, window)\n" +
" return 1\n" +
"else\n" +
" return 0\n" +
"end";
private final JedisPool jedisPool;
public RateLimiterService(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public boolean tryAcquire(String key, int limit, int window) {
try (Jedis jedis = jedisPool.getResource()) {
long now = System.currentTimeMillis();
Object result = jedis.eval(SCRIPT, 1, key, String.valueOf(limit), String.valueOf(window * 1000), String.valueOf(now));
return Integer.parseInt(result.toString()) == 1;
}
}
}
c.创建限流拦截器
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Aspect
@Component
public class RateLimitAspect {
@Autowired
private RateLimiterService rateLimiterService;
@Around("@annotation(rateLimit)")
public Object rateLimit(ProceedingJoinPoint joinPoint, RateLimit rateLimit) throws Throwable {
String key = "rate:limiter:" + joinPoint.getSignature().toShortString();
int limit = rateLimit.limit();
int window = rateLimit.window();
boolean allowed = rateLimiterService.tryAcquire(key, limit, window);
if (allowed) {
return joinPoint.proceed();
} else {
throw new RuntimeException("请求频率过高，请稍后再试");
}
}
}
d.使用示例
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class UserController {
@RateLimit(limit = 5, window = 60) // 每分钟最多5次请求
@GetMapping("/like")
public String like(@RequestParam String userId) {
return "点赞成功";
}
}
06.代码实现
package com.ruoyi.redis4.demo04;
import java.time.Instant;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
public class Demo02 {
private static final JedisPool pool = new JedisPool("localhost", 6379);
// Lua 脚本内容
private static final String LUA_SCRIPT =
"local key = KEYS[1] " +
"local limit = tonumber(ARGV[1]) " +
"local window = tonumber(ARGV[2]) " +
"local now = tonumber(ARGV[3]) " +
" " +
"-- 移除过期的记录 " +
"redis.call('ZREMRANGEBYSCORE', key, 0, now - window) " +
" " +
"-- 获取当前窗口的请求数量 " +
"local count = redis.call('ZCARD', key) " +
" " +
"if count < limit then " +
" -- 如果没有超过限制，将当前时间作为分数添加到集合中 " +
" redis.call('ZADD', key, now, now) " +
" -- 设置集合的过期时间为窗口大小 " +
" redis.call('EXPIRE', key, window / 1000) " +
" return 1 -- 请求通过 " +
"else " +
" return 0 -- 请求被限流 " +
"end";
public static boolean isAllowed(String key, int limit, int windowInMillis) {
try (Jedis jedis = pool.getResource()) {
long now = Instant.now().toEpochMilli();
Object result = jedis.eval(LUA_SCRIPT, 1, key, String.valueOf(limit), String.valueOf(windowInMillis), String.valueOf(now));
return result.equals(1L);
}
}
public static void main(String[] args) {
// 测试滑动窗口限流
for (int i = 0; i < 10; i++) {
boolean allowed = isAllowed("user:123", 5, 60000); // 限制每60秒最多5次请求
System.out.println(allowed ? "请求通过" : "请求被限流");
try {
Thread.sleep(1000); // 模拟请求间隔
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
5.8 [4]漏斗算法：string结构
01.定义
漏斗算法：使用Redisson的AtomicLong对象来控制请求的流入速率
02.原理
初始化漏斗：设置漏斗的初始状态
请求处理：根据漏斗状态判断是否处理请求
更新漏斗状态：定期更新漏斗状态
03.常用API
RAtomicLong：Redisson提供的原子计数器
get()：获取漏斗的状态
set()：设置漏斗的状态
04.使用步骤
获取Redisson客户端：初始化Redisson客户端
获取AtomicLong对象：通过Redisson客户端获取AtomicLong对象
请求处理：调用get()方法获取漏斗状态，判断是否可以处理请求
更新漏斗状态：定期调用set()方法更新漏斗状态
05.代码示例
import org.redisson.api.RAtomicLong;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
public class FunnelRateLimiterExample {
@Autowired
private RedissonClient redissonClient;
public boolean isAllowed(String funnelKey, long rateLimit) {
RAtomicLong lastRequestTime = redissonClient.getAtomicLong(funnelKey);
long currentTime = System.currentTimeMillis();
if (lastRequestTime.get() == 0 || currentTime - lastRequestTime.get() >= rateLimit) {
lastRequestTime.set(currentTime);
return true;
}
return false;
}
}
5.9 [5]setnx+lua脚本：string结构
00.汇总
1.依赖
2.配置
3.创建限流类型
4.创建限流注解
5.编写限流的Lua脚本
6.配置RedisConfig
7.编写限流切面RateLimitAspect
8.使用注解
01.依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
</dependencies>
02.配置
spring:
redis: host: 127.0.0.1
port: 6379
database: 0
password:
timeout: 10s
lettuce:
pool:
min-idle: 0 #连接池中的最小空闲连接数为 0。这意味着在没有任何请求时，连接池可以没有空闲连接。
max-idle: 8 #连接池中的最大空闲连接数为 8。当连接池中的空闲连接数超过这个值时，多余的连接可能会被关闭以节省资源。
max-active: 8 #连接池允许的最大活动连接数为 8。在并发请求较高时，连接池最多可以创建 8 个连接来满足需求。
max-wait: -1ms #当连接池中的连接都被使用且没有空闲连接时，新的连接请求等待获取连接的最大时间。这里设置为 -1ms，表示无限等待，直到有可用连接为止。
03.创建限流类型
public enum LimitType {
/** * 针对某一个ip进行限流 */
IP("IP") ;
private final String type;
LimitType(String type) {
this.type = type;
}
public String getType() {
return type;
}
}
04.创建限流注解
import java.lang.annotation.*;
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
/** * 限流类型 * @return */
LimitType limitType() default LimitType.IP;
/** * 限流key * @return */
String key() default "";
/** * 限流时间 * @return */
int time() default 60;
/** * 限流次数 * @return */
int count() default 100;
}
05.编写限流的Lua脚本
local key = KEYS[1]
local time = tonumber(ARGV[1])
local count = tonumber(ARGV[2])
local current = redis.call('get', key)
if current and tonumber(current) > count then
return tonumber(current)
end
current = redis.call('incr', key)
if tonumber(current) == 1 then
redis.call('expire', key, time)
end
return tonumber(current)
06.配置RedisConfig
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scripting.support.ResourceScriptSource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
@Configuration public class RedisConfig {
/** * RedisTemplate配置 * * @param factory * @return */
@Bean
RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(factory);
StringRedisSerializer serializer = new StringRedisSerializer(StandardCharsets.UTF_8);
// 使用StringRedisSerializer来序列化和反序列化redis的key值
template.setKeySerializer(serializer);
template.setValueSerializer(serializer);
template.setHashKeySerializer(serializer);
template.setHashValueSerializer(serializer);
template.afterPropertiesSet();
return template;
}
/** * Redis Lua 脚本 * * @return */
@Bean DefaultRedisScript<Long> limitScript() {
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
script.setResultType(Long.class);
// 我这里是以资源文件的形式来加载的lua脚本
script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/limit.lua")));
return script;
}
}
07.编写限流切面RateLimitAspect
import com.example.luatest.annotition.RateLimiter;
import com.example.luatest.exception.IPException;
import lombok.extern.java.Log;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import java.util.Collections;
import java.util.List;
@Aspect
@Component //切面类也需要加入到ioc容器
public class RateLimitAspect {
private static final Logger LOGGER = LoggerFactory.getLogger(RateLimitAspect.class);
private final RedisTemplate<String, Object> redisTemplate;
private final DefaultRedisScript<Long> limitScript;
public RateLimitAspect(RedisTemplate<String, Object> redisTemplate, DefaultRedisScript<Long> limitScript) {
this.redisTemplate = redisTemplate;
this.limitScript = limitScript;
}
@Before("@annotation(rateLimiter)")
public void isAllowed(JoinPoint proceedingJoinPoint, RateLimiter rateLimiter) throws IPException, InstantiationException, IllegalAccessException {
String ip = null;
Object[] args = proceedingJoinPoint.getArgs();
for (Object arg : args) {
if (arg instanceof HttpServletRequest) {
HttpServletRequest request = (HttpServletRequest) arg;
ip = request.getRemoteHost(); break;
}
}
LOGGER.info("ip:{}", ip);
if (ip == null) {
throw new IPException("ip is null");
}
//拼接redis建
String key = rateLimiter.key() + ip;
// 执行 Lua 脚本进行限流判断
List<String> keyList = Collections.singletonList(key);
Long result = redisTemplate.execute(limitScript, keyList, key, Integer.toString(rateLimiter.count()), Integer.toString(rateLimiter.time()));
LOGGER.info("result:{}", result);
if (result != null && result > rateLimiter.count()) {
throw new IPException("IP [" + ip + "] 访问过于频繁，已超出限流次数");
}
}
}
08.使用注解
import com.example.luatest.annotition.RateLimiter;
import com.example.luatest.enum_.LimitType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.servlet.http.HttpServletRequest;
@RestController
@RequestMapping("/rate")
public class RateController {
@RateLimiter(count = 100, time = 60, limitType = LimitType.IP) @GetMapping("/someMethod")
public void someMethod(HttpServletRequest request) {
// 方法的具体逻辑
}
}
6 幂等性：同一时间只有一个请求
6.1 汇总：6个
00.汇总
a.自定义注解和AOP
自定义注解结合AOP（面向切面编程）是一种优雅的方式来防止重复提交
通过在方法上添加自定义注解，并在AOP切面中拦截请求，可以有效地防止重复提交
b.Redisson分布式锁
在分布式环境中，可以使用Redis实现分布式锁来防止重复提交
通过在请求处理前获取锁，确保同一时间只有一个请求能够被处理，从而避免重复提交的问题
c.setnx+Lua脚本
通过Redis的SETNX命令结合Lua脚本实现原子性操作，确保在请求处理前成功获取锁，避免重复提交
d.zookeeper分布式锁
利用Zookeeper的临时有序节点特性实现分布式锁，确保同一时间只有一个客户端持有锁，从而防止重复提交
e.前端拦截
在前端，通过禁用按钮或隐藏按钮来防止用户重复提交请求
这种方法简单直接，但只能防止正常操作下的重复提交，对于恶意请求无效
f.后端拦截
HashMap：在请求处理前检查是否已处理过相同的请求ID
固定大小数组：用于存储最近处理过的请求ID
双重检测锁（DCL）：结合锁机制来确保线程安全
LRUMap：使用最近最少使用算法来管理请求ID
01.客户端控制
a.按钮只可操作一次
最简单的方法是将按钮设置为只可操作一次
用户点击提交按钮后，立即将按钮置为不可用状态或显示加载动画，防止用户重复点击
这种方法简单易行，但只能解决用户误操作的问题，无法应对网络波动或重试机制导致的重复请求
b.Token机制
Token 机制是一种常见的解决方案
用户在进入页面时，后端生成一个唯一的 Token 并存储在 Session 或数据库中，同时将 Token 返回给前端
前端在每次请求时将 Token 作为参数发送给后端，后端在处理请求前验证 Token 是否有效
如果 Token 有效，则处理请求并将其标记为已使用；如果 Token 无效，则返回错误提示
这种方法可以有效防止重复提交，但需要后端和前端配合实现
c.使用Post/Redirect/Get模式
Post/Redirect/Get（PRG）模式是一种常见的解决方案
用户提交表单后，后端处理请求并返回一个重定向响应，将用户引导到一个信息页面
这样，即使用户刷新页面或使用浏览器的前进/后退按钮，也不会重复提交表单
这种方法可以有效防止重复提交，不过同样需要后端和前端配合实现
02.服务端控制
a.使用唯一索引
利用数据库的唯一索引机制可以有效防止重复数据插入
例如，订单表可以使用订单号作为唯一索引，当重复请求尝试插入相同订单号的数据时
数据库会抛出唯一索引冲突的异常，从而阻止重复数据插入
这种方法简单易行，但只能防止重复插入，无法处理更新操作
b.乐观锁
乐观锁是一种常见的并发控制机制
在表结构中添加一个版本号字段（version），每次更新数据时，版本号加一
在处理更新请求时，后端会检查版本号是否一致，如果版本号一致，则更新数据并增加版本号，如果版本号不一致，则返回错误提示
这种方法可以有效防止并发更新导致的重复操作，但需要在表结构中添加额外字段
c.Select + Insert/Update/Delete
在操作之前先查询数据库，判断是否已经存在相同的数据
如果不存在，则执行插入或更新操作
这种方法在单 JVM 环境中可以有效防止重复操作，但在分布式环境中需要结合分布式锁来保证幂等性
d.分布式锁
在分布式系统中，可以使用分布式锁来保证幂等性
例如，使用 Redis 或 Zookeeper 实现分布式锁
在处理请求时，后端会尝试获取分布式锁，如果获取成功，则处理请求并释放锁，如果获取失败，则返回错误提示
这种方法可以有效防止分布式环境下的并发操作，但需要引入第三方系统
e.状态机幂等
在设计单据相关的业务时，可以使用状态机来保证幂等性
例如，订单状态可以分为“待支付”、“已支付”、“已发货”等
如果订单已经处于“已支付”状态，再次调用支付接口时，可以直接返回成功响应，而不进行实际操作
这种方法可以有效防止重复操作，但需要对业务逻辑进行详细设计
f.防重表
防重表是一种常见的解决方案
后端创建一个防重表，记录每个请求的唯一标识符
在处理请求时，后端会检查防重表中是否存在相同的标识符
如果不存在，则插入记录并处理请求；如果存在，则返回错误提示
这种方法可以有效防止重复请求，但需要额外维护防重表
g.缓冲队列
将请求快速接收并放入缓冲队列中，后续使用异步任务处理队列中的数据
在处理过程中，可以过滤掉重复的请求
这种方法可以提高系统的吞吐量，但无法及时返回请求结果，需要后续轮询获取处理结果
h.全局唯一号
在请求中使用全局唯一号（如 UUID）作为标识符
后端根据唯一号判断请求是否重复，如果重复，则返回错误提示；如果不重复，则处理请求并记录唯一号
这种方法可以有效防止重复请求，但需要生成和管理全局唯一号
6.2 幂等性
01.概念
幂等性在 API 设计中指的是无论客户端发送多少次相同的请求，服务器的状态和响应结果都保持一致
换句话说，多次执行相同的操作与一次执行的效果相同。这在防止重复提交（如重复订单创建、重复支付等）时尤为重要
02.幂等性、防抖、去重
防抖：通常用于限制高频率的事件触发，确保在一定时间内只执行一次操作，理解为在短时间内对相同请求只处理一次，忽略重复的请求
去重：指的是识别并忽略重复的请求，以防止同一操作被执行多次。去重是实现幂等性的一个重要手段
因此，幂等性可以看作是通过防抖和去重等技术手段来实现的，以确保接口在面对重复请求时的行为一致性和系统的稳定性
03.哪些请求天生就是幂等的？
首先，我们要知道查询类的请求一般都是天然幂等的，除此之外，删除请求在大多数情况下也是幂等的，但是ABA场景下除外
比如，先请求了一次删除A的操作，但由于响应超时，又自动请求了一次删除A的操作，如果在两次请求之间，又插入了一次A，而实际上新插入的这一次A，是不应该被删除的，这就是ABA问题，不过，在大多数业务场景中，ABA问题都是可以忽略的
除了查询和删除之外，还有更新操作，同样的更新操作在大多数场景下也是天然幂等的，其例外是也会存在ABA的问题，更重要的是，比如执行update table set a = a + 1 where v = 1这样的更新就非幂等了
最后，就还剩插入了，插入大多数情况下都是非幂等的，除非是利用数据库唯一索引来保证数据不会重复产生
04.为什么需要幂等
a.超时重试
当发起一次RPC请求时，难免会因为网络不稳定而导致请求失败，一般遇到这样的问题我们希望能够重新请求一次
正常情况下没有问题，但有时请求实际上已经发出去了，只是在请求响应时网络异常或者超时
此时，请求方如果再重新发起一次请求，那被请求方就需要保证幂等了
b.异步回调
异步回调是提升系统接口吞吐量的一种常用方式，很明显，此类接口一定是需要保证幂等性的
c.消息队列
现在常用的消息队列框架，比如Kafka、RocketMQ、RabbitMQ在消息传递时
都会采取At least once原则（也就是至少一次原则，在消息传递时，不允许丢消息，但是允许有重复的消息）
既然消息队列不保证不会出现重复的消息，那消费者自然要保证处理逻辑的幂等性了
05.实现幂等的关键因素
a.关键因素1
幂等唯一标识，可以叫它幂等号或者幂等令牌或者全局ID，总之就是客户端与服务端一次请求时的唯一标识
一般情况下由客户端来生成，也可以让第三方来统一分配
b.关键因素2
有了唯一标识以后，服务端只需要确保这个唯一标识只被使用一次即可，一种常见的方式就是利用数据库的唯一索引
6.3 [1]自定义注解+AOP：过期时间
00.幂等性
a.核心概念
幂等性：确保某个操作无论执行多少次，产生的效果都是相同的
目的：保证操作的安全性和一致性，防止重复执行带来的副作用
过期时间：设置过期时间来确保幂等性记录不会无限期地保留，从而避免资源浪费
b.实现方式
自定义注解：定义一个注解用于标识需要幂等性控制的方法
AOP切面：在方法执行前检查请求是否已处理过，决定是否允许执行方法
c.核心不同点
幂等性检查：幂等性通常涉及唯一标识符（如请求ID），用于检查请求是否已处理
持久性：幂等性需要记录已处理的请求，以确保重复请求不会再次执行
01.思路
a.定义自定义注解
创建一个自定义注解@Idempotent，用于标识需要进行幂等性控制的方法
注解包含三个属性：name（用于指定从方法参数中获取对象的参数名）、field（用于指定从对象中获取作为幂等性KEY的属性名）、type（用于指定参数的类型）
b.设计请求数据结构
定义一个统一的请求数据结构RequestData，包含两个部分：Header和body
Header中包含一个token字段，用于进行幂等性验证
c.AOP切面处理
使用AOP技术定义一个切面IdempotentAspect，拦截所有标注了@Idempotent注解的方法
在切面中，通过反射机制获取方法的参数值，并根据注解中的name和field属性提取幂等性KEY
使用RedisIdempotentStorage检查该KEY是否已存在于存储中，如果存在则返回“重复请求”，否则继续执行目标方法并在执行后删除该KEY
d.Token生成与存储
提供一个接口IdGeneratorController用于生成幂等性token，并将其存储在Redis中以便后续验证
RedisIdempotentStorage实现了IdempotentStorage接口，负责将生成的token保存到Redis，并在请求处理完成后删除该token
e.请求示例
在调用需要幂等性控制的接口之前，客户端首先通过IdGeneratorController获取一个token
客户端将获取到的token放入请求头中，然后调用目标接口（例如OrderController中的saveOrder方法）
02.代码示例
a.自定义注解
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(value = ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Idempotent {
/**
* 参数名，表示将从哪个参数中获取属性值。
* 获取到的属性值将作为KEY。
*
* @return
*/
String name() default "";
/**
* 属性，表示将获取哪个属性的值。
*
* @return
*/
String field() default "";
/**
* 参数类型
*
* @return
*/
Class type();
}
b.统一的请求入参对象
@Data
public class RequestData<T> {
private Header header;
private T body;
}
@Data
public class Header {
private String token;
}
@Data
public class Order {
String orderNo;
}
c.AOP处理
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.CodeSignature;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
public class AopUtils {
public static Object getFieldValue(Object obj, String name) throws Exception {
Field[] fields = obj.getClass().getDeclaredFields();
Object object = null;
for (Field field : fields) {
field.setAccessible(true);
if (field.getName().toUpperCase().equals(name.toUpperCase())) {
object = field.get(obj);
break;
}
}
return object;
}
public static Map<String, Object> getParamValue(ProceedingJoinPoint joinPoint) {
Object[] paramValues = joinPoint.getArgs();
String[] paramNames = ((CodeSignature) joinPoint.getSignature()).getParameterNames();
Map<String, Object> param = new HashMap<>(paramNames.length);
for (int i = 0; i < paramNames.length; i++) {
param.put(paramNames[i], paramValues[i]);
}
return param;
}
}
-----------------------------------------------------------------------------------------------------
@Aspect
@Component
public class IdempotentAspect {
@Resource
private RedisIdempotentStorage redisIdempotentStorage;
@Pointcut("@annotation(com.springboot.micrometer.annotation.Idempotent)")
public void idempotent() {
}
@Around("idempotent()")
public Object methodAround(ProceedingJoinPoint joinPoint) throws Throwable {
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
Idempotent idempotent = method.getAnnotation(Idempotent.class);
String field = idempotent.field();
String name = idempotent.name();
Class clazzType = idempotent.type();
String token = "";
Object object = clazzType.newInstance();
Map<String, Object> paramValue = AopUtils.getParamValue(joinPoint);
if (object instanceof RequestData) {
RequestData idempotentEntity = (RequestData) paramValue.get(name);
token = String.valueOf(AopUtils.getFieldValue(idempotentEntity.getHeader(), field));
}
if (redisIdempotentStorage.delete(token)) {
return joinPoint.proceed();
}
return "重复请求";
}
}
d.Token值生成
@RestController
@RequestMapping("/idGenerator")
public class IdGeneratorController {
@Resource
private RedisIdempotentStorage redisIdempotentStorage;
@RequestMapping("/getIdGeneratorToken")
public String getIdGeneratorToken() {
String generateId = IdGeneratorUtil.generateId();
redisIdempotentStorage.save(generateId);
return generateId;
}
}
-----------------------------------------------------------------------------------------------------
public interface IdempotentStorage {
void save(String idempotentId);
boolean delete(String idempotentId);
}
-----------------------------------------------------------------------------------------------------
@Component
public class RedisIdempotentStorage implements IdempotentStorage {
@Resource
private RedisTemplate<String, Serializable> redisTemplate;
@Override
public void save(String idempotentId) {
redisTemplate.opsForValue().set(idempotentId, idempotentId, 10, TimeUnit.MINUTES);
}
@Override
public boolean delete(String idempotentId) {
return redisTemplate.delete(idempotentId);
}
}
-----------------------------------------------------------------------------------------------------
public class IdGeneratorUtil {
public static String generateId() {
return UUID.randomUUID().toString();
}
}
e.请求示例
调用接口之前，先申请一个token，然后带着服务端返回的token值，再去请求。
@RestController
@RequestMapping("/order")
public class OrderController {
@RequestMapping("/saveOrder")
@Idempotent(name = "requestData", type = RequestData.class, field = "token")
public String saveOrder(@RequestBody RequestData<Order> requestData) {
return "success";
}
}
03.代码示例
a.注解类Idempotent
@Inherited
@Target(ElementType.METHOD)
@Retention(value = RetentionPolicy.RUNTIME)
public @interface Idempotent {
/**
* 幂等操作的唯一标识，使用spring el表达式 用#来引用方法参数
* @return Spring-EL expression
*/
String key() default "";
/**
* 有效期 默认：1 有效期要大于程序执行时间，否则请求还是可能会进来
* @return expireTime
*/
int expireTime() default 1;
/**
* 时间单位 默认：s
* @return TimeUnit
*/
TimeUnit timeUnit() default TimeUnit.SECONDS;
/**
* 提示信息，可自定义
* @return String
*/
String info() default "重复请求，请稍后重试";
/**
* 是否在业务完成后删除key true:删除 false:不删除
* @return boolean
*/
boolean delKey() default false;
}
-----------------------------------------------------------------------------------------------------
String key();
幂等操作的唯一标识，使用Spring EL表达式 用#来引用方法参数 。 可为空则取 当前 url + args 做表示
-----------------------------------------------------------------------------------------------------
int expireTime() default 1;
有效期 默认：1 有效期要大于程序执行时间，否则请求还是可能会进来
-----------------------------------------------------------------------------------------------------
TimeUnit timeUnit() default TimeUnit.SECONDS;
时间单位 默认：s（秒）
-----------------------------------------------------------------------------------------------------
String info() default "请稍后重试";
幂等失败提示信息，可自定义
-----------------------------------------------------------------------------------------------------
boolean delKey() default false;
是否在业务完成后删除key true:删除 false:不删除
b.定义获取幂等Key的接口类
public interface KeyResolver {
/**
* 解析处理 key
*
* @param idempotent 接口注解标识
* @param point 接口切点信息
* @return 处理结果
*/
String resolver(Idempotent idempotent, JoinPoint point);
}
c.上述接口的实现
public class ExpressionResolver implements KeyResolver {
private static final SpelExpressionParser PARSER = new SpelExpressionParser();
private static final LocalVariableTableParameterNameDiscoverer DISCOVERER = new LocalVariableTableParameterNameDiscoverer();
@Override
public String resolver(Idempotent idempotent, JoinPoint point) {
//获取被拦截方法的所有参数
Object[] arguments = point.getArgs();
//从字节码的局部变量表中解析出参数名称
String[] params = DISCOVERER.getParameterNames(getMethod(point));
//SpEL表达式执行的上下文环境，用于存放变量
StandardEvaluationContext context = new StandardEvaluationContext();
//遍历方法参数名和对应的参数值，将它们一一绑定到StandardEvaluationContext中。
//这样SpEL表达式就可以引用这些参数值
if (params != null && params.length > 0) {
for (int len = 0; len < params.length; len++) {
context.setVariable(params[len], arguments[len]);
}
}
//使用SpelExpressionParser来解析Idempotent注解中的key属性，将其作为SpEL表达式字符串
Expression expression = PARSER.parseExpression(idempotent.key());
//转换结果为String类型返回
return expression.getValue(context, String.class);
}
/**
* 根据切点解析方法信息
*
* @param joinPoint 切点信息
* @return Method 原信息
*/
private Method getMethod(JoinPoint joinPoint) {
//将joinPoint.getSignature()转换为MethodSignature
//Signature是AOP中表示连接点签名的接口，而MethodSignature是它的具体实现，专门用于表示方法的签名。
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
//获取到方法的声明。这将返回代理对象所持有的方法声明。
Method method = signature.getMethod();
//判断获取到的方法是否属于一个接口
//因为在Java中，当通过Spring AOP或其它代理方式调用接口的方法时，实际被执行的对象是一个代理对象，直接获取到的方法可能来自于接口声明而不是实现类。
if (method.getDeclaringClass().isInterface()) {
try {
//通过反射获取目标对象的实际类（joinPoint.getTarget().getClass()）中同名且参数类型相同的方法
//这样做是因为代理类可能对方法进行了增强，直接调用实现类的方法可以确保获取到最准确的实现细节
method = joinPoint.getTarget().getClass().getDeclaredMethod(joinPoint.getSignature().getName(),
method.getParameterTypes());
} catch (SecurityException | NoSuchMethodException e) {
throw new RuntimeException(e);
}
}
return method;
}
}
d.自定义幂等异常类
public class IdempotentException extends RuntimeException {
public IdempotentException() {
super();
}
public IdempotentException(String message) {
super(message);
}
public IdempotentException(String message, Throwable cause) {
super(message, cause);
}
public IdempotentException(Throwable cause) {
super(cause);
}
protected IdempotentException(String message, Throwable cause, boolean enableSuppression,
boolean writableStackTrace) {
super(message, cause, enableSuppression, writableStackTrace);
}
}
e.幂等性切面（AOP）实现类
@Aspect
@Slf4j
public class IdempotentAspect {
private static final ThreadLocal<Map<String, Object>> THREAD_CACHE = ThreadLocal.withInitial(HashMap::new);
private static final String KEY = "key";
private static final String DEL_KEY = "delKey";
@Resource
private Redisson redisson;
@Resource
private KeyResolver keyResolver;
@Pointcut("@annotation(com.XXX.XXX.Idempotent)") //Idempotent注解的包路径
public void pointCut() {
}
@Before("pointCut()")
public void beforePointCut(JoinPoint joinPoint) {
//获取到当前请求的属性，进而得到HttpServletRequest对象，以便后续获取请求URL和参数信息。
ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder
.getRequestAttributes();
HttpServletRequest request = requestAttributes.getRequest();
//从JoinPoint中获取方法签名，并确认该方法是否被@Idempotent注解标记。如果是，则继续执行幂等性检查逻辑；如果不是，则直接返回，不进行幂等处理。。
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
if (!method.isAnnotationPresent(Idempotent.class)) {
return;
}
Idempotent idempotent = method.getAnnotation(Idempotent.class);
String key;
// 若没有配置 幂等 标识编号，则使用 url + 参数列表作为区分;如果提供了key规则，则利用keyResolver根据提供的规则和切点信息生成键
if (!StringUtils.hasLength(idempotent.key())) {
String url = request.getRequestURL().toString();
String argString = Arrays.asList(joinPoint.getArgs()).toString();
key = url + argString;
} else {
// 使用jstl 规则区分
key = keyResolver.resolver(idempotent, joinPoint);
}
//从注解中读取并设置幂等操作的过期时间、描述信息、时间单位以及是否删除键的标志。
long expireTime = idempotent.expireTime();
String info = idempotent.info();
TimeUnit timeUnit = idempotent.timeUnit();
boolean delKey = idempotent.delKey();
//尝试从RMapCache（基于Redis的并发安全映射缓存）中获取键对应的值，如果存在，则说明重复操作，抛出IdempotentException
RMapCache<String, Object> rMapCache = redisson.getMapCache(CommonConstants.RMAP_CACHE_KEY);
String value = LocalDateTime.now().toString().replace("T", " ");
Object v1;
if (null != rMapCache.get(key)) {
throw new IdempotentException(info);
}
//使用synchronized关键字保证多线程环境下操作的原子性，然后使用putIfAbsent方法尝试添加键值对到缓存中，若添加成功（即返回null），则记录日志并继续执行业务逻辑；若添加失败（即键已存在），同样抛出IdempotentException。
synchronized (this) {
v1 = rMapCache.putIfAbsent(key, value, expireTime, timeUnit);
if (null != v1) {
throw new IdempotentException(info);
} else {
log.info("[idempotent]:has stored key={},value={},expireTime={}{},now={}", key, value, expireTime,
timeUnit, LocalDateTime.now().toString());
}
}
//将幂等键和是否删除键的标志存储到THREAD_CACHE（线程局部变量）中，供后续afterPointCut方法使用
Map<String, Object> map = THREAD_CACHE.get();
map.put(KEY, key);
map.put(DEL_KEY, delKey);
}
@After("pointCut()")
public void afterPointCut(JoinPoint joinPoint) {
//尝试从THREAD_CACHE（线程局部变量）中获取之前存储的幂等相关信息，如果为空或者不存在，则直接返回，不做进一步处理
Map<String, Object> map = THREAD_CACHE.get();
if (CollectionUtils.isEmpty(map)) {
return;
}
//检查RMapCache（基于Redis的并发安全映射缓存）是否为空，如果为空（即没有任何条目），表明无需执行任何清理动作，直接返回
RMapCache<Object, Object> mapCache = redisson.getMapCache(CommonConstants.RMAP_CACHE_KEY);
if (mapCache.size() == 0) {
return;
}
//检查RMapCache（基于Redis的并发安全映射缓存）是否为空，如果为空（即没有任何条目），表明无需执行任何清理动作，直接返回
String key = map.get(KEY).toString();
boolean delKey = (boolean) map.get(DEL_KEY);
if (delKey) {
mapCache.fastRemove(key);
log.info("[idempotent]:has removed key={}", key);
}
//无论是否移除了键，最后都会清空当前线程局部变量THREAD_CACHE中的数据，避免内存泄漏
THREAD_CACHE.remove();
}
}
f.自定义一个Spring条件类
对于有些服务不需要做幂等校验，通过这个条件类来控制幂等是否生效，比如：如果是gateway，那就不生效幂等
-----------------------------------------------------------------------------------------------------
public class IdempotentCondition implements Condition {
@Override
public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
//如果服务配置文件中配置spring.cloud.gateway属性，则返回flase，否则为true
return !context.getEnvironment().containsProperty("spring.cloud.gateway") ;
}
}
g.自定义一个Spring自动配置类，幂等插件初始化
//条件评估为true时，该配置类中的bean才会被注册到Spring容器中
@Conditional(IdempotentCondition.class)
//proxyBeanMethods = false表示不使用代理来调用配置类中的bean方法，这样可以提高性能，但如果你在配置类内部依赖于其他@Bean方法的副作用，则需谨慎使用。
@Configuration(proxyBeanMethods = false)
//在RedisAutoConfiguration之后自动配置本类
@AutoConfigureAfter(RedisAutoConfiguration.class)
public class IdempotentAutoConfiguration {
/**
* 切面 拦截处理所有 @Idempotent
* @return Aspect
*/
@Bean
public IdempotentAspect idempotentAspect() {
return new IdempotentAspect();
}
/**
* key 解析器
* @return KeyResolver
*/
@Bean
@ConditionalOnMissingBean(KeyResolver.class)
public KeyResolver keyResolver() {
return new ExpressionResolver();
}
}
h.幂等组件使用示例代码
a.在需要实现幂等的接口上添加以下注解
@Idempotent(key = "#user.id", expireTime = 5, timeUnit=imeUnit.S>ECONDS,info = "请勿重复查询",delKey= true)
b.示例代码
/**
* Map
*
* @param request 请求
* @param response 响应
* @param millis 延时，毫秒
* @return 返回 Map
*/
@SneakyThrows
@Idempotent(key = "#user.id", expireTime = 5, timeUnit=imeUnit.SECONDS,info = "请勿重复查询",delKey= true)
@RequestMapping("/user")
public Map<String, Object> map(HttpServletRequest request, HttpServletResponse response, Long millis,User user) {
if (millis != null) {
// 延时
Thread.sleep(millis);
}
Map<String, Object> map = new HashMap<>(8);
map.put("uuid", UUID.randomUUID().toString());
log.info(String.valueOf(map));
return map;
}
04.代码示例
a.使用Token机制
a.依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
b.生成Token：在Controller中生成Token并存储在Redis中
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;
import java.util.UUID;
@RestController
public class TokenController {
@Autowired
private StringRedisTemplate redisTemplate;
@GetMapping("/formByRedis")
public ModelAndView showFormByRedis() {
ModelAndView form = new ModelAndView("form");
String token = UUID.randomUUID().toString();
// 设置过期时间为5分钟
redisTemplate.opsForValue().set(token, token, 5, TimeUnit.MINUTES);
form.addObject("token", token);
return form;
}
}
c.传递Token：在表单中添加隐藏字段来传递Token
<form action="/base/submitByRedis" method="post">
<input type="hidden" name="token" th:value="${token}">
<!-- 其他表单字段 -->
<button type="submit">Submit</button>
</form>
d.验证Token：在Controller中验证Token
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
@PostMapping("/submitByRedis")
public String handleFormByRedis(@RequestParam String token) {
String redisToken = redisTemplate.opsForValue().get(token);
if (redisToken == null) {
throw new RuntimeException("Duplicate submit detected");
}
// 删除Token
redisTemplate.delete(token);
// 处理表单数据
return "success";
}
b.使用Spring AOP
a.创建切面：创建一个切面类DuplicateSubmitAspect
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
@Aspect
@Component
public class DuplicateSubmitAspect {
@Autowired
private StringRedisTemplate redisTemplate;
@Around("@annotation(preventDuplicateSubmit)")
public Object around(ProceedingJoinPoint joinPoint, PreventDuplicateSubmit preventDuplicateSubmit) throws Throwable {
StringBuilder key = new StringBuilder();
// 获取class
String simpleName = joinPoint.getTarget().getClass().getSimpleName();
key.append(simpleName);
// 获取请求方法
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
String methodName = method.getName();
key.append(":").append(methodName);
// 获取请求参数
Object[] args = joinPoint.getArgs();
for (Object arg : args) {
key.append(":").append(arg.toString());
}
// 判断是否已经请求过
if (redisTemplate.hasKey(key.toString())) {
throw new RuntimeException("请勿重复提交");
}
// 标记请求已经处理过
redisTemplate.opsForValue().set(key.toString(), "1", preventDuplicateSubmit.expireSeconds(), TimeUnit.SECONDS);
return joinPoint.proceed();
}
}
b.使用注解：在Controller方法上使用@PreventDuplicateSubmit注解
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class AnnotationController {
@PostMapping("/submitByAnnotation")
@PreventDuplicateSubmit
public String handleFormByAnnotation(@RequestParam String param) {
// 处理表单数据
return "success";
}
}
6.4 [1]自定义注解+拦截器：若依
01.自定义防重复提交注解
a.代码
/**
* 自定义注解防止表单重复提交
*
* @author ruoyi
*
*/
@Inherited
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RepeatSubmit
{
/**
* 间隔时间(ms)，小于此时间视为重复提交
*/
public int interval() default 5000;
/**
* 提示消息
*/
public String message() default "不允许重复提交，请稍候再试";
}
b.说明
a.@Inherited
该元注解表示如果一个类使用了这个 RepeatSubmit 注解，那么它的子类也会自动继承这个注解
这在某些需要对一组相关的控制器方法进行统一重复提交检查的场景下很有用，子类无需再次显式添加该注解
b.@Target(ElementType.METHOD)
表明这个注解只能应用在方法上。在实际应用中，通常会将其添加到控制器类的处理请求的方法上
比如 Spring MVC 的 @RequestMapping 注解修饰的方法
c.@Retention(RetentionPolicy.RUNTIME)
意味着该注解在运行时仍然存在，可以通过反射机制获取到
这样在运行时，通过 AOP（面向切面编程）等技术拦截方法调用时
就能够读取到注解的属性值，从而实现重复提交的检查逻辑
d.@Documented
这个元注解用于将注解包含在 JavaDoc 中
当生成项目文档时，使用了该注解的方法会在文档中显示该注解及其属性
方便其他开发者了解该方法具有防止重复提交的功能以及相关的配置参数
c.代码
/**
* 间隔时间(ms)，小于此时间视为重复提交
*/
public int interval() default 5000;
d.说明
定义了一个名为 interval 的属性，类型为 int，表示两次提交之间允许的最小时间间隔，单位是毫秒
默认值为 5000，即 5 秒。如果两次提交的时间间隔小于这个值，就会被视为重复提交
e.代码
/**
* 提示消息
*/
public String message() default "不允许重复提交，请稍候再试";
f.说明
定义了一个名为 message 的属性，类型为 String，用于在检测到重复提交时返回给客户端的提示消息
默认消息为 “不允许重复提交，请稍候再试”。开发者可以根据具体业务需求，在使用注解时自定义这个提示消息
02.防止重复提交的抽象类
a.抽象类可以自己有具体方法
/**
* 防止重复提交拦截器
*
* @author ruoyi
*/
@Component
public abstract class RepeatSubmitInterceptor implements HandlerInterceptor
{
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception
{
if (handler instanceof HandlerMethod)
{
HandlerMethod handlerMethod = (HandlerMethod) handler;
Method method = handlerMethod.getMethod();
RepeatSubmit annotation = method.getAnnotation(RepeatSubmit.class);
if (annotation != null)
{
if (this.isRepeatSubmit(request, annotation))
{
AjaxResult ajaxResult = AjaxResult.error(annotation.message());
ServletUtils.renderString(response, JSON.toJSONString(ajaxResult));
return false;
}
}
return true;
}
else
{
return true;
}
}
/**
* 验证是否重复提交由子类实现具体的防重复提交的规则
*
* @param request 请求信息
* @param annotation 防重复注解参数
* @return 结果
* @throws Exception
*/
public abstract boolean isRepeatSubmit(HttpServletRequest request, RepeatSubmit annotation);
}
b.preHandle方法
a.说明
自定义抽象类拦截器 RepeatSubmitInterceptor 实现了 HandlerInterceptor 接口，重写 preHandle 方法
b.preHandle方法是负责拦截请求的
a.说明1
如果isRepeatSubmit方法返回true，表示当前请求是重复提交
此时会创建一个包含错误信息的AjaxResult对象，错误信息就是RepeatSubmit注解中设置的message
然后通过ServletUtils.renderString方法将AjaxResult对象转换为 JSON 字符串
并将其作为响应返回给客户端，同时返回false，阻止请求继续处理
b.说明2
如果方法上不存在RepeatSubmit注解，或者isRepeatSubmit方法返回false
表示当前请求不是重复提交，就返回true，允许请求继续执行后续的处理流程
c.代码
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception
{
if (handler instanceof HandlerMethod)
{
HandlerMethod handlerMethod = (HandlerMethod) handler;
Method method = handlerMethod.getMethod();
RepeatSubmit annotation = method.getAnnotation(RepeatSubmit.class);
if (annotation != null)
{
if (this.isRepeatSubmit(request, annotation))
{
AjaxResult ajaxResult = AjaxResult.error(annotation.message());
ServletUtils.renderString(response, JSON.toJSONString(ajaxResult));
return false;
}
}
return true;
}
else
{
return true;
}
}
d.参数说明
HttpServletRequest request：提供了关于当前 HTTP 请求的信息，如请求头、请求参数、请求方法等
HttpServletResponse response：用于设置 HTTP 响应，例如设置响应头、响应状态码、写入响应内容等
Object handler：代表即将被执行的处理器对象，在 Spring MVC 中，它通常是一个 HandlerMethod，但也可能是其他类型
e.方法解释
if (handler instanceof HandlerMethod)
{
HandlerMethod handlerMethod = (HandlerMethod) handler;
Method method = handlerMethod.getMethod();
RepeatSubmit annotation = method.getAnnotation(RepeatSubmit.class);
-------------------------------------------------------------------------------------------------
首先检查handler是否是HandlerMethod类型的，不是的话，直接放行，不做重复提交检查
因为该拦截器主要针对被@RepeatSubmit注解标记的方法进行处理
如果handler是HandlerMethod类型的话，将handler转换成为HandlerMethod并获取对应的Method对象
然后通过getMethod（）方法获取方法，并通过getAnnotation方法获取RepeatSubmit注解
-------------------------------------------------------------------------------------------------
if (annotation != null) {
if (this.isRepeatSubmit(request, annotation)) {
AjaxResult ajaxResult = AjaxResult.error(annotation.message());
ServletUtils.renderString(response, JSON.toJSONString(ajaxResult));
return false;
}
}
return true;
-------------------------------------------------------------------------------------------------
判断是否获取到RepeatSubmit注解，没有获取到，返回true，允许请求继续执行后续的处理流程
运用isRepeatSubmit方法，判断是否是重复提交
如果当前请求是重复提交将注解的错误信息封装给结果映射对象
并调用renderString 方法将字符串渲染到客户端
-------------------------------------------------------------------------------------------------
/**
* 将字符串渲染到客户端
*
* @param response 渲染对象
* @param string 待渲染的字符串
*/
public static void renderString(HttpServletResponse response, String string)
{
try
{
response.setStatus(200);
response.setContentType("application/json");
response.setCharacterEncoding("utf-8");
response.getWriter().print(string);
}
catch (IOException e)
{
e.printStackTrace();
}
}
c.isRepeatSubmit方法
a.判断是否重复提交，true重复提交，false不重复提交
/**
* 验证是否重复提交由子类实现具体的防重复提交的规则
*
* @param request 请求信息
* @param annotation 防重复注解参数
* @return 结果
* @throws Exception
*/
public abstract boolean isRepeatSubmit(HttpServletRequest request, RepeatSubmit annotation);
-------------------------------------------------------------------------------------------------
public final String REPEAT_PARAMS = "repeatParams";
public final String REPEAT_TIME = "repeatTime";
// 令牌自定义标识
@Value("${token.header}")
private String header; // token.header = "Authorization"
@Autowired
private RedisCache redisCache;
@SuppressWarnings("unchecked")
@Override
public boolean isRepeatSubmit(HttpServletRequest request, RepeatSubmit annotation)
{
String nowParams = "";
if (request instanceof RepeatedlyRequestWrapper)
{
RepeatedlyRequestWrapper repeatedlyRequest = (RepeatedlyRequestWrapper) request;
nowParams = HttpHelper.getBodyString(repeatedlyRequest);
}
// body参数为空，获取Parameter的数据
if (StringUtils.isEmpty(nowParams))
{
nowParams = JSON.toJSONString(request.getParameterMap());
}
Map<String, Object> nowDataMap = new HashMap<String, Object>();
nowDataMap.put(REPEAT_PARAMS, nowParams);
nowDataMap.put(REPEAT_TIME, System.currentTimeMillis());
// 请求地址（作为存放cache的key值）
String url = request.getRequestURI();
// 唯一值（没有消息头则使用请求地址）
String submitKey = StringUtils.trimToEmpty(request.getHeader(header));
// 唯一标识（指定key + url + 消息头）
String cacheRepeatKey = CacheConstants.REPEAT_SUBMIT_KEY + url + submitKey;
Object sessionObj = redisCache.getCacheObject(cacheRepeatKey);
if (sessionObj != null)
{
Map<String, Object> sessionMap = (Map<String, Object>) sessionObj;
if (sessionMap.containsKey(url))
{
Map<String, Object> preDataMap = (Map<String, Object>) sessionMap.get(url);
if (compareParams(nowDataMap, preDataMap) && compareTime(nowDataMap, preDataMap, annotation.interval()))
{
return true;
}
}
}
Map<String, Object> cacheMap = new HashMap<String, Object>();
cacheMap.put(url, nowDataMap);
redisCache.setCacheObject(cacheRepeatKey, cacheMap, annotation.interval(), TimeUnit.MILLISECONDS);
return false;
}
b.代码片段
@SuppressWarnings("unchecked")
-------------------------------------------------------------------------------------------------
注解 @SuppressWarnings("unchecked")： 这个注解用于抑制编译器的 “unchecked” 警告
在代码中，可能存在一些未经检查的类型转换操作，使用该注解可以告诉编译器忽略这些警告
c.代码片段
String nowParams = "";
-------------------------------------------------------------------------------------------------
初始化一个字符串变量 nowParams 用于存储当前请求的参数
d.代码片段
if (request instanceof RepeatedlyRequestWrapper) {
RepeatedlyRequestWrapper repeatedlyRequest = (RepeatedlyRequestWrapper) request;
nowParams = HttpHelper.getBodyString(repeatedlyRequest);
}
-------------------------------------------------------------------------------------------------
判断 当前请求是否是RepeatedlyRequestWrapper类型的
RepeatedlyRequestWrapper是自定义的允许多次请求的请求体（详情见备注）
如果是的话，强转对象，并且 通过getBodyString方法获取请求体的字符串内容，并且赋值给nowParams
e.代码片段
// body参数为空，获取Parameter的数据
if (StringUtils.isEmpty(nowParams))
{
nowParams = JSON.toJSONString(request.getParameterMap());
}
Map<String, Object> nowDataMap = new HashMap<String, Object>();
nowDataMap.put(REPEAT_PARAMS, nowParams); //REPEAT_PARAMS = "repeatParams"
nowDataMap.put(REPEAT_TIME, System.currentTimeMillis()); //REPEAT_TIME = "repeatTime"
-------------------------------------------------------------------------------------------------
if (StringUtils.isEmpty(nowParams))：如果通过上述方式获取的 nowParams 为空，说明请求体可能为空，此时通过 JSON.toJSONString(request.getParameterMap()) 将请求参数转换为 JSON 字符串，并赋值给 nowParams。这样无论请求参数是在请求体中还是在 URL 参数中，都能获取到
Map<String, Object> nowDataMap = new HashMap<String, Object>(); 创建一个新的 HashMap 用于存储当前请求的数据
nowDataMap.put(REPEAT_ PARAMS, nowParams); 将获取到的请求参数存入 nowDataMap 中，使用常量 REPEAT_PARAMS 作为键
nowDataMap.put(REPEAT_TIME, System.currentTimeMillis()); 将当前时间戳存入 nowDataMap 中，使用常量 REPEAT_TIME 作为键
f.代码片段
// 请求地址（作为存放cache的key值）
String url = request.getRequestURI();
// 唯一值（没有消息头则使用请求地址）
String submitKey = StringUtils.trimToEmpty(request.getHeader(header));
// 唯一标识（指定key + url + 消息头）
String cacheRepeatKey = CacheConstants.REPEAT_SUBMIT_KEY + url + submitKey;
-------------------------------------------------------------------------------------------------
String url = request.getRequestURI(); 获取当前请求的 URI
String submitKey = StringUtils.trimToEmpty(request.getHeader(header)); 从请求头中获取指定的键值（header 变量可能是在类中定义的一个常量，表示要获取的请求头字段），并去除两端的空白字符。如果请求头中不存在该字段，则返回空字符串
String cacheRepeatKey = CacheConstants . REPEAT_SUBMIT_KEY + url + submitKey; 使用一个常量 CacheConstants.REPEAT _SUBMIT_KEY 与请求 URI 和 submitKey 拼接生成一个唯一的缓存键 cacheRepeatKey。这个键用于在缓存中存储和检索与该请求相关的重复提交信息
g.代码片段
Object sessionObj = redisCache.getCacheObject(cacheRepeatKey);
if (sessionObj != null)
{
Map<String, Object> sessionMap = (Map<String, Object>) sessionObj;
if (sessionMap.containsKey(url))
{
Map<String, Object> preDataMap = (Map<String, Object>) sessionMap.get(url);
if (compareParams(nowDataMap, preDataMap) && compareTime(nowDataMap, preDataMap, annotation.interval()))
{
return true;
}
}
}
Map<String, Object> cacheMap = new HashMap<String, Object>();
cacheMap.put(url, nowDataMap);
redisCache.setCacheObject(cacheRepeatKey, cacheMap, annotation.interval(), TimeUnit.MILLISECONDS);
rerurn false;
-------------------------------------------------------------------------------------------------
通过缓存键 先去 redis 中，看是否存在相同的缓存信息，如果存在，说明之前有过类似的请求，进入判断
因为这里 redisCache.setCacheObject(cacheRepeatKey, cacheMap, annotation.interval(), TimeUnit.MILLISECONDS); 传了map，所以说 redisCache.getCacheObject(cacheRepeatKey); 得到的map，就是同样的类型的，所以键值就是 url 。
检查 sessionMap 这个 Map 中是否包含以当前请求的 url 作为键的记录。这一步是因为在缓存的数据结构中，url 被用作内层键来存储每个请求的具体数据。如果存在这个键，说明之前已经有针对该 url 的请求被缓存。
-------------------------------------------------------------------------------------------------
调用 compareParams 方法比较当前请求的数据 nowDataMap 和之前请求的数据 preDataMap 的参数是否相同
同时调用 compareTime 方法比较当前请求时间和之前请求时间的间隔是否小于 @RepeatSubmit 注解中
配置的 interval 时间。如果参数相同且时间间隔小于设定值，说明当前请求可能是重复提交，返回 true
-------------------------------------------------------------------------------------------------
如果缓存中不存在当前请求 url 的记录，或者当前请求不被判定为重复提交，则执行以下操作：
Map<String, Object> cacheMap = new HashMap<String, Object>();
创建一个新的 HashMap 用于存储当前请求的数据
cacheMap.put(url, nowDataMap);
将当前请求的 url 作为键，nowDataMap（包含当前请求参数和时间）作为值存入 cacheMap
redisCache.setCacheObject(cacheRepeatKey, cacheMap, annotation.interval(), TimeUnit.MILLISECONDS);
将 cacheMap 以 cacheRepeatKey 为键存入 Redis 缓存中
缓存时间为 @RepeatSubmit 注解中配置的 interval 时间，时间单位为毫秒
这样下次相同 url 的请求过来时，就可以从缓存中获取到之前的请求数据进行比较
d.compareParams方法
/**
* 判断参数是否相同
*/
private boolean compareParams(Map<String, Object> nowMap, Map<String, Object> preMap) {
String nowParams = (String) nowMap.get(REPEAT_PARAMS);
String preParams = (String) preMap.get(REPEAT_PARAMS);
return nowParams.equals(preParams);
}
e.compareTime方法
/**
* 判断两次间隔时间
*/
private boolean compareTime(Map<String, Object> nowMap, Map<String, Object> preMap, int interval) {
long time1 = (Long) nowMap.get(REPEAT_TIME);
long time2 = (Long) preMap.get(REPEAT_TIME);
if ((time1 - time2) < interval)
{
return true;
}
return false;
}
6.5 [2]redisson分布式锁：string结构
01.定义
分布式锁：在分布式环境中，通过Redisson实现锁机制，确保同一时间只有一个请求能够被处理，从而避免重复提交
02.原理
获取锁：在请求处理前尝试获取Redisson锁
处理请求：只有获取到锁的请求才能继续处理
释放锁：请求处理完成后释放锁
03.常用API
RLock：Redisson提供的锁对象
lock()：获取锁
unlock()：释放锁
04.使用步骤
获取Redisson客户端：初始化Redisson客户端
获取锁对象：通过Redisson客户端获取锁对象
尝试获取锁：调用lock()方法获取锁
处理请求：在获取锁后处理请求
释放锁：调用unlock()方法释放锁
05.代码示例
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
public class RedissonLockExample {
@Autowired
private RedissonClient redissonClient;
public void processRequest(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
try {
lock.lock();
// 处理请求逻辑
} finally {
lock.unlock();
}
}
}
06.代码示例
package com.ruoyi.redis4.demo03;
import java.util.concurrent.TimeUnit;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
// redisson分布式锁
public class Demo01 {
public static void main(String[] args) {
// 创建 Redisson 客户端
Config config = new Config();
config.useSingleServer()
.setAddress("redis://localhost:6379"); // 替换为你的 Redis 地址和端口
RedissonClient redisson = Redisson.create(config);
// 获取锁
RLock lock = redisson.getLock("myLock");
try {
// 尝试获取锁，最多等待 10 秒，锁定 30 秒
boolean isLockAcquired = lock.tryLock(10, 30, TimeUnit.SECONDS);
if (isLockAcquired) {
try {
// 业务逻辑
System.out.println("锁已获得，执行业务逻辑...");
Thread.sleep(10000); // 模拟业务处理
} finally {
// 释放锁
lock.unlock();
System.out.println("锁已释放");
}
} else {
System.out.println("未能获取锁，稍后重试...");
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 关闭 Redisson 客户端
redisson.shutdown();
}
}
}
07.代码示例
a.初始版本
public CreateOrderResponse createPaymentNo_初始版本(CreateOrderRequest request) {
//1.调用支付单创建方法
String paymentNo = createPaymentorder(request);
//2.组装返回对象
CreateorderResponse response = new CreateOrderResponse();
response.setCode("200");
response.setMessage("支付单创建成功")；
response.setPaymentNo(paymentNo);
return response;
}
b.进化版本
public CreateOrderResponse createPaymentNo_进化版本(CreateOrderRequest request) {
//1.调用支付单创建方法
String orderNo = request.getorderNo(）;
PaymentOrder paymentOrder = queryPaymentorderByorderNo(orderNo);
//2.如果支付单存在，则直接返回
CreateorderResponse response = new CreateOrderResponse();
if(Object.nonNull(paymentOrder)) {
response.setCode("200");
response.setMessage("付单已经存在，请勿重复创建")；
response.setPaymentNo(paymentOrder.getPaymentNo);
return response;
}
//3.调用支付单创建方法
String paymentNo = createPaymentorder(request);
//4.组装返回对象
response.setCode("200");
response.setMessage("支付单创建成功")；
response.setPaymentNo(paymentNo);
return response;
}
c.终极版本
public CreateOrderResponse createPaymentNo_终极版本(CreateOrderRequest request) {
//1.调用支付单创建方法
String orderNo = request.getorderNo(）;
PaymentOrder paymentOrder = queryPaymentorderByorderNo(orderNo);
//2.如果支付单存在，则直接返回
CreateorderResponse response = new CreateOrderResponse();
if(Object.nonNull(paymentOrder)) {
response.setCode("200");
response.setMessage("付单已经存在，请勿重复创建")；
response.setPaymentNo(paymentOrder.getPaymentNo);
return response;
}
//3.redis分布式锁，锁住这个订单
String redisKey = "OrderNo_" + orderNo;
boolean success = lock(redisKey, 60L);
if(!success) {
response.setCode("500");
response.setMessage（"系统繁忙");
response.setPaymentNo(nuLL);
return response;
}
String paymentNo = null;
try {
paymentNo = createPaymentorder(request);
} catch (Exception e) {
// 打印错误日志
} finally {
unLock(redisKey);
}
//4.组装返回对象
response.setCode("200");
response.setMessage("支付单创建成功")；
response.setPaymentNo(paymentNo);
return response;
}
6.6 [3]setnx+lua脚本：string结构
00.思路
生成唯一请求标识：每个请求都需要携带一个唯一的标识（如 requestId），用于区分不同的请求
使用 Redis 的 SETNX 命令：SETNX（Set if Not eXists）命令可以在键不存在时设置键值，返回 1 表示设置成功，返回 0 表示键已存在。这可以用来判断请求是否已经被处理过
Lua 脚本保证原子性：通过 Lua 脚本将 SETNX 和设置过期时间（TTL）操作封装在一起，确保它们的原子性，避免并发情况下的竞态条件
设置过期时间：为请求标识设置一个合理的过期时间，防止 Redis 中积累过多无效的键
01.原理
a.锁的获取
使用 SETNX 命令尝试设置一个键（锁的标识）
如果该键不存在，则设置成功并返回 1，表示获取到锁；如果键已存在，则返回 0，表示未能获取到锁
为了防止死锁，通常会设置一个过期时间（例如，px 参数）
b.锁的释放
释放锁的过程中，使用 Lua 脚本确保只有持有锁的线程可以释放它。脚本逻辑如下
首先获取锁的当前值
检查当前值是否与请求释放锁的值相同。如果相同，则调用 DEL 命令删除锁；如果不同，则返回释放失败
c.锁的过期机制
在获取锁时设置过期时间，可以避免因程序崩溃等情况导致的死锁
02.命令
a.使用
setnx命令，一般使用String结构
b.加锁命令
SETNX key value，当键不存在时，对键进行设置操作并返回成功，否则返回失败。KEY 是锁的唯一标识，一般按业务来决定命名
c.解锁命令
DEL key，通过删除键值对释放锁，以便其他线程可以通过 SETNX 命令来获取锁
d.锁超时
EXPIRE key timeout, 设置 key 的超时时间，以保证即使锁没有被显式释放，锁也可以在一定时间后自动释放，避免资源被永远锁住
03.setnx+Lua脚本
a.两个原因
a.setnx：存在死锁
使用 SETNX 单独设置锁会遇到【死锁】问题（客户端宕机时锁不会释放）
并且【无法保证原子性】（设置锁后设置超时时间的两步操作存在不一致的可能）
b.Lua脚本：lua原子性+事务
【Lua 脚本将 SETNX 和 EXPIRE 操作合并为一个原子操作】，避免了由于客户端故障导致的死锁问题，并确保了锁的超时自动释放
使用Lua脚本在释放锁时验证唯一标识，确保只有持有锁的客户端才能释放锁，避免误释放问题
b.单独setnx
a.无法自动释放锁
使用 SETNX 创建的锁没有过期时间，万一持锁的客户端在业务执行过程中崩溃，锁将永远不会释放，导致死锁问题
b.无法保证操作的原子性
在分布式环境下，通常需要为锁设置一个超时时间，以防止锁长时间占用。最直观的做法是先调用 SETNX 再调用 EXPIRE 设置过期时间，但这样会导致 操作不具备原子性
如果在调用 SETNX 成功之后，客户端还没来得及设置 EXPIRE 而宕机，那么这个锁将没有超时时间，导致死锁
因此，单独使用 SETNX 不能完全解决分布式锁的需求
c.setnx（存在死锁）+Lua脚本（lua原子性+事务）
a.确保原子性
通过 Lua 脚本，可以在一次请求中完成 SETNX 和 EXPIRE 操作，确保在 Redis 内部操作的原子性
即便客户端在锁定过程中出现故障，Redis 也会确保这个脚本执行的原子性
b.防止死锁
Lua 脚本设置了锁的过期时间，因此，即便客户端崩溃或超时，锁也会自动释放，避免死锁
c.简化逻辑
将锁的创建和过期时间设置合并到一个操作中，不需要客户端多次调用 Redis
04.代码示例
package com.ruoyi.redis4.demo03;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;
import java.util.Collections;
// setnx+Lua脚本
public class Demo02 {
private static final String LOCK_KEY = "myLock";
private static final long LOCK_EXPIRE_TIME = 30000; // 锁过期时间，单位：毫秒
private static final String LOCK_VALUE = String.valueOf(System.currentTimeMillis() + LOCK_EXPIRE_TIME + 1); // 锁的值
private static Jedis jedis = new Jedis("localhost", 6379); // 替换为你的 Redis 地址和端口
public static boolean acquireLock(String lockKey, String lockValue) {
String result = jedis.set(lockKey, lockValue, SetParams.setParams().nx().px(LOCK_EXPIRE_TIME));
return "OK".equals(result);
}
public static boolean releaseLock(String lockKey, String lockValue) {
// Lua 脚本：只有当值匹配时才删除锁
String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Long result = (Long) jedis.eval(luaScript, Collections.singletonList(lockKey), Collections.singletonList(lockValue));
return result != null && result == 1;
}
public static void main(String[] args) {
String lockValue = String.valueOf(System.currentTimeMillis() + LOCK_EXPIRE_TIME + 1); // 锁的值
// 尝试获取锁
if (acquireLock(LOCK_KEY, lockValue)) {
try {
// 业务逻辑
System.out.println("锁已获得，执行业务逻辑...");
Thread.sleep(10000); // 模拟业务处理
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 释放锁
if (releaseLock(LOCK_KEY, lockValue)) {
System.out.println("锁已释放");
} else {
System.out.println("锁释放失败");
}
}
} else {
System.out.println("未能获取锁，稍后重试...");
}
// 关闭 Jedis 客户端
jedis.close();
}
}
05.代码示例
package com.ruoyi.redis4.demo03;
import java.util.Collections;
import org.apache.commons.lang3.StringUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.SetParams;
// setnx+Lua脚本
public class Demo03 {
static final String _LOCKKEY = "REDISLOCK"; // 锁 key
static final String _FLAGID = "UUID:6379"; // 标识（UUID）
static final Integer _TimeOut = 90; // 最大超时时间
public static void main(String[] args) {
Jedis jedis = JedisUtils.getJedis();
// 加锁
boolean lockResult = lock(jedis, _LOCKKEY, _FLAGID, _TimeOut);
// 逻辑业务处理
if (lockResult) {
System.out.println("加锁成功");
} else {
System.out.println("加锁失败");
}
// 手动释放锁
if (unLock(jedis, _LOCKKEY, _FLAGID)) {
System.out.println("锁释放成功");
} else {
System.out.println("锁释放成功");
}
}
/**
* @param jedis Redis 客户端
* @param key 锁名称
* @param flagId 锁标识（锁值），用于标识锁的归属
* @param secondsTime 最大超时时间
* @return
*/
public static boolean lock(Jedis jedis, String key, String flagId, Integer secondsTime) {
SetParams params = new SetParams();
params.ex(secondsTime);
params.nx();
String res = jedis.set(key, flagId, params);
if (StringUtils.isNotBlank(res) && res.equals("OK"))
return true;
return false;
}
/**
* 释放分布式锁
* @param jedis Redis 客户端
* @param lockKey 锁的 key
* @param flagId 锁归属标识
* @return 是否释放成功
*/
public static boolean unLock(Jedis jedis, String lockKey, String flagId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(flagId));
if ("1L".equals(result)) { // 判断执行结果
return true;
}
return false;
}
}
06.代码示例
a.Lua脚本，用于原子性地执行 SETNX 并设置过期时间
-- set_if_not_exists.lua
-- KEYS[1] - request key
-- ARGV[1] - request value (可以是任何标识，如用户ID)
-- ARGV[2] - expiration time in seconds
if redis.call("SETNX", KEYS[1], ARGV[1]) == 1 then
redis.call("EXPIRE", KEYS[1], ARGV[2])
return 1
else
return 0
end
b.Java实现
a.引入依赖
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>4.3.1</version>
</dependency>
b.Lua脚本加载与执行
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
public class IdempotencyService {
private final JedisPool jedisPool;
private final String luaScript =
"if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then " +
" redis.call('EXPIRE', KEYS[1], ARGV[2]) " +
" return 1 " +
"else " +
" return 0 " +
"end";
public IdempotencyService(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
/**
* 尝试执行幂等性检查
* @param requestId 唯一请求标识
* @param requestValue 请求的值，可以是用户ID或其他标识
* @param expireTime 过期时间（秒）
* @return true 表示请求未被处理过，可以继续处理；false 表示请求已被处理，应该忽略或返回重复响应
*/
public boolean tryExecute(String requestId, String requestValue, int expireTime) {
try (Jedis jedis = jedisPool.getResource()) {
Object result = jedis.eval(luaScript,
1,
requestId,
requestValue,
String.valueOf(expireTime));
Long res = (Long) result;
return res == 1;
} catch (Exception e) {
// 处理异常，如日志记录
e.printStackTrace();
// 根据具体需求决定是否允许请求通过
return false;
}
}
}
c.使用示例
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
public class OrderController {
private final IdempotencyService idempotencyService;
public OrderController() {
// 配置 Jedis 连接池
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(128);
// 根据实际情况设置其他参数，如主机、端口等
JedisPool jedisPool = new JedisPool(poolConfig, "localhost", 6379);
this.idempotencyService = new IdempotencyService(jedisPool);
}
/**
* 提交订单接口
* @param requestId 唯一请求标识，客户端生成
* @param userId 用户ID
* @param orderDetails 订单详情
* @return 处理结果
*/
public String submitOrder(String requestId, String userId, String orderDetails) {
// 尝试执行幂等性检查
boolean canProceed = idempotencyService.tryExecute(requestId, userId, 60); // 过期时间60秒
if (!canProceed) {
return "Duplicate request. Order has already been submitted.";
}
// 处理订单逻辑
// 例如：创建订单、扣减库存、记录日志等
// ...
return "Order submitted successfully.";
}
public static void main(String[] args) {
OrderController controller = new OrderController();
// 模拟客户端发送请求
String requestId = "unique-request-id-123"; // 应确保每个请求唯一
String userId = "user-456";
String orderDetails = "item=book&quantity=2";
String response1 = controller.submitOrder(requestId, userId, orderDetails);
System.out.println(response1); // 应输出："Order submitted successfully."
String response2 = controller.submitOrder(requestId, userId, orderDetails);
System.out.println(response2); // 应输出："Duplicate request. Order has already been submitted."
}
}
d.生成唯一请求标识
为了确保接口的幂等性，每个请求都需要携带一个唯一的请求标识 (requestId)
客户端生成：客户端在发起请求时生成一个 UUID 并作为请求标识传递给服务器
服务器生成：服务器在接收到请求后生成一个唯一标识并返回给客户端，但这种方式在防止重复提交时可能不够有效，因为重复请求需要携带相同的标识
-------------------------------------------------------------------------------------------------
建议采用 客户端生成 的方式，并通过请求头或请求参数传递 requestId
import java.util.UUID;
public class Client {
public static void main(String[] args) {
String requestId = UUID.randomUUID().toString();
// 将 requestId 传递给服务器的接口
}
}
e.注意事项
唯一性保证：确保每个请求的 requestId 唯一，避免不同请求使用相同的标识
过期时间设置：根据业务需求合理设置键的过期时间，防止 Redis 中积累过多无效键。过期时间应覆盖请求可能的重复提交时间窗口
幂等性策略设计：不仅要在技术上实现幂等性，还需要在业务逻辑上设计合适的幂等性策略，如订单号的唯一性、事务的一致性等
异常处理：在实现过程中，需处理 Redis 连接失败或脚本执行错误等异常情况，确保系统的稳定性
分布式环境：在分布式系统中，确保所有实例都连接到同一个 Redis 实例或集群，确保幂等性检查的一致性
6.7 [4]zookeeper分布式锁：临时节点和序列号
01.原理
使用 ZooKeeper 提供的分布式锁功能，可以通过临时节点和序列号来实现锁机制
创建一个顺序节点，每次获得锁时，检查最小的序列号
02.ZooKeeper原生API
package com.ruoyi.redis4.demo03;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
// zookeeper分布式锁
public class Demo03 {
private static final String ZK_ADDRESS = "localhost:2181"; // ZooKeeper 地址
private static final String LOCK_PATH = "/myLock"; // 锁路径
private static final int SESSION_TIMEOUT = 3000; // 会话超时时间
private ZooKeeper zooKeeper;
private String currentLockPath; // 当前锁的路径
public Demo03() throws IOException {
this.zooKeeper = new ZooKeeper(ZK_ADDRESS, SESSION_TIMEOUT, event -> {
if (event.getType() == Watcher.Event.EventType.None) {
// 连接成功或失去连接
}
});
}
public boolean acquireLock() throws InterruptedException, KeeperException {
String lockNode = LOCK_PATH + "/lock-";
// 创建一个临时顺序节点
currentLockPath = zooKeeper.create(lockNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 获取所有锁节点并按顺序排列
List<String> lockNodes = zooKeeper.getChildren(LOCK_PATH, false);
Collections.sort(lockNodes);
// 获取当前锁节点的序号
String currentNodeName = currentLockPath.substring(currentLockPath.lastIndexOf("/") + 1);
// 检查是否为第一个节点
if (currentNodeName.equals(lockNodes.get(0))) {
return true; // 获取到锁
}
// 如果不是第一个节点，则需要监听前一个节点
int index = lockNodes.indexOf(currentNodeName);
String previousNode = lockNodes.get(index - 1);
// 设置监听器，等待前一个节点被删除
final CountDownLatch latch = new CountDownLatch(1);
zooKeeper.exists(LOCK_PATH + "/" + previousNode, event -> {
if (event.getType() == Watcher.Event.EventType.NodeDeleted) {
latch.countDown(); // 前一个节点被删除，继续获取锁
}
});
latch.await(); // 等待前一个节点被删除
return true; // 获取到锁
}
public void releaseLock() throws InterruptedException, KeeperException {
if (currentLockPath != null) {
zooKeeper.delete(currentLockPath, -1); // 删除当前锁节点
}
}
public static void main(String[] args) {
try {
Demo03 lock = new Demo03();
// 尝试获取锁
if (lock.acquireLock()) {
try {
System.out.println("成功获得锁，执行业务逻辑...");
Thread.sleep(10000); // 模拟业务处理
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.releaseLock(); // 释放锁
System.out.println("锁已释放");
}
} else {
System.out.println("未能获得锁，稍后重试...");
}
lock.zooKeeper.close(); // 关闭 ZooKeeper 客户端
} catch (IOException | InterruptedException | KeeperException e) {
e.printStackTrace();
}
}
}
03.Curator框架
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class ZkDistributedLock {
private static final String ZK_ADDRESS = "localhost:2181"; // ZooKeeper 地址
private static final String LOCK_PATH = "/myLock"; // 锁路径
public void doWork() {
CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_ADDRESS, new ExponentialBackoffRetry(1000, 3));
client.start();
InterProcessMutex lock = new InterProcessMutex(client, LOCK_PATH);
try {
lock.acquire(); // 获取锁
// 执行业务逻辑
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
lock.release(); // 释放锁
} catch (Exception e) {
e.printStackTrace();
}
}
client.close(); // 关闭客户端
}
}
6.8 [5]前端拦截：禁用、隐藏
01.定义
前端拦截：通过禁用按钮或隐藏按钮来防止用户重复提交请求
02.原理
按钮禁用：在用户提交请求后，立即禁用提交按钮，防止重复点击
按钮隐藏：在请求提交后隐藏按钮，避免用户再次提交
03.常用API
JavaScript的addEventListener：监听按钮点击事件
HTML的disabled属性：禁用按钮
04.使用步骤
监听按钮点击事件：使用JavaScript监听按钮的点击事件
禁用或隐藏按钮：在事件处理函数中禁用或隐藏按钮
恢复按钮状态：在请求完成后恢复按钮状态
05.代码示例
<button id="submitButton" onclick="handleSubmit()">提交</button>
<script>
function handleSubmit() {
var button = document.getElementById("submitButton");
button.disabled = true; // 禁用按钮
// 模拟请求
setTimeout(function() {
button.disabled = false; // 恢复按钮状态
}, 3000);
}
</script>
6.9 [6]后端拦截：检查重复请求
01.定义
后端拦截：在请求处理前检查是否已处理过相同的请求ID，防止重复提交
02.原理
请求ID检查：在请求处理前检查请求ID是否已存在
存储请求ID：处理请求后存储请求ID
03.常用API
HashMap：存储请求ID
synchronized：确保线程安全
04.使用步骤
检查请求ID：在请求处理前检查请求ID是否已存在
处理请求：如果请求ID不存在，则处理请求
存储请求ID：处理请求后存储请求ID
05.代码示例
import java.util.HashMap;
import java.util.Map;
public class RequestHandler {
private final Map<String, Boolean> requestMap = new HashMap<>();
public synchronized boolean handleRequest(String requestId) {
if (requestMap.containsKey(requestId)) {
return false; // 重复请求
}
requestMap.put(requestId, true);
// 处理请求逻辑
return true;
}
}
6.10 [7]token机制：验证重复性
01.思路
可以通过token的机制避免重复提交，当用户访问页面的时候，请求后端服务拿到一个token
然后下一次接口点击的时候把token带过来，服务端对token进行验证，验证该token是否被使用过
如果没有被使用过才可以进行点击。验证的逻辑可以放在数据库中，通过数据库的悲观锁或者乐观锁都可以实现
6.11 [8]布隆过滤器：验证重复性
01.说明
快速判断某个元素是否存在于集合中，可以在服务器端使用布隆过滤器记录某个操作是否已经被执行过，从而防止重复执行。
02.思路
如果布隆过滤器不存在，则一定不存在，所以，如果没查到，说明一定没有幂等操作，直接执行就行了
如果查询布隆过滤器发现有命中，则需要在服务数据库做一次幂等判断
大多数情况下，需要幂等的情况占比小，所以可以用布隆过滤器做一次fail-fast的快速校验