快捷導(dǎo)航

Redis+Caffeine實(shí)現(xiàn)分布式二級(jí)緩存組件實(shí)戰(zhàn)教程

更新時(shí)間：2022年08月06日 10:19:20 作者：上帝愛吃蘋果-

這篇文章主要介紹了Redis+Caffeine實(shí)現(xiàn)分布式二級(jí)緩存組件實(shí)戰(zhàn)教程,介紹了分布式二級(jí)緩存的優(yōu)勢(shì)，使用組件的方法，通過示例代碼給大家介紹的非常詳細(xì)，需要的朋友可以參考下

前言

在生產(chǎn)中已有實(shí)踐，本組件僅做個(gè)人學(xué)習(xí)交流分享使用。github：https://github.com/axinSoochow/redis-caffeine-cache-starter
個(gè)人水平有限，歡迎大家在評(píng)論區(qū)輕噴。

所謂二級(jí)緩存

緩存就是將數(shù)據(jù)從讀取較慢的介質(zhì)上讀取出來放到讀取較快的介質(zhì)上，如磁盤-->內(nèi)存。

平時(shí)我們會(huì)將數(shù)據(jù)存儲(chǔ)到磁盤上，如：數(shù)據(jù)庫(kù)。如果每次都從數(shù)據(jù)庫(kù)里去讀取，會(huì)因?yàn)榇疟P本身的IO影響讀取速度，所以就有了像redis這種的內(nèi)存緩存?？梢詫?shù)據(jù)讀取出來放到內(nèi)存里，這樣當(dāng)需要獲取數(shù)據(jù)時(shí)，就能夠直接從內(nèi)存中拿到數(shù)據(jù)返回，能夠很大程度的提高速度。
但是一般redis是單獨(dú)部署成集群，所以會(huì)有網(wǎng)絡(luò)IO上的消耗，雖然與redis集群的鏈接已經(jīng)有連接池這種工具，但是數(shù)據(jù)傳輸上也還是會(huì)有一定消耗。所以就有了進(jìn)程內(nèi)緩存，如：caffeine。當(dāng)應(yīng)用內(nèi)緩存有符合條件的數(shù)據(jù)時(shí)，就可以直接使用，而不用通過網(wǎng)絡(luò)到redis中去獲取，這樣就形成了兩級(jí)緩存。應(yīng)用內(nèi)緩存叫做一級(jí)緩存，遠(yuǎn)程緩存（如redis）叫做二級(jí)緩存。

系統(tǒng)是否需要緩存CPU占用:如果你有某些應(yīng)用需要消耗大量的cpu去計(jì)算獲得結(jié)果。
數(shù)據(jù)庫(kù)IO占用:如果你發(fā)現(xiàn)你的數(shù)據(jù)庫(kù)連接池比較空閑，那么不應(yīng)該用緩存。但是如果數(shù)據(jù)庫(kù)連接池比較繁忙，甚至經(jīng)常報(bào)出連接不夠的報(bào)警，那么是時(shí)候應(yīng)該考慮緩存了。

分布式二級(jí)緩存的優(yōu)勢(shì)

Redis用來存儲(chǔ)熱點(diǎn)數(shù)據(jù)，Redis中沒有的數(shù)據(jù)則直接去數(shù)據(jù)庫(kù)訪問。
已經(jīng)有Redis了，干嘛還需要了解Guava，Caffeine這些進(jìn)程緩存呢:

Redis如果不可用，這個(gè)時(shí)候我們只能訪問數(shù)據(jù)庫(kù)，很容易造成雪崩，但一般不會(huì)出現(xiàn)這種情況。
訪問Redis會(huì)有一定的網(wǎng)絡(luò)I/O以及序列化反序列化開銷，雖然性能很高但是其終究沒有本地方法快，可以將最熱的數(shù)據(jù)存放在本地，以便進(jìn)一步加快訪問速度。這個(gè)思路并不是我們做互聯(lián)網(wǎng)架構(gòu)獨(dú)有的，在計(jì)算機(jī)系統(tǒng)中使用L1,L2,L3多級(jí)緩存，用來減少對(duì)內(nèi)存的直接訪問，從而加快訪問速度。

所以如果僅僅是使用Redis，能滿足我們大部分需求，但是當(dāng)需要追求更高的性能以及更高的可用性的時(shí)候，那就不得不了解多級(jí)緩存。

二級(jí)緩存操作過程數(shù)據(jù)讀流程描述

redis 與本地緩存都查詢不到值的時(shí)候，會(huì)觸發(fā)更新過程，整個(gè)過程是加鎖的緩存失效流程描述

redis更新與刪除緩存key都會(huì)觸發(fā)，清除redis緩存后

如何使用組件？

組件是基于Spring Cache框架上改造的，在項(xiàng)目中使用分布式緩存，僅僅需要在緩存注解上增加：cacheManager ="L2_CacheManager"，或者 cacheManager = CacheRedisCaffeineAutoConfiguration.分布式二級(jí)緩存

//這個(gè)方法會(huì)使用分布式二級(jí)緩存來提供查詢
@Cacheable(cacheNames = CacheNames.CACHE_12HOUR, cacheManager = "L2_CacheManager")
public Config getAllValidateConfig() { 
}

如果你想既使用分布式緩存，又想用分布式二級(jí)緩存組件，那你需要向Spring注入一個(gè) @Primary 的 CacheManager bean

@Primary
@Bean("deaultCacheManager")
public RedisCacheManager cacheManager(RedisConnectionFactory factory) {
    // 生成一個(gè)默認(rèn)配置，通過config對(duì)象即可對(duì)緩存進(jìn)行自定義配置
    RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
    // 設(shè)置緩存的默認(rèn)過期時(shí)間，也是使用Duration設(shè)置
    config = config.entryTtl(Duration.ofMinutes(2)).disableCachingNullValues();

    // 設(shè)置一個(gè)初始化的緩存空間set集合
    Set<String> cacheNames =  new HashSet<>();
    cacheNames.add(CacheNames.CACHE_15MINS);
    cacheNames.add(CacheNames.CACHE_30MINS);

    // 對(duì)每個(gè)緩存空間應(yīng)用不同的配置
    Map<String, RedisCacheConfiguration> configMap = new HashMap<>();
    configMap.put(CacheNames.CACHE_15MINS, config.entryTtl(Duration.ofMinutes(15)));
    configMap.put(CacheNames.CACHE_30MINS, config.entryTtl(Duration.ofMinutes(30)));
  
    // 使用自定義的緩存配置初始化一個(gè)cacheManager
    RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
        .initialCacheNames(cacheNames)  // 注意這兩句的調(diào)用順序，一定要先調(diào)用該方法設(shè)置初始化的緩存名，再初始化相關(guān)的配置
        .withInitialCacheConfigurations(configMap)
        .build();
    return cacheManager;
}

然后:

//這個(gè)方法會(huì)使用分布式二級(jí)緩存
@Cacheable(cacheNames = CacheNames.CACHE_12HOUR, cacheManager = "L2_CacheManager")
public Config getAllValidateConfig() {
}

//這個(gè)方法會(huì)使用分布式緩存
@Cacheable(cacheNames = CacheNames.CACHE_12HOUR)
public Config getAllValidateConfig2() {
}

核心實(shí)現(xiàn)方法

核心其實(shí)就是實(shí)現(xiàn) org.springframework.cache.CacheManager接口與繼承org.springframework.cache.support.AbstractValueAdaptingCache，在Spring緩存框架下實(shí)現(xiàn)緩存的讀與寫。

RedisCaffeineCacheManager實(shí)現(xiàn)CacheManager 接口

RedisCaffeineCacheManager.class 主要來管理緩存實(shí)例，根據(jù)不同的 CacheNames 生成對(duì)應(yīng)的緩存管理bean，然后放入一個(gè)map中。

package com.axin.idea.rediscaffeinecachestarter.support;

import com.axin.idea.rediscaffeinecachestarter.CacheRedisCaffeineProperties;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.stats.CacheStats;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RedisCaffeineCacheManager implements CacheManager {

    private final Logger logger = LoggerFactory.getLogger(RedisCaffeineCacheManager.class);

    private static ConcurrentMap<String, Cache> cacheMap = new ConcurrentHashMap<String, Cache>();

    private CacheRedisCaffeineProperties cacheRedisCaffeineProperties;

    private RedisTemplate<Object, Object> stringKeyRedisTemplate;

    private boolean dynamic = true;

    private Set<String> cacheNames;
    {
        cacheNames = new HashSet<>();
        cacheNames.add(CacheNames.CACHE_15MINS);
        cacheNames.add(CacheNames.CACHE_30MINS);
        cacheNames.add(CacheNames.CACHE_60MINS);
        cacheNames.add(CacheNames.CACHE_180MINS);
        cacheNames.add(CacheNames.CACHE_12HOUR);
    }
    public RedisCaffeineCacheManager(CacheRedisCaffeineProperties cacheRedisCaffeineProperties,
                                     RedisTemplate<Object, Object> stringKeyRedisTemplate) {
        super();
        this.cacheRedisCaffeineProperties = cacheRedisCaffeineProperties;
        this.stringKeyRedisTemplate = stringKeyRedisTemplate;
        this.dynamic = cacheRedisCaffeineProperties.isDynamic();
    }

    //——————————————————————— 進(jìn)行緩存工具 ——————————————————————
    /**
    * 清除所有進(jìn)程緩存
    */
    public void clearAllCache() {
        stringKeyRedisTemplate.convertAndSend(cacheRedisCaffeineProperties.getRedis().getTopic(), new CacheMessage(null, null));
    }

    /**
    * 返回所有進(jìn)程緩存(二級(jí)緩存)的統(tǒng)計(jì)信息
    * result:{"緩存名稱":統(tǒng)計(jì)信息}
    * @return
    */
    public static Map<String, CacheStats> getCacheStats() {
        if (CollectionUtils.isEmpty(cacheMap)) {
            return null;
        }

        Map<String, CacheStats> result = new LinkedHashMap<>();
        for (Cache cache : cacheMap.values()) {
            RedisCaffeineCache caffeineCache = (RedisCaffeineCache) cache;
            result.put(caffeineCache.getName(), caffeineCache.getCaffeineCache().stats());
        }
        return result;
    }

    //—————————————————————————— core —————————————————————————
    @Override
    public Cache getCache(String name) {
        Cache cache = cacheMap.get(name);
        if(cache != null) {
            return cache;
        }
        if(!dynamic && !cacheNames.contains(name)) {
            return null;
        }

        cache = new RedisCaffeineCache(name, stringKeyRedisTemplate, caffeineCache(name), cacheRedisCaffeineProperties);
        Cache oldCache = cacheMap.putIfAbsent(name, cache);
        logger.debug("create cache instance, the cache name is : {}", name);
        return oldCache == null ? cache : oldCache;
    }

    @Override
    public Collection<String> getCacheNames() {
        return this.cacheNames;
    }

    public void clearLocal(String cacheName, Object key) {
        //cacheName為null 清除所有進(jìn)程緩存
        if (cacheName == null) {
            log.info("清除所有本地緩存");
            cacheMap = new ConcurrentHashMap<>();
            return;
        }

        Cache cache = cacheMap.get(cacheName);
        if(cache == null) {
            return;
        }

        RedisCaffeineCache redisCaffeineCache = (RedisCaffeineCache) cache;
        redisCaffeineCache.clearLocal(key);
    }

    /**
    * 實(shí)例化本地一級(jí)緩存
    * @param name
    * @return
    */
    private com.github.benmanes.caffeine.cache.Cache<Object, Object> caffeineCache(String name) {
        Caffeine<Object, Object> cacheBuilder = Caffeine.newBuilder();
        CacheRedisCaffeineProperties.CacheDefault cacheConfig;
        switch (name) {
            case CacheNames.CACHE_15MINS:
                cacheConfig = cacheRedisCaffeineProperties.getCache15m();
                break;
            case CacheNames.CACHE_30MINS:
                cacheConfig = cacheRedisCaffeineProperties.getCache30m();
                break;
            case CacheNames.CACHE_60MINS:
                cacheConfig = cacheRedisCaffeineProperties.getCache60m();
                break;
            case CacheNames.CACHE_180MINS:
                cacheConfig = cacheRedisCaffeineProperties.getCache180m();
                break;
            case CacheNames.CACHE_12HOUR:
                cacheConfig = cacheRedisCaffeineProperties.getCache12h();
                break;
            default:
                cacheConfig = cacheRedisCaffeineProperties.getCacheDefault();
        }
        long expireAfterAccess = cacheConfig.getExpireAfterAccess();
        long expireAfterWrite = cacheConfig.getExpireAfterWrite();
        int initialCapacity = cacheConfig.getInitialCapacity();
        long maximumSize = cacheConfig.getMaximumSize();
        long refreshAfterWrite = cacheConfig.getRefreshAfterWrite();

        log.debug("本地緩存初始化：");
        if (expireAfterAccess > 0) {
            log.debug("設(shè)置本地緩存訪問后過期時(shí)間，{}秒", expireAfterAccess);
            cacheBuilder.expireAfterAccess(expireAfterAccess, TimeUnit.SECONDS);
        }
        if (expireAfterWrite > 0) {
            log.debug("設(shè)置本地緩存寫入后過期時(shí)間，{}秒", expireAfterWrite);
            cacheBuilder.expireAfterWrite(expireAfterWrite, TimeUnit.SECONDS);
        }
        if (initialCapacity > 0) {
            log.debug("設(shè)置緩存初始化大小{}", initialCapacity);
            cacheBuilder.initialCapacity(initialCapacity);
        }
        if (maximumSize > 0) {
            log.debug("設(shè)置本地緩存最大值{}", maximumSize);
            cacheBuilder.maximumSize(maximumSize);
        }
        if (refreshAfterWrite > 0) {
            cacheBuilder.refreshAfterWrite(refreshAfterWrite, TimeUnit.SECONDS);
        }
        cacheBuilder.recordStats();
        return cacheBuilder.build();
    }
}

RedisCaffeineCache 繼承 AbstractValueAdaptingCache

核心是get方法與put方法。

package com.axin.idea.rediscaffeinecachestarter.support;

import com.axin.idea.rediscaffeinecachestarter.CacheRedisCaffeineProperties;
import com.github.benmanes.caffeine.cache.Cache;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cache.support.AbstractValueAdaptingCache;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class RedisCaffeineCache extends AbstractValueAdaptingCache {

    private final Logger logger = LoggerFactory.getLogger(RedisCaffeineCache.class);

    private String name;

    private RedisTemplate<Object, Object> redisTemplate;

    @Getter
    private Cache<Object, Object> caffeineCache;

    private String cachePrefix;

    /**
     * 默認(rèn)key超時(shí)時(shí)間 3600s
     */
    private long defaultExpiration = 3600;

    private Map<String, Long> defaultExpires = new HashMap<>();
    {
        defaultExpires.put(CacheNames.CACHE_15MINS, TimeUnit.MINUTES.toSeconds(15));
        defaultExpires.put(CacheNames.CACHE_30MINS, TimeUnit.MINUTES.toSeconds(30));
        defaultExpires.put(CacheNames.CACHE_60MINS, TimeUnit.MINUTES.toSeconds(60));
        defaultExpires.put(CacheNames.CACHE_180MINS, TimeUnit.MINUTES.toSeconds(180));
        defaultExpires.put(CacheNames.CACHE_12HOUR, TimeUnit.HOURS.toSeconds(12));
    }

    private String topic;
    private Map<String, ReentrantLock> keyLockMap = new ConcurrentHashMap();

    protected RedisCaffeineCache(boolean allowNullValues) {
        super(allowNullValues);
    }

    public RedisCaffeineCache(String name, RedisTemplate<Object, Object> redisTemplate,
                              Cache<Object, Object> caffeineCache, CacheRedisCaffeineProperties cacheRedisCaffeineProperties) {
        super(cacheRedisCaffeineProperties.isCacheNullValues());
        this.name = name;
        this.redisTemplate = redisTemplate;
        this.caffeineCache = caffeineCache;
        this.cachePrefix = cacheRedisCaffeineProperties.getCachePrefix();
        this.defaultExpiration = cacheRedisCaffeineProperties.getRedis().getDefaultExpiration();
        this.topic = cacheRedisCaffeineProperties.getRedis().getTopic();
        defaultExpires.putAll(cacheRedisCaffeineProperties.getRedis().getExpires());
    }

    @Override
    public String getName() {
        return this.name;
    }

    @Override
    public Object getNativeCache() {
        return this;
    }

    @Override
    public <T> T get(Object key, Callable<T> valueLoader) {
        Object value = lookup(key);
        if (value != null) {
            return (T) value;
        }
        //key在redis和緩存中均不存在
        ReentrantLock lock = keyLockMap.get(key.toString());

        if (lock == null) {
            logger.debug("create lock for key : {}", key);
            keyLockMap.putIfAbsent(key.toString(), new ReentrantLock());
            lock = keyLockMap.get(key.toString());
        }
        try {
            lock.lock();
            value = lookup(key);
            if (value != null) {
                return (T) value;
            }
            //執(zhí)行原方法獲得value
            value = valueLoader.call();
            Object storeValue = toStoreValue(value);
            put(key, storeValue);
            return (T) value;
        } catch (Exception e) {
            throw new ValueRetrievalException(key, valueLoader, e.getCause());
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void put(Object key, Object value) {
        if (!super.isAllowNullValues() && value == null) {
            this.evict(key);
            return;
        }
        long expire = getExpire();
        logger.debug("put：{},expire:{}", getKey(key), expire);
        redisTemplate.opsForValue().set(getKey(key), toStoreValue(value), expire, TimeUnit.SECONDS);

        //緩存變更時(shí)通知其他節(jié)點(diǎn)清理本地緩存
        push(new CacheMessage(this.name, key));
        //此處put沒有意義，會(huì)收到自己發(fā)送的緩存key失效消息
//        caffeineCache.put(key, value);
    }

    @Override
    public ValueWrapper putIfAbsent(Object key, Object value) {
        Object cacheKey = getKey(key);
        // 使用setIfAbsent原子性操作
        long expire = getExpire();
        boolean setSuccess;
        setSuccess = redisTemplate.opsForValue().setIfAbsent(getKey(key), toStoreValue(value), Duration.ofSeconds(expire));

        Object hasValue;
        //setNx結(jié)果
        if (setSuccess) {
            push(new CacheMessage(this.name, key));
            hasValue = value;
        }else {
            hasValue = redisTemplate.opsForValue().get(cacheKey);
        }

        caffeineCache.put(key, toStoreValue(value));
        return toValueWrapper(hasValue);
    }

    @Override
    public void evict(Object key) {
        // 先清除redis中緩存數(shù)據(jù)，然后清除caffeine中的緩存，避免短時(shí)間內(nèi)如果先清除caffeine緩存后其他請(qǐng)求會(huì)再?gòu)膔edis里加載到caffeine中
        redisTemplate.delete(getKey(key));

        push(new CacheMessage(this.name, key));

        caffeineCache.invalidate(key);
    }

    @Override
    public void clear() {
        // 先清除redis中緩存數(shù)據(jù)，然后清除caffeine中的緩存，避免短時(shí)間內(nèi)如果先清除caffeine緩存后其他請(qǐng)求會(huì)再?gòu)膔edis里加載到caffeine中
        Set<Object> keys = redisTemplate.keys(this.name.concat(":*"));
        for (Object key : keys) {
            redisTemplate.delete(key);
        }

        push(new CacheMessage(this.name, null));
        caffeineCache.invalidateAll();
    }

    /**
     * 取值邏輯
     * @param key
     * @return
     */
    @Override
    protected Object lookup(Object key) {
        Object cacheKey = getKey(key);
        Object value = caffeineCache.getIfPresent(key);
        if (value != null) {
            logger.debug("從本地緩存中獲得key, the key is : {}", cacheKey);
            return value;
        }

        value = redisTemplate.opsForValue().get(cacheKey);

        if (value != null) {
            logger.debug("從redis中獲得值，將值放到本地緩存中, the key is : {}", cacheKey);
            caffeineCache.put(key, value);
        }
        return value;
    }

    /**
     * @description 清理本地緩存
     */
    public void clearLocal(Object key) {
        logger.debug("clear local cache, the key is : {}", key);
        if (key == null) {
            caffeineCache.invalidateAll();
        } else {
            caffeineCache.invalidate(key);
        }
    }

    //————————————————————————————私有方法——————————————————————————

    private Object getKey(Object key) {
        String keyStr = this.name.concat(":").concat(key.toString());
        return StringUtils.isEmpty(this.cachePrefix) ? keyStr : this.cachePrefix.concat(":").concat(keyStr);
    }

    private long getExpire() {
        long expire = defaultExpiration;
        Long cacheNameExpire = defaultExpires.get(this.name);
        return cacheNameExpire == null ? expire : cacheNameExpire.longValue();
    }

    /**
     * @description 緩存變更時(shí)通知其他節(jié)點(diǎn)清理本地緩存
     */
    private void push(CacheMessage message) {
        redisTemplate.convertAndSend(topic, message);
    }
}

關(guān)于分布式本地緩存失效

現(xiàn)在的線上生產(chǎn)的都是多個(gè)節(jié)點(diǎn)，如果本節(jié)點(diǎn)的緩存失效了，是需要通過中間件來通知其他節(jié)點(diǎn)失效消息的。本組件考慮到學(xué)習(xí)分享讓大家引入的依賴少點(diǎn)，就直接通過 redis 來發(fā)送消息了，實(shí)際生產(chǎn)過程中換成成熟的消息中間件（kafka、RocketMQ）來做通知更為穩(wěn)妥。

到此這篇關(guān)于Redis+Caffeine實(shí)現(xiàn)分布式二級(jí)緩存組件實(shí)戰(zhàn)教程的文章就介紹到這了,更多相關(guān)Redis Caffeine分布式二級(jí)緩存內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: