caffeine_redis自定義二級緩存

更新時間：2023年04月19日 11:17:42 作者：小貍花

這篇文章詳細介紹了caffeine_redis 自定義二級緩存，文中有相關(guān)的背景前提與出現(xiàn)的問題，感興趣的同學(xué)可以參考一下

背景

最近產(chǎn)品下發(fā)一個需求：考慮在程序中加緩存，剛開始以為只是 Redis 緩存，后面才直到是本地緩存（Caffeine） + Redis。

在 SpringBoot2.x 后默認的緩存就是 Caffeine，所以本地緩存也選擇了 Caffeine。

ps：我們的數(shù)據(jù)不是從程序中插入或者更新，是每天會有數(shù)據(jù)專門同步。

問題

基于提出的需求，我認為主要有以下兩個問題：

因為有本地緩存，如何保證數(shù)據(jù)一致性。當一個節(jié)點數(shù)據(jù)改變，其他節(jié)點的數(shù)據(jù)如何失效？
數(shù)據(jù)不對，需要重新同步，緩存如何失效？

流程圖

接下來就是配合產(chǎn)品和其他開發(fā)人員畫出流程圖，如下：

使用一張配置表，記錄是否需要緩存，是否開啟緩存，來達到通知時候緩存失效的情況。
因為項目要求一般，即使消息丟失，也不會存在太大的影響，所以最終選擇了 redis 里面的訂閱、發(fā)布功能，實現(xiàn)通知其他節(jié)點失效本地緩存。

開發(fā)

上面問題清楚了，流程圖也清楚了。那就準備開始寫 bug 了。整體思路是自定義注解實現(xiàn)切面，盡量降低對業(yè)務(wù)代碼的耦合度。

CacheConfig

主要是結(jié)合業(yè)務(wù)定義一個 CacheManager，代碼里面的解釋都有。因為這個是直接占用程序內(nèi)存的，所有得特別注意最大可緩存條數(shù)，別把內(nèi)存肝爆了。當然也不能太小了，因為還要考慮命中率的問題。所以這就得結(jié)合實際得業(yè)務(wù)來確定最終的大小。

@Bean(name = JKDDCX)
@Primary
public CacheManager cacheManager() {
     CaffeineCacheManager cacheManager  = new CaffeineCacheManager();
        cacheManager.setCaffeine(Caffeine.newBuilder()
                // 設(shè)置最后一次寫入或訪問后經(jīng)過固定時間過期
                .expireAfterAccess(EXPIRE, TIME_UNIT)
                //設(shè)置本地緩存寫入后過期時間
                .expireAfterWrite(EXPIRE, TIME_UNIT)
                // 初始的緩存空間大小
                .initialCapacity(500)
                // 緩存的最大條數(shù)
                .maximumSize(1000));// 使用人數(shù) * 5 （每個人不同的入?yún)?5 條）\
  return cacheManager;
}

@CaffeineCache

自定義注解，把可以用到的參數(shù)都能加上。

@Target({ ElementType.METHOD ,ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface CaffeineCache {

     public String moudleId() default "";
    
     //用于在數(shù)據(jù)庫中配置參數(shù)
     public String methodId() default "";
     public String cachaName() default "";
    
     //動態(tài)切換實際的 CacheManager
     public String cacheManager() default "";

}

CacheMessageListener

緩存監(jiān)聽器，主要是保證多節(jié)點數(shù)據(jù)一致性的問題。當一個節(jié)點緩存更新，通知其他的節(jié)點相應(yīng)處理。主要技術(shù)是 Redis 的發(fā)布、訂閱功能，實現(xiàn) MessageListener 接口。

當然下面還有個細節(jié)就是一般生產(chǎn)環(huán)境是禁用 Redis#keys 命令的，所以得換個方式掃描對應(yīng)的 key。

public class CacheMessageListener implements MessageListener {
     @Override
    public void onMessage(Message message, byte[] pattern) {
        CacheMessage cacheMessage = (CacheMessage) redisTemplate.getValueSerializer().deserialize(message.getBody());
        logger.info("收到redis清除緩存消息, 開始清除本地緩存, the cacheName is {}, the key is {}", cacheMessage.getCacheName(), cacheMessage.getKey());
//		redisCaffeineCacheManager.clearLocal(cacheMessage.getCacheName(), cacheMessage.getKey());

        /**
         * 如果是一個類上使用了 注解 @CaffeineCache ，那么所有接口都會緩存。
         * 下面的邏輯是：除了當前模塊的接口訪問的入?yún)?key，其他的 redis 緩存都會被清除
         * （比如此模塊的表更新了，但是當前調(diào)用此接口只是緩存了當前這個入?yún)⒌膔edis,其他的數(shù)據(jù)刪除）
         */
        String prefixKey = RedisConstant.WXYMG_DATA_CACHE + cacheMessage.getCacheName();
        Set<String> keys = redisTemplate.execute((RedisCallback<Set<String>>) connection -> {
            Set<String> keysTmp = new HashSet<>();
            Cursor<byte[]> cursor = connection.scan(new ScanOptions.ScanOptionsBuilder().
                    match(prefixKey + "*").
                    count(50).build());
            while (cursor.hasNext()) {
                keysTmp.add(new String(cursor.next()));
            }
            return keysTmp;
        });
        Iterator iterator = keys.iterator();
        while (iterator.hasNext()) {
            if (iterator.next().toString().equals(cacheMessage.getKey())) {
                iterator.remove();
            }
        }
        redisTemplate.delete(keys);

        cacheConfig.cacheManager().getCache(cacheMessage.getCacheName()).clear(); //cacheName 下的都刪除
    }
}

CaffeineCacheAspect

然后就是切面的邏輯處理，里面的內(nèi)容和流程圖一模一樣，只是使用代碼實現(xiàn)了需求。

其中：下面的代碼是 Redis 發(fā)布消息。

redisTemplate.convertAndSend(CacheConfig.TOPIC, new CacheMessage(caffeineCache.cachaName(), redisKey));

CacheMessage

這是在 Redis 發(fā)布消息的時候一個消息體，也是自定義的，可以加更多的參數(shù)屬性

public class CacheMessage implements Serializable {

	private static final long serialVersionUID = -1L;

	private String cacheName;

	private Object key;

	public CacheMessage(String cacheName, Object key) {
		super();
		this.cacheName = cacheName;
		this.key = key;
	}

}