基于Spring接口集成Caffeine+Redis兩級緩存

第基于Spring接口集成Caffeine+Redis兩級緩存目錄前言改造JSR107規(guī)范CacheCacheManager配置使用分布式環(huán)境改造定義消息體Redis消息配置消息消費(fèi)邏輯修改DoubleCache測試總結(jié)

前言

在上一篇文章Redis+Caffeine兩級緩存的實(shí)現(xiàn)中，我們介紹了3種整合Caffeine和Redis作為兩級緩存使用的方法，雖然說能夠?qū)崿F(xiàn)功能，但實(shí)現(xiàn)手法還是太粗糙了，并且遺留了一些問題沒有處理。本文將在上一篇的基礎(chǔ)上，圍繞兩個方面進(jìn)行進(jìn)一步的改造：

JSR107定義了緩存使用規(guī)范，spring中提供了基于這個規(guī)范的接口，所以我們可以直接使用spring中的接口進(jìn)行Caffeine和Redis兩級緩存的整合改造在分布式環(huán)境下，如果一臺主機(jī)的本地緩存進(jìn)行修改，需要通知其他主機(jī)修改本地緩存，解決分布式環(huán)境下本地緩存一致性問題

好了，在明確了需要的改進(jìn)問題后，下面我們開始正式修改。

改造

在上篇文章的v3版本中，我們使用自定義注解的方式實(shí)現(xiàn)了兩級緩存通過一個注解管理的功能。本文我們換一種方式，直接通過擴(kuò)展spring提供的接口來實(shí)現(xiàn)這個功能，在進(jìn)行整合之前，我們需要簡單了解一下JSR107緩存規(guī)范。

JSR107規(guī)范

在JSR107緩存規(guī)范中定義了5個核心接口，分別是CachingProvider,CacheManager,Cache,Entry和Expiry，參考下面這張圖，可以看到除了Entry和Expiry以外，從上到下都是一對多的包含關(guān)系。

從上面這張圖我們可以看出，一個應(yīng)用可以創(chuàng)建并管理多個CachingProvider，同樣一個CachingProvider也可以管理多個CacheManager，緩存管理器CacheManager中則維護(hù)了多個Cache。

Cache是一個類似Map的數(shù)據(jù)結(jié)構(gòu)，Entry就是其中存儲的每一個key-value數(shù)據(jù)對，并且每個Entry都有一個過期時間Expiry。而我們在使用spring集成第三方的緩存時，只需要實(shí)現(xiàn)Cache和CacheManager這兩個接口就可以了，下面分別具體來看一下。

Cache

spring中的Cache接口規(guī)范了緩存組件的定義，包含了緩存的各種操作，實(shí)現(xiàn)具體緩存操作的管理。例如我們熟悉的RedisCache、EhCacheCache等，都實(shí)現(xiàn)了這個接口。

在Cache接口中，定義了get、put、evict、clear等方法，分別對應(yīng)緩存的存入、取出、刪除、清空操作。不過我們這里不直接使用Cache接口，上面這張圖中的AbstractValueAdaptingCache是一個抽象類，它已經(jīng)實(shí)現(xiàn)了Cache接口，是spring在Cache接口的基礎(chǔ)上幫助我們進(jìn)行了一層封裝，所以我們直接繼承這個類就可以。

繼承AbstractValueAdaptingCache抽象類后，除了創(chuàng)建Cache的構(gòu)造方法外，還需要實(shí)現(xiàn)下面的幾個方法：

//在緩存中實(shí)際執(zhí)行查找的操作，父類的get()方法會調(diào)用這個方法

protectedabstractObjectlookup(Objectkey);

//通過key獲取緩存值，如果沒有找到，會調(diào)用valueLoader的call()方法

publicTTget(Objectkey,CallableTvalueLoader);

//將數(shù)據(jù)放入緩存中

publicvoidput(Objectkey,Objectvalue);

//刪除緩存

publicvoidevict(Objectkey);

//清空緩存中所有數(shù)據(jù)

publicvoidclear();

//獲取緩存名稱，一般在CacheManager創(chuàng)建時指定

StringgetName();

//獲取實(shí)際使用的緩存

ObjectgetNativeCache();

因?yàn)橐蟁edisTemplate和Caffeine的Cache，所以這些都需要在緩存的構(gòu)造方法中傳入，除此之外構(gòu)造方法中還需要再傳出緩存名稱cacheName，以及在配置文件中實(shí)際配置的一些緩存參數(shù)。先看一下構(gòu)造方法的實(shí)現(xiàn)：

publicclassDoubleCacheextendsAbstractValueAdaptingCache{

privateStringcacheName;

privateRedisTemplateObject,ObjectredisTemplate;

privateCacheObject,ObjectcaffeineCache;

privateDoubleCacheConfigdoubleCacheConfig;

protectedDoubleCache(booleanallowNullValues){

super(allowNullValues);

publicDoubleCache(StringcacheName,RedisTemplateObject,ObjectredisTemplate,

CacheObject,ObjectcaffeineCache,

DoubleCacheConfigdoubleCacheConfig){

super(doubleCacheConfig.getAllowNull());

this.cacheName=cacheName;

this.redisTemplate=redisTemplate;

this.caffeineCache=caffeineCache;

this.doubleCacheConfig=doubleCacheConfig;

//...

}

抽象父類的構(gòu)造方法中只有一個boolean類型的參數(shù)allowNullValues，表示是否允許緩存對象為null。除此之外，AbstractValueAdaptingCache中還定義了兩個包裝方法來配合這個參數(shù)進(jìn)行使用，分別是toStoreValue和fromStoreValue，特殊用途是用于在緩存null對象時進(jìn)行包裝、以及在獲取時進(jìn)行解析并返回。

我們之后會在CacheManager中調(diào)用后面這個自己實(shí)現(xiàn)的構(gòu)造方法，來實(shí)例化Cache對象，參數(shù)中DoubleCacheConfig是使用@ConfigurationProperties讀取的yml配置文件封裝的數(shù)據(jù)對象，會在后面使用。

當(dāng)一個方法添加了@Cacheable注解時，執(zhí)行時會先調(diào)用父類AbstractValueAdaptingCache中的get(key)方法，它會再調(diào)用我們自己實(shí)現(xiàn)的lookup方法。在實(shí)際執(zhí)行查找操作的lookup方法中，我們的邏輯仍然是先查找Caffeine、沒有找到時再查找Redis：

@Override

protectedObjectlookup(Objectkey){

//先從caffeine中查找

Objectobj=caffeineCache.getIfPresent(key);

if(Objects.nonNull(obj)){

("getdatafromcaffeine");

returnobj;

//再從redis中查找

StringredisKey=+":"+key;

obj=redisTemplate.opsForValue().get(redisKey);

if(Objects.nonNull(obj)){

("getdatafromredis");

caffeineCache.put(key,obj);

returnobj;

}

如果lookup方法的返回結(jié)果不為null，那么就會直接返回結(jié)果給調(diào)用方。如果返回為null時，就會執(zhí)行原方法，執(zhí)行完成后調(diào)用put方法，將數(shù)據(jù)放入緩存中。接下來我們實(shí)現(xiàn)put方法：

@Override

publicvoidput(Objectkey,Objectvalue){

if(!isAllowNullValues()Objects.isNull(value)){

log.error("thevalueNULLwillnotbecached");

return;

//使用toStoreValue(value)包裝，解決caffeine不能存null的問題

caffeineCache.put(key,toStoreValue(value));

//null對象只存在caffeine中一份就夠了，不用存redis了

if(Objects.isNull(value))

return;

StringredisKey=this.cacheName+":"+key;

OptionalLongexpireOpt=Optional.ofNullable(doubleCacheConfig)

.map(DoubleCacheConfig::getRedisExpire);

if(expireOpt.isPresent()){

redisTemplate.opsForValue().set(redisKey,toStoreValue(value),

expireOpt.get(),TimeUnit.SECONDS);

}else{

redisTemplate.opsForValue().set(redisKey,toStoreValue(value));

}

上面我們對于是否允許緩存空對象進(jìn)行了判斷，能夠緩存空對象的好處之一就是可以避免緩存穿透。需要注意的是，Caffeine中是不能直接緩存null的，因此可以使用父類提供的toStoreValue()方法，將它包裝成一個NullValue類型。在取出對象時，如果是NullValue，也不用我們自己再去調(diào)用fromStoreValue()將這個包裝類型還原，父類的get方法中已經(jīng)幫我們做好了。

另外，上面在put方法中緩存空對象時，只在Caffeine緩存中一份即可，可以不用在Redis中再存一份。

緩存的刪除方法evict()和清空方法clear()的實(shí)現(xiàn)就比較簡單了，直接刪除一跳或全部數(shù)據(jù)即可：

@Override

publicvoidevict(Objectkey){

redisTemplate.delete(this.cacheName+":"+key);

caffeineCache.invalidate(key);

@Override

publicvoidclear(){

SetObjectkeys=redisTemplate.keys(this.cacheName.concat(":*"));

for(Objectkey:keys){

redisTemplate.delete(String.valueOf(key));

caffeineCache.invalidateAll();

}

獲取緩存cacheName和實(shí)際緩存的方法實(shí)現(xiàn)：

@Override

publicStringgetName(){

returnthis.cacheName;

@Override

publicObjectgetNativeCache(){

returnthis;

}

最后，我們再來看一下帶有兩個參數(shù)的get方法，為什么把這個方法放到最后來說呢，因?yàn)槿绻覀冎皇鞘褂米⒔鈦砉芾砭彺娴脑?，那么這個方法不會被調(diào)用到，簡單看一下實(shí)現(xiàn)：

@Override

publicTTget(Objectkey,CallableTvalueLoader){

ReentrantLocklock=newReentrantLock();

try{

lock.lock();//加鎖

Objectobj=lookup(key);

if(Objects.nonNull(obj)){

return(T)obj;

//沒有找到

obj=valueLoader.call();

put(key,obj);//放入緩存

return(T)obj;

}catch(Exceptione){

log.error(e.getMessage());

}finally{

lock.unlock();

returnnull;

}

方法的實(shí)現(xiàn)比較容易理解，還是先調(diào)用lookup方法尋找是否已經(jīng)緩存了對象，如果沒有找到那么就調(diào)用Callable中的call方法進(jìn)行獲取，并在獲取完成后存入到緩存中去。至于這個方法如何使用，具體代碼我們放在后面使用這一塊再看。

需要注意的是，這個方法的接口注釋中強(qiáng)調(diào)了需要我們自己來保證方法同步，因此這里使用了ReentrantLock進(jìn)行了加鎖操作。到這里，Cache的實(shí)現(xiàn)就完成了，下面我們接著看另一個重要的接口CacheManager。

CacheManager

從名字就可以看出，CacheManager是一個緩存管理器，它可以被用來管理一組Cache。在上一篇文章的v2版本中，我們使用的CaffeineCacheManager就實(shí)現(xiàn)了這個接口，除此之外還有RedisCacheManager、EhCacheCacheManager等也都是通過這個接口實(shí)現(xiàn)。

下面我們要自定義一個類實(shí)現(xiàn)CacheManager接口，管理上面實(shí)現(xiàn)的DoubleCache作為spring中的緩存使用。接口中需要實(shí)現(xiàn)的方法只有下面兩個：

//根據(jù)cacheName獲取Cache實(shí)例，不存在時進(jìn)行創(chuàng)建

CachegetCache(Stringname);

//返回管理的所有cacheName

CollectionStringgetCacheNames();

在自定義的緩存管理器中，我們要使用ConcurrentHashMap維護(hù)一組不同的Cache，再定義一個構(gòu)造方法，在參數(shù)中傳入已經(jīng)在spring中配置好的RedisTemplate，以及相關(guān)的緩存配置參數(shù)：

publicclassDoubleCacheManagerimplementsCacheManager{

MapString,CachecacheMap=newConcurrentHashMap();

privateRedisTemplateObject,ObjectredisTemplate;

privateDoubleCacheConfigdcConfig;

publicDoubleCacheManager(RedisTemplateObject,ObjectredisTemplate,

DoubleCacheConfigdoubleCacheConfig){

this.redisTemplate=redisTemplate;

this.dcConfig=doubleCacheConfig;

//...

}

然后實(shí)現(xiàn)getCache方法，邏輯很簡單，先根據(jù)name從Map中查找對應(yīng)的Cache，如果找到則直接返回，這個參數(shù)name就是上一篇文章中提到的cacheName，CacheManager根據(jù)它實(shí)現(xiàn)不同Cache的隔離。

如果沒有根據(jù)名稱找到緩存的話，那么新建一個DoubleCache對象，并放入Map中。這里使用的ConcurrentHashMap的putIfAbsent()方法放入，避免重復(fù)創(chuàng)建Cache以及造成Cache內(nèi)數(shù)據(jù)的丟失。具體代碼如下：

@Override

publicCachegetCache(Stringname){

Cachecache=cacheMap.get(name);

if(Objects.nonNull(cache)){

returncache;

cache=newDoubleCache(name,redisTemplate,createCaffeineCache(),dcConfig);

CacheoldCache=cacheMap.putIfAbsent(name,cache);

returnoldCache==nullcache:oldCache;

}

在上面創(chuàng)建DoubleCache對象的過程中，需要先創(chuàng)建一個Caffeine的Cache對象作為參數(shù)傳入，這一過程主要是根據(jù)實(shí)際項(xiàng)目的配置文件中的具體參數(shù)進(jìn)行初始化，代碼如下：

privatecom.github.benmanes.caffeine.cache.CachecreateCaffeineCache(){

CaffeineObject,ObjectcaffeineBuilder=Caffeine.newBuilder();

OptionalDoubleCacheConfigdcConfigOpt=Optional.ofNullable(this.dcConfig);

dcConfigOpt.map(DoubleCacheConfig::getInit)

.ifPresent(init-caffeineBuilder.initialCapacity(init));

dcConfigOpt.map(DoubleCacheConfig::getMax)

.ifPresent(max-caffeineBuilder.maximumSize(max));

dcConfigOpt.map(DoubleCacheConfig::getExpireAfterWrite)

.ifPresent(eaw-caffeineBuilder.expireAfterWrite(eaw,TimeUnit.SECONDS));

dcConfigOpt.map(DoubleCacheConfig::getExpireAfterAccess)

.ifPresent(eaa-caffeineBuilder.expireAfterAccess(eaa,TimeUnit.SECONDS));

dcConfigOpt.map(DoubleCacheConfig::getRefreshAfterWrite)

.ifPresent(raw-caffeineBuilder.refreshAfterWrite(raw,TimeUnit.SECONDS));

returncaffeineBuilder.build();

}

getCacheNames方法很簡單，直接返回Map的keySet就可以了，代碼如下：

@Override

publicCollectionStringgetCacheNames(){

returncacheMap.keySet();

}

配置使用

在application.yml文件中配置緩存的參數(shù)，代碼中使用@ConfigurationProperties接收到DoubleCacheConfig類中：

doublecache:

allowNull:true

init:128

max:1024

expireAfterWrite:30#Caffeine過期時間

redisExpire:60#Redis緩存過期時間

配置自定義的DoubleCacheManager作為默認(rèn)的緩存管理器：

@Configuration

publicclassCacheConfig{

@Autowired

DoubleCacheConfigdoubleCacheConfig;

@Bean

publicDoubleCacheManagercacheManager(RedisTemplateObject,ObjectredisTemplate,

DoubleCacheConfigdoubleCacheConfig){

returnnewDoubleCacheManager(redisTemplate,doubleCacheConfig);

}

Service中的代碼還是老樣子，不需要在代碼中手動操作緩存，只要直接在方法上使用@Cache相關(guān)注解即可：

@Service@Slf4j

@AllArgsConstructor

publicclassOrderServiceImplimplementsOrderService{

privatefinalOrderMapperorderMapper;

@Cacheable(value="order",key="#id")

publicOrdergetOrderById(Longid){

OrdermyOrder=orderMapper.selectOne(newLambdaQueryWrapperOrder()

.eq(Order::getId,id));

returnmyOrder;

@CachePut(cacheNames="order",key="#order.id")

publicOrderupdateOrder(Orderorder){

orderMapper.updateById(order);

returnorder;

@CacheEvict(cacheNames="order",key="#id")

publicvoiddeleteOrder(Longid){

orderMapper.deleteById(id);

//沒有注解,使用get(key,callable)方法

publicOrdergetOrderById2(Longid){

DoubleCacheManagercacheManager=SpringContextUtil.getBean(DoubleCacheManager.class);

Cachecache=cacheManager.getCache("order");

Orderorder=(Order)cache.get(id,(CallableObject)()-{

("getdatafromdatabase");

OrdermyOrder=orderMapper.selectOne(newLambdaQueryWrapperOrder()

.eq(Order::getId,id));

returnmyOrder;

returnorder;

注意最后這個沒有添加任何注解的方法，只有以這種方式調(diào)用時才會執(zhí)行我們在DoubleCache中自己實(shí)現(xiàn)的get(key,callable)方法。到這里，基于JSR107規(guī)范和spring接口的兩級緩存改造就完成了，下面我們看一下遺漏的第二個問題。

分布式環(huán)境改造

前面我們說了，在分布式環(huán)境下，可能會存在各個主機(jī)上一級緩存不一致的問題。當(dāng)一臺主機(jī)修改了本地緩存后，其他主機(jī)是沒有感知的，仍然保持了之前的緩存，那么這種情況下就可能取到臟數(shù)據(jù)。既然我們在項(xiàng)目中已經(jīng)使用了Redis，那么就可以使用它的發(fā)布/訂閱功能來使各個節(jié)點(diǎn)的緩存進(jìn)行同步。

定義消息體

在使用Redis發(fā)送消息前，需要先定義一個消息對象。其中的數(shù)據(jù)包括消息要作用于的Cache名稱、操作類型、數(shù)據(jù)以及發(fā)出消息的源主機(jī)標(biāo)識：

@Data

@NoArgsConstructor

@AllArgsConstructor

publicclassCacheMassageimplementsSerializable{

privatestaticfinallongserialVersionUID=-3574997636829868400L;

privateStringcacheName;

privateCacheMsgTypetype;//標(biāo)識更新或刪除操作

privateObjectkey;

privateObjectvalue;

privateStringmsgSource;//源主機(jī)標(biāo)識，用來避免重復(fù)操作

}

定義一個枚舉來標(biāo)識消息的類型，是要進(jìn)行更新還是刪除操作：

publicenumCacheMsgType{

UPDATE,

DELETE;

}

消息體中的msgSource是添加的一個消息源主機(jī)的標(biāo)識，添加這個是為了避免收到當(dāng)前主機(jī)發(fā)送的消息后，再進(jìn)行重復(fù)操作，也就是說收到本機(jī)發(fā)出的消息直接丟掉什么都不做就可以了。源主機(jī)標(biāo)識這里使用的是主機(jī)ip加項(xiàng)目端口的方式，獲取方法如下：

publicstaticStringgetMsgSource()throwsUnknownHostException{

Stringhost=InetAddress.getLocalHost().getHostAddress();

Environmentenv=SpringContextUtil.getBean(Environment.class);

Stringport=env.getProperty("server.port");

returnhost+":"+port;

}

這樣消息體的定義就完成了，之后只要調(diào)用redisTemplate的convertAndSend方法就可以把這個對象發(fā)布到指定的主題上了。

Redis消息配置

要使用Redis的消息監(jiān)聽功能，需要配置兩項(xiàng)內(nèi)容：

MessageListenerAdapter：消息監(jiān)聽適配器，可以在其中指定自定義的監(jiān)聽代理類，并且可以自定義使用哪個方法處理監(jiān)聽邏輯RedisMessageListenerContainer：一個可以為消息監(jiān)聽器提供異步行為的容器，并且提供消息轉(zhuǎn)換和分派等底層功能

@Configuration

publicclassMessageConfig{

publicstaticfinalStringTOPIC="cache.msg";

@Bean

RedisMessageListenerContainercontainer(MessageListenerAdapterlistenerAdapter,

RedisConnectionFactoryredisConnectionFactory){

RedisMessageListenerContainercontainer=newRedisMessageListenerContainer();

container.setConnectionFactory(redisConnectionFactory);

container.addMessageListener(listenerAdapter,newPatternTopic(TOPIC));

returncontainer;

@Bean

MessageListenerAdapteradapter(RedisMessageReceiverreceiver){

returnnewMessageListenerAdapter(receiver,"receive");

}

在上面的監(jiān)聽適配器MessageListenerAdapter中，我們傳入了一個自定義的RedisMessageReceiver接收并處理消息，并指定使用它的receive方法來處理監(jiān)聽到的消息，下面我們就來看看它如何接收消息并消費(fèi)。

消息消費(fèi)邏輯

定義一個類RedisMessageReceiver來接收并消費(fèi)消息，需要在它的方法中實(shí)現(xiàn)以下功能：

反序列化接收到的消息，轉(zhuǎn)換為前面定義的CacheMassage類型對象根據(jù)消息的主機(jī)標(biāo)識判斷這條消息是不是本機(jī)發(fā)出的，如果是那么直接丟棄，只有接收到其他主機(jī)發(fā)出的消息才進(jìn)行處理使用cacheName得到具體使用的那一個DoubleCache實(shí)例根據(jù)消息的類型判斷要執(zhí)行的是更新還是刪除操作，調(diào)用對應(yīng)的方法

@Slf4j@Component

@AllArgsConstructor

publicclassRedisMessageReceiver{

privatefinalRedisTemplateredisTemplate;

privatefinalDoubleCacheManagermanager;

//接收通知，進(jìn)行處理

publicvoidreceive(Stringmessage)throwsUnknownHostException{

CacheMassagemsg=(CacheMassage)redisTemplate

.getValueSerializer().deserialize(message.getBytes());

(msg.toString());

//如果是本機(jī)發(fā)出的消息，那么不進(jìn)行處理

if(msg.getMsgSource().equals(MessageSourceUtil.getMsgSource())){

("收到本機(jī)發(fā)出的消息，不做處理");

return;

DoubleCachecache=(DoubleCache)manager.getCache(msg.getCacheName());

if(msg.getType()==CacheMsgType.UPDATE){

cache.updateL1Cache(msg.getKey(),msg.getValue());

("更新本地緩存");

if(msg.getType()==CacheMsgType.DELETE){

("刪除本地緩存");

cache.evictL1Cache(msg.getKey());

}

在上面的代碼中，調(diào)用了DoubleCache中更新一級緩存方法updateL1Cache、刪除一級緩存方法evictL1Cache，我們會后面在DoubleCache中進(jìn)行添加。

修改DoubleCache

在DoubleCache中先添加上面提到的兩個方法，由CacheManager獲取到具體緩存后調(diào)用，進(jìn)行一級緩存的更新或刪除操作：

//更新一級緩存

publicvoidupdateL1Cache(Objectkey,Objectvalue){

caffeineCache.put(key,value);

//刪除一級緩存

publicvoidevictL1Cache(Objectkey){

caffeineCache.invalidate(key);

}

好