Spring Boot多级缓存实现方案

1.背景

缓存，就是让数据更接近使用者，让访问速度加快，从而提升系统性能。工作机制大概是先从缓存中加载数据，如果没有，再从慢速设备(eg:数据库)中加载数据并同步到缓存中。

所谓多级缓存，是指在整个系统架构的不同系统层面进行数据缓存，以提升访问速度。主要分为三层缓存：网关nginx缓存、分布式缓存、本地缓存。这里的多级缓存就是用redis分布式缓存+caffeine本地缓存整合而来。

平时我们在开发过程中，一般都是使用redis实现分布式缓存、caffeine操作本地缓存，但是发现只使用redis或者是caffeine实现缓存都有一些问题：

一级缓存：Caffeine是一个一个高性能的 Java 缓存库；使用 Window TinyLfu 回收策略，提供了一个近乎最佳的命中率。优点数据就在应用内存所以速度快。缺点受应用内存的限制，所以容量有限；没有持久化，重启服务后缓存数据会丢失；在分布式环境下缓存数据数据无法同步；
二级缓存：redis是一高性能、高可用的key-value数据库，支持多种数据类型，支持集群，和应用服务器分开部署易于横向扩展。优点支持多种数据类型，扩容方便；有持久化，重启应用服务器缓存数据不会丢失；他是一个集中式缓存，不存在在应用服务器之间同步数据的问题。缺点每次都需要访问redis存在IO浪费的情况。

综上所述，我们可以通过整合redis和caffeine实现多级缓存，解决上面单一缓存的痛点，从而做到相互补足。

项目推荐：基于SpringBoot2.x、SpringCloud和SpringCloudAlibaba企业级系统架构底层框架封装，解决业务开发时常见的非功能性需求，防止重复造轮子，方便业务快速开发和企业技术栈框架统一管理。引入组件化的思想实现高内聚低耦合并且高度可配置化，做到可插拔。严格控制包依赖和统一版本管理，做到最少化依赖。注重代码规范和注释，非常适合个人学习和企业使用

Github地址：https://github.com/plasticene/plasticene-boot-starter-parent

Gitee地址：https://gitee.com/plasticene3/plasticene-boot-starter-parent

微信公众号：Shepherd进阶笔记

交流探讨qun：Shepherd_126

2.整合实现

2.1思路

Spring 本来就提供了Cache的支持，最核心的就是实现Cache和CacheManager接口。但是Spring Cache存在以下问题：

Spring Cache 仅支持单一的缓存来源，即：只能选择 Redis 实现或者 Caffeine 实现，并不能同时使用。
数据一致性：各层缓存之间的数据一致性问题，如应用层缓存和分布式缓存之前的数据一致性问题。

由此我们可以通过重新实现Cache和CacheManager接口，整合redis和caffeine，从而实现多级缓存。在讲实现原理之前先看看多级缓存调用逻辑图：

2.2实现

首先，我们需要一个多级缓存配置类，方便对缓存属性的动态配置，通过开关做到可插拔。

@ConfigurationProperties(prefix = "multilevel.cache")
@Data
public class MultilevelCacheProperties {

    /**
     * 一级本地缓存最大比例
     */
    private Double maxCapacityRate = 0.2;

    /**
     * 一级本地缓存与最大缓存初始化大小比例
     */
    private Double initRate = 0.5;

    /**
     * 消息主题
     */
    private String topic = "multilevel-cache-topic";

    /**
     * 缓存名称
     */
    private String name = "multilevel-cache";

    /**
     * 一级本地缓存名称
     */
    private String caffeineName = "multilevel-caffeine-cache";

    /**
     * 二级缓存名称
     */
    private String redisName = "multilevel-redis-cache";

    /**
     * 一级本地缓存过期时间
     */
    private Integer caffeineExpireTime = 300;

    /**
     * 二级缓存过期时间
     */
    private Integer redisExpireTime = 600;


    /**
     * 一级缓存开关
     */
    private Boolean caffeineSwitch = true;

}

在自动配置类使用@EnableConfigurationProperties(MultilevelCacheProperties.class)注入即可使用。

接下来就是重新实现spring的Cache接口，整合caffeine本地缓存和redis分布式缓存实现多级缓存

package com.plasticene.boot.cache.core.manager;

import com.plasticene.boot.cache.core.listener.CacheMessage;
import com.plasticene.boot.cache.core.prop.MultilevelCacheProperties;
import com.plasticene.boot.common.executor.plasticeneThreadExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.caffeine.CaffeineCache;
import org.springframework.cache.support.AbstractValueAdaptingCache;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @author fjzheng
 * @version 1.0
 * @date 2022/7/20 17:03
 */
@Slf4j
public class MultilevelCache extends AbstractValueAdaptingCache {

    @Resource
    private MultilevelCacheProperties multilevelCacheProperties;
    @Resource
    private RedisTemplate redisTemplate;


    ExecutorService cacheExecutor = new plasticeneThreadExecutor(
            Runtime.getRuntime().availableProcessors() * 2,
            Runtime.getRuntime().availableProcessors() * 20,
            Runtime.getRuntime().availableProcessors() * 200,
           "cache-pool"
    );

    private RedisCache redisCache;
    private CaffeineCache caffeineCache;

    public MultilevelCache(boolean allowNullValues,RedisCache redisCache, CaffeineCache caffeineCache) {
        super(allowNullValues);
        this.redisCache = redisCache;
        this.caffeineCache = caffeineCache;
    }


    @Override
    public String getName() {
        return multilevelCacheProperties.getName();

    }

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

    @Override
    public <T> T get(Object key, Callable<T> valueLoader) {
        Object value = lookup(key);
        return (T) value;
    }

    /**
     *  注意：redis缓存的对象object必须序列化 implements Serializable, 不然缓存对象不成功。
     *  注意：这里asyncPublish()方法是异步发布消息，然后让分布式其他节点清除本地缓存,防止当前节点因更新覆盖数据而其他节点本地缓存保存是脏数据
     *  这样本地缓存数据才能成功存入
     * @param key
     * @param value
     */
    @Override
    public void put(@NonNull Object key, Object value) {
        redisCache.put(key, value);
        // 异步清除本地缓存
        if (multilevelCacheProperties.getCaffeineSwitch()) {
            asyncPublish(key, value);
        }
    }

    /**
     * key不存在时，再保存，存在返回当前值不覆盖
     * @param key
     * @param value
     * @return
     */
    @Override
    public ValueWrapper putIfAbsent(@NonNull Object key, Object value) {
        ValueWrapper valueWrapper = redisCache.putIfAbsent(key, value);
        // 异步清除本地缓存
        if (multilevelCacheProperties.getCaffeineSwitch()) {
            asyncPublish(key, value);
        }
        return valueWrapper;
    }


    @Override
    public void evict(Object key) {
        // 先清除redis中缓存数据，然后通过消息推送清除所有节点caffeine中的缓存，
        // 避免短时间内如果先清除caffeine缓存后其他请求会再从redis里加载到caffeine中
        redisCache.evict(key);
        // 异步清除本地缓存
        if (multilevelCacheProperties.getCaffeineSwitch()) {
            asyncPublish(key, null);
        }
    }

    @Override
    public boolean evictIfPresent(Object key) {
        return false;
    }

    @Override
    public void clear() {
        redisCache.clear();
        // 异步清除本地缓存
        if (multilevelCacheProperties.getCaffeineSwitch()) {
            asyncPublish(null, null);
        }
    }



    @Override
    protected Object lookup(Object key) {
        Assert.notNull(key, "key不可为空");
        ValueWrapper value;
        if (multilevelCacheProperties.getCaffeineSwitch()) {
            // 开启一级缓存，先从一级缓存缓存数据
            value = caffeineCache.get(key);
            if (Objects.nonNull(value)) {
                log.info("查询caffeine 一级缓存 key:{}, 返回值是:{}", key, value.get());
                return value.get();
            }
        }
        value = redisCache.get(key);
        if (Objects.nonNull(value)) {
            log.info("查询redis 二级缓存 key:{}, 返回值是:{}", key, value.get());
            // 异步将二级缓存redis写到一级缓存caffeine
            if (multilevelCacheProperties.getCaffeineSwitch()) {
                ValueWrapper finalValue = value;
                cacheExecutor.execute(()->{
                    caffeineCache.put(key, finalValue.get());
                });
            }
            return value.get();
        }
        return null;
    }

    /**
     * 缓存变更时通知其他节点清理本地缓存
     * 异步通过发布订阅主题消息，其他节点监听到之后进行相关本地缓存操作，防止本地缓存脏数据
     */
    void asyncPublish(Object key, Object value) {
        cacheExecutor.execute(()->{
            CacheMessage cacheMessage = new CacheMessage();
            cacheMessage.setCacheName(multilevelCacheProperties.getName());
            cacheMessage.setKey(key);
            cacheMessage.setValue(value);
            redisTemplate.convertAndSend(multilevelCacheProperties.getTopic(), cacheMessage);
        });
    }



}

缓存消息监听：我们通监听caffeine键值的移除、打印日志方便排查问题，通过监听redis发布的消息，实现分布式集群多节点本地缓存清除从而达到数据一致性。

消息体

@Data
public class CacheMessage implements Serializable {
    private String cacheName;
    private Object key;
    private Object value;
    private Integer type;
}

caffeine移除监听：

@Slf4j
public class CaffeineCacheRemovalListener implements RemovalListener<Object, Object> {
    @Override
    public void onRemoval(@Nullable Object k, @Nullable Object v, @NonNull RemovalCause cause) {
        log.info("[移除缓存] key:{} reason:{}", k, cause.name());
        // 超出最大缓存
        if (cause == RemovalCause.SIZE) {

        }
        // 超出过期时间
        if (cause == RemovalCause.EXPIRED) {
            // do something
        }
        // 显式移除
        if (cause == RemovalCause.EXPLICIT) {
            // do something
        }
        // 旧数据被更新
        if (cause == RemovalCause.REPLACED) {
            // do something
        }
    }
}

redis消息监听：

@Slf4j
@Data
public class RedisCacheMessageListener implements MessageListener {

    private CaffeineCache caffeineCache;
    @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("监听的redis message: {}" + message.toString());
        CacheMessage cacheMessage = JsonUtils.parseObject(message.toString(), CacheMessage.class);
        if (Objects.isNull(cacheMessage.getKey())) {
            caffeineCache.invalidate();
        } else {
            caffeineCache.evict(cacheMessage.getKey());
        }
    }
}

最后，通过自动配置类，注入相关bean：

**
 * @author fjzheng
 * @version 1.0
 * @date 2022/7/20 17:24
 */
@Configuration
@EnableConfigurationProperties(MultilevelCacheProperties.class)
public class MultilevelCacheAutoConfiguration {

    @Resource
    private MultilevelCacheProperties multilevelCacheProperties;

    ExecutorService cacheExecutor = new plasticeneThreadExecutor(
            Runtime.getRuntime().availableProcessors() * 2,
            Runtime.getRuntime().availableProcessors() * 20,
            Runtime.getRuntime().availableProcessors() * 200,
            "cache-pool"
    );

    @Bean
    @ConditionalOnMissingBean({RedisTemplate.class})
    public  RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<Object, Object> template = new RedisTemplate<Object, Object>();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setDefaultSerializer(new Jackson2JsonRedisSerializer<>(Object.class));
        template.setHashValueSerializer(new Jackson2JsonRedisSerializer<>(Object.class));
        return template;
    }

    @Bean
    public RedisCache redisCache (RedisConnectionFactory redisConnectionFactory) {
        RedisCacheWriter redisCacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(redisConnectionFactory);
        RedisCacheConfiguration redisCacheConfiguration = defaultCacheConfig();
        redisCacheConfiguration = redisCacheConfiguration.entryTtl(Duration.of(multilevelCacheProperties.getRedisExpireTime(), ChronoUnit.SECONDS));
        redisCacheConfiguration = redisCacheConfiguration.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()));
        redisCacheConfiguration = redisCacheConfiguration.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
        RedisCache redisCache = new CustomRedisCache(multilevelCacheProperties.getRedisName(), redisCacheWriter, redisCacheConfiguration);
        return redisCache;
    }

    /**
     * 由于Caffeine 不会再值过期后立即执行清除，而是在写入或者读取操作之后执行少量维护工作，或者在写入读取很少的情况下，偶尔执行清除操作。
     * 如果我们项目写入或者读取频率很高，那么不用担心。如果想入写入和读取操作频率较低，那么我们可以通过Cache.cleanUp()或者加scheduler去定时执行清除操作。
     * Scheduler可以迅速删除过期的元素，***Java 9 +***后的版本，可以通过Scheduler.systemScheduler(), 调用系统线程，达到定期清除的目的
     * @return
     */
    @Bean
    @ConditionalOnClass(CaffeineCache.class)
    @ConditionalOnProperty(name = "multilevel.cache.caffeineSwitch", havingValue = "true", matchIfMissing = true)
    public CaffeineCache caffeineCache() {
        int maxCapacity = (int) (Runtime.getRuntime().totalMemory() * multilevelCacheProperties.getMaxCapacityRate());
        int initCapacity = (int) (maxCapacity * multilevelCacheProperties.getInitRate());
        CaffeineCache caffeineCache = new CaffeineCache(multilevelCacheProperties.getCaffeineName(), Caffeine.newBuilder()
                // 设置初始缓存大小
                .initialCapacity(initCapacity)
                // 设置最大缓存
                .maximumSize(maxCapacity)
                // 设置缓存线程池
                .executor(cacheExecutor)
                // 设置定时任务执行过期清除操作
//                .scheduler(Scheduler.systemScheduler())
                // 监听器(超出最大缓存)
                .removalListener(new CaffeineCacheRemovalListener())
                // 设置缓存读时间的过期时间
                .expireAfterAccess(Duration.of(multilevelCacheProperties.getCaffeineExpireTime(), ChronoUnit.SECONDS))
                // 开启metrics监控
                .recordStats()
                .build());
        return caffeineCache;
    }

    @Bean
    @ConditionalOnBean({CaffeineCache.class, RedisCache.class})
    public MultilevelCache multilevelCache(RedisCache redisCache, CaffeineCache caffeineCache) {
        MultilevelCache multilevelCache = new MultilevelCache(true, redisCache, caffeineCache);
        return multilevelCache;
    }

    @Bean
    public RedisCacheMessageListener redisCacheMessageListener(@Autowired CaffeineCache caffeineCache) {
        RedisCacheMessageListener redisCacheMessageListener = new RedisCacheMessageListener();
        redisCacheMessageListener.setCaffeineCache(caffeineCache);
        return redisCacheMessageListener;
    }



    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(@Autowired RedisConnectionFactory redisConnectionFactory,
                                                                       @Autowired RedisCacheMessageListener redisCacheMessageListener) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
        redisMessageListenerContainer.addMessageListener(redisCacheMessageListener, new ChannelTopic(multilevelCacheProperties.getTopic()));
        return redisMessageListenerContainer;
    }

}

3.使用

使用非常简单，只需要通过multilevelCache操作即可：

@RestController
@RequestMapping("/api/data")
@Api(tags = "api数据")
@Slf4j
public class ApiDataController {

    @Resource
    private MultilevelCache multilevelCache;

  
    @GetMapping("/put/cache")
    public void put() {
        DataSource ds = new DataSource();
        ds.setName("多级缓存");
        ds.setType(1);
        ds.setCreateTime(new Date());
        ds.setHost("127.0.0.1");
        multilevelCache.put("test-key", ds);
    }

    @GetMapping("/get/cache")
    public DataSource get() {
        DataSource dataSource = multilevelCache.get("test-key", DataSource.class);
        return dataSource;
    }

}

4.总结

以上全部就是关于多级缓存的实现方案总结，多级缓存就是为了解决项目服务中单一缓存使用不足的缺点。应用场景有：接口权限校验，每次请求接口都需要根据当前登录人有哪些角色，角色有哪些权限，如果每次都去查数据库性能开销比较严重，再加上权限一般不怎么会频繁变更，所以使用多级缓存是最合适不过了；还有就是很多管理系统列表界面都有组织架构信息(所属部门、小组等)，这些信息同样可以使用多级缓存来完美提升性能。