单机环境下Caffeine和Redis两级缓存的实现与问题解决

news/2024/12/2 1:15:05/

1. Caffeine和Redis介绍

Caffeine 是一个高性能、基于 Java 的缓存库。它被设计用来作为 Java 应用程序的本地缓存解决方案,提供了比传统缓存更高效的功能和性能。Caffeine 可以缓存任意类型的数据,并具有丰富的配置选项,以满足不同应用的缓存需求。

Caffeine 通过使用多种缓存策略(如基于大小、时间、引用等),支持自动过期、最大容量限制、以及基于异步加载缓存操作。它的设计目标是高效、低延迟,并能在多线程环境中保持性能。

Redis 是一个开源的、基于内存的数据结构存储系统,支持多种数据结构,如字符串(string)、哈希(hash)、列表(list)、集合(set)、有序集合(sorted set)等。它不仅可以用作数据库,还可以作为缓存和消息中间件。由于其速度非常快(大多数操作在内存中进行),Redis 被广泛应用于需要快速读取和写入的场景,例如缓存、会话管理、实时数据分析、消息队列等。

在本文章中,我们就详细介绍一下,如何使用Caffeine和Redis搭建一个本地+远程的二级缓存结构并且让它们的交互变得更加丝滑。

2. 搭建过程

2.1 SpringBoot 项目引入依赖

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>com.github.ben-manes.caffeine</groupId><artifactId>caffeine</artifactId><version>3.1.8</version></dependency>

2.2 注册Caffeine中cache对象为Bean对象

@Bean
public Cache<String, Object> localCache() {return Caffeine.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).initialCapacity(100).maximumSize(1000).build();}

后续使用只需要注入这个类型为Cache的bean对象就可以调用其中的增删查改方法.

基本使用: 

cache.put("key1","value1");//存入key-value键值对
cache.getIfPresent("key1");//根据key删除键值对
cache.invalidate("key1");//根据key删除键值对

2.3 注入操作Redis客户端的对象stringRedisTemplate

@Autowired
private StringRedisTemplate stringRedisTemplate;

基本使用:

stringRedisTemplate.opsForValue().set("key1", "value1");//存入key-value键值对
stringRedisTemplate.opsForValue().get("key1");//根据key获取键值对
stringRedisTemplate.opsForValue().del("key1");//根据key删除键值对

2.4 缓存工具类CacheUtil结合使用

我们给出get方法作为示例:

@Configuration
public class CacheUtil {@Autowiredprivate Cache cache;@Autowiredprivate StringRedisTemplate stringRedisTemplate;public boolean set(String key, String value) {String localCache = cache.getIfPresent("key1";//Caffeine本地缓存获取key-value键值对String remoteCache = stringRedisTemplate.opsForValue().get("key1");//Redis远程缓存获取key-value键值对return true;} catch (Exception e) {log.error("CacheUtil error, set({}, {})", key, value, e, tryConnectTime.isDegrade());return false;}}
}

3. 进阶问题

3.1 如何让两者产生关联?

我们上述的例子实际上仅仅是把这两个缓存的get相关代码"放到了一起",实际上并没有让这两者产生任何关联性.思考一下,在Redis中查询不到数据的时候,我们是不是会查询数据库,再把对应数据放入Redis中.那么,我们实际上可以使用"桥接模式",仍然让Caffeine中的缓存数据从Redis中获取.

注册bean对象----本地cache和redis客户端
/*** 配置Caffeine本地缓存* @return*/@Beanpublic Cache<String, Object> localCache() {return Caffeine.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).initialCapacity(100).maximumSize(1000).build();}/*** 获取redis本地客户端* @param lettuceConnectionFactory* @return*/@Beanpublic RedisClient redisClient(@Autowired LettuceConnectionFactory lettuceConnectionFactory) {return (RedisClient) lettuceConnectionFactory.getNativeClient();}

其中redis客户端的获取使用了"工厂模式",直接使用注入的工厂即可.

注册缓存上下文(cacheFrontendContext)为bean对象----结合本地cache和redis客户端对象
/*** 缓存上下文* @param redisClient* @param cache* @return*/@Beanpublic CacheFrontendContext cacheFrontendContext(@Autowired RedisClient redisClient, @Autowired Cache cache) {return new CacheFrontendContext(redisClient, cache);}

这个对象可以用来帮助我们设置Caffeine跟踪Redis中的数据变化.

public class CacheFrontendContext {private static final Logger log = LoggerFactory.getLogger(CacheFrontendContext.class);@Getterprivate CacheFrontend cacheFrontend;private RedisClient redisClient;@Getterprivate Cache cache;StatefulRedisConnection<String, String> connection;public CacheFrontendContext(RedisClient redisClient, Cache cache) {this.redisClient = redisClient;this.cache = cache;}
}
核心追踪方法
connection.addListener(message -> {List<Object> content = message.getContent(StringCodec.UTF8::decodeKey);log.info("type:{}, content:{}",message.getType(), content);if (message.getType().equalsIgnoreCase("invalidate")) {List<String> keys = (List<String>) content.get(1);keys.forEach(key -> {cache.invalidate(key);});}
});

当Redis中数据变动后,当我们尝试从Caffeine中根据某个key值获取对应value时,会监听消息的类型,如果类型为"invalidate"(已经变动),就自动清除Caffeine中这对key-value缓存,再重新将Redis中的数据放入Caffeine.

Caffeine-Redis桥接关键
connection = redisClient.connect();CacheFrontend<String, String> frontend = ClientSideCaching.enable(new CaffeineCacheAccessor(cache),connection,TrackingArgs.Builder.enabled());
this.cacheFrontend = frontend;

通过这个代码片段设置对Redis消息的追踪.

CaffeineCacheAccessor----Caffeine-Redis桥接类

这个类的是十分必要的,Caffeine和Redis本身是毫不相干的两个组件,要将它们结合在一起,除了追踪以外,仍然需要告诉Caffeine:获取到Redis中的最新数据后,应该怎么处理这些数据,再存放到Caffeine中.

public class CaffeineCacheAccessor implements CacheAccessor {private static final Logger log = LoggerFactory.getLogger(CaffeineCacheAccessor.class);private Cache cache;public CaffeineCacheAccessor(Cache cache) {this.cache = cache;}@Overridepublic Object get(Object key) {log.info("caffeine get => {}", key);return cache.getIfPresent(key);}@Overridepublic void put(Object key, Object value) {log.info("caffeine set => {}:{}", key, value);cache.put(key, value);}@Overridepublic void evict(Object key) {log.info("caffeine evict => {}", key);cache.invalidate(key);}
}

通过这样一系列的设置,我们就能够把get部分的代码简化:

@Autowired
private CacheFrontendContext cacheFrontendContext;public String get(String key) {return cacheFrontendContext.getCacheFrontend().get(key).toString();
}

在存数据时只需要将数据存入Redis中即可,在读取时优先读取Caffeine中的数据,如果不存在或者过期了,Caffeine会自动从Redis中读取数据,成功让Caffeine和Redis产生了依赖关系.

3.2 Redis挂了怎么办?(熔断降级与断开重连的实现)

经过上述的操作,Caffeine的确和Redis产生了依赖关系,但是如果Redis挂了怎么办?我们就不能再向Redis中存数据了.那么我们就需要实现熔断降级,就是解开这些组件的耦合关系.在这个案例中,就是实现Caffeine本地缓存的存取数据,不至于影响到整个系统.那么我们实际上可以构造一个check方法轮询来确保Redis的连接状态,如果连接断开了我们就尝试重新连接,如果多次重新连接失败,就利用Caffeine来存取数据.

check方法
@SneakyThrowspublic void check() {if (connection != null) {if (connection.isOpen()) {  return;}}try {tryConnectTime.increase();//成功降级就减少重连的频率if (tryConnectTime.isDegrade()) Thread.sleep(5000);//重新建立连接connection = redisClient.connect();CacheFrontend<String, String> frontend = ClientSideCaching.enable(new CaffeineCacheAccessor(cache),connection,TrackingArgs.Builder.enabled());this.cacheFrontend = frontend;//添加监听,如果redis数据变动,caffeine获取时自动清除缓存connection.addListener(message -> {List<Object> content = message.getContent(StringCodec.UTF8::decodeKey);log.info("type:{}, content:{}",message.getType(), content);if (message.getType().equalsIgnoreCase("invalidate")) {List<String> keys = (List<String>) content.get(1);keys.forEach(key -> {cache.invalidate(key);});}});log.warn("The redis client side connection had been reconnected.");}catch (RuntimeException e) {log.error("The redis client side connection had been disconnected, waiting reconnect...", e);}}
TryConnectTime----用于计重连次数的类
public class TryConnectTime {private volatile AtomicInteger time = new AtomicInteger(0);public void increase() {time.incrementAndGet();}public void reset() {time.set(0);}public boolean isDegrade() {return time.get() > 5;//五次尝试重连失败,就熔断降级}
}
@Beanpublic TryConnectTime tryConnectTime() {return new TryConnectTime();
}

CacheUtil工具类set方法示例:

public boolean set(String key, String value) {try {if (tryConnectTime.isDegrade()) {return setByCaffeine(key, value);}return setByRedis(key, value);} catch (Exception e) {log.error("CacheUtil error, set({}, {}), isDegrade:{}", key, value, e, tryConnectTime.isDegrade());return false;}
}

3.3 Redis重连后的数据一致性问题

那么我们之前说Caffeine是依赖于Redis中的数据的,如果Redis重启后,在这段Redis挂掉期间的缓存数据是存放在Caffeine中的,当Redis服务又可用时会清除它自己的所有缓存数据,会不会把Caffeine中实际有用的数据当作过期的数据,从而进行覆盖呢?实际上是有可能的.

解决方法也十分简单,我们只需要记录下这段期间(熔断降级后到Redis服务可用)内对Caffeine缓存数据的变动,另外设置一个Caffeine的bean对象,把这些数据在Redis重新成功连接时,再设置回到Redis中.(因为有两个Cache类型的bean对象,需要使用@Qualifier根据名称注入,@Autowired默认是根据类型注入)

@Bean
public Cache<String, Object> waitingSyncDataCache() {return Caffeine.newBuilder().expireAfterWrite(120, TimeUnit.MINUTES).initialCapacity(100).maximumSize(3000).build();
}

check完整方法

public class CacheFrontendContext {private static final Logger log = LoggerFactory.getLogger(CacheFrontendContext.class);@Getterprivate CacheFrontend cacheFrontend;private RedisClient redisClient;@Getterprivate Cache cache;@Qualifier("waitingSyncDataCache")private Cache waitingSyncDataCache;@Autowiredprivate TryConnectTime tryConnectTime;@Autowiredprivate StringRedisTemplate stringRedisTemplate;StatefulRedisConnection<String, String> connection;public CacheFrontendContext(RedisClient redisClient, Cache cache) {this.redisClient = redisClient;this.cache = cache;}@SneakyThrowspublic void check() {if (connection != null) {if (connection.isOpen()) {if (!waitingSyncDataCache.asMap().entrySet().isEmpty()) {syncDataToRedis(waitingSyncDataCache);}tryConnectTime.reset();return;}}try {tryConnectTime.increase();if (tryConnectTime.isDegrade()) Thread.sleep(5000);//重新建立连接connection = redisClient.connect();CacheFrontend<String, String> frontend = ClientSideCaching.enable(new CaffeineCacheAccessor(cache),connection,TrackingArgs.Builder.enabled());this.cacheFrontend = frontend;if (!waitingSyncDataCache.asMap().entrySet().isEmpty()) {syncDataToRedis(waitingSyncDataCache);}//添加监听,如果redis数据变动,caffeine自动清除缓存connection.addListener(message -> {List<Object> content = message.getContent(StringCodec.UTF8::decodeKey);log.info("type:{}, content:{}",message.getType(), content);if (message.getType().equalsIgnoreCase("invalidate")) {List<String> keys = (List<String>) content.get(1);keys.forEach(key -> {cache.invalidate(key);});}});log.warn("The redis client side connection had been reconnected.");}catch (RuntimeException e) {log.error("The redis client side connection had been disconnected, waiting reconnect...", e);}}private void syncDataToRedis(Cache waitingSyncDataCache) {Set<Map.Entry<String, String>> entrySet = waitingSyncDataCache.asMap().entrySet();for (Map.Entry<String, String> entry : entrySet) {if (!stringRedisTemplate.hasKey(entry.getKey())) {stringRedisTemplate.opsForValue().set(entry.getKey(), entry.getValue());log.info("同步key:{},value:{}到Redis客户端",entry.getKey(), entry.getValue());}waitingSyncDataCache.invalidate(entry.getKey());}}
}


http://www.ppmy.cn/news/1551626.html

相关文章

Mybatis:数据CRUD操作之环境准备以及相关知识学习文章清单

先讲Mybatis的CRUD操作所需的环境搭建起来&#xff1a; 环境准备完毕后&#xff0c;可以进行CRUD数据操作啦~ 请见以下几篇&#xff1a; Mybatis&#xff1a;CRUD数据操作之全量查询 Mybatis&#xff1a;CRUD数据操作之单条件查询 Mybatis&#xff1a;CRUD数据操作之多条件查…

在Java中使用Apache POI导入导出Excel(四)

本文将继续介绍POI的使用&#xff0c;上接在Java中使用Apache POI导入导出Excel&#xff08;三&#xff09; 使用Apache POI组件操作Excel&#xff08;四&#xff09; 31、外边框 外边框非常适合对信息部分进行分组&#xff0c;并且可以轻松添加到列和行中 使用 POI API。方…

良好的并发编程习惯之封闭(Confinement)

创作内容丰富的干货文章很费心力&#xff0c;感谢点过此文章的读者&#xff0c;点一个关注鼓励一下作者&#xff0c;激励他分享更多的精彩好文&#xff0c;谢谢大家&#xff01; “共享可变状态”有两个要点&#xff1a;“共享”和“可变”。封闭的策略是&#xff1a;不共享就完…

【如何提升代码工程质量】code review篇

应该对于基本上所有软件相关的公司来说&#xff0c;都有committer机制&#xff0c;即代码写好之后会提交合并请求&#xff0c;待相关人员code review通过后再进行合入&#xff0c;所以code review就是代码合入代码仓库的最后一道关卡&#xff0c;对于代码质量的影响也是不容忽视…

路由引入中次优路由和路由环路问题

A公司用的是IS-IS&#xff0c;B公司用的是OSPF&#xff0c;现在这两个公司要合并&#xff0c;网络要相通 项目目标 前期准备 配置IP地址&#xff1a;完成IP地址规划&#xff0c;A公司和B公司内部网络通过路由器R2和R4环回接口模拟。配置路由器接口的IP地址并测试所有直连链路的…

深度学习:利用GPU进行训练

深度学习&#xff1a;利用GPU进行训练 在现代深度学习框架中&#xff0c;如PyTorch&#xff0c;利用GPU加速模型训练是一种常见的做法。GPU&#xff08;图形处理单元&#xff09;由于其并行处理能力&#xff0c;特别适合执行大量的矩阵运算&#xff0c;这在训练神经网络时尤为…

python爬虫案例——猫眼电影数据抓取之字体解密,多套字体文件解密方法(20)

文章目录 1、任务目标2、网站分析3、代码编写1、任务目标 目标网站:猫眼电影(https://www.maoyan.com/films?showType=2) 要求:抓取该网站下,所有即将上映电影的预约人数,保证能够获取到实时更新的内容;如下: 2、网站分析 进入目标网站,打开开发者模式,经过分析,我…

C#里怎么样使用Array.BinarySearch函数?

C#里怎么样使用Array.BinarySearch函数? 因为二分算法如此重要,所以要多加练习。 但是它的返回值,也有三种状态,导致很多人使用它的时候, 也感觉到迷惑的。 在这里的例子演示了三种返回值的使用: /** C# Program to Search an element with Array Indices*/ using …