基于Redission 之Caffeine和Redis集成工

概述：

要将Caffeine本地缓存和Redisson分布式缓存结合起来使用，可以创建一个工具类，它首先尝试从本地Caffeine缓存中获取数据，如果本地缓存中没有找到，则从Redisson分布式缓存中获取，并在获取后将数据回填到本地缓存中。

注意点：

并发处理：Caffeine 已经是线程安全的，所以本地缓存的并发访问不是问题。对于 Redisson，客户端本身也是线程安全的，但是在处理写回策略和缓存穿透时，可能需要额外的并发控制。
写回策略（Write-Back / Write-Behind）：在这种策略下，数据首先写入本地缓存，然后异步地写入后端存储（例如 Redis）。这可以通过一个队列和后台线程来实现，该线程定期将更改写入后端存储。
缓存穿透保护：缓存穿透是指查询不存在的数据，导致请求直接打到数据库上。为了防止缓存穿透，可以使用空对象模式或布隆过滤器。空对象模式是指即使值不存在也在缓存中存储一个特殊的空对象，而布隆过滤器可以在请求到达缓存之前过滤掉不存在的键。
同步写入Redis和本地缓存：当本地缓存被写入时，同时将数据同步写入Redis。这样可以确保两者的数据一致性。但这种方法会增加每次写入操作的延迟。
使用锁或同步机制：在更新本地缓存的同时，使用锁或其他同步机制来确保数据也被写入Redis。如果本地缓存失效，可以通过锁来保证在读取Redis之前数据已经被写入。
设置合理的过期时间：在Redis中为缓存数据设置一个比本地缓存更长的过期时间，这样即使本地缓存失效，数据仍然可以从Redis中获取。
延迟本地缓存的过期时间：可以在本地缓存的基础上添加一个短暂的延迟时间，以确保Redis中的数据在本地缓存失效前已经更新。
使用缓存刷新策略：定期或在本地缓存即将失效时，异步刷新本地缓存的数据。这样可以确保本地缓存中的数据在大多数时间都是最新的。

工具类：

以下是这样一个工具类的简单示例：

java复制代码import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.RemovalCause;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import com.github.benmanes.caffeine.cache.stats.CacheStats;
import com.github.benmanes.caffeine.cache.RemovalListener;

/**
 * @Author derek_smart
 * @Date 202/4/24 14:55
 * @Description caffeine和redis 缓存组合工具类
 * <p>
 * 写回策略（Write-Back / Write-Behind）：在这种策略下，数据首先写入本地缓存，然后异步地写入后端存储（例如 Redis）。这可以通过一个队列和后台线程来实现，该线程定期将更改写入后端存储。
 * 一个特殊的空对象 `NULL_PLACEHOLDER` 存储到本地缓存和 Redis 中，这样下次查询相同的键时就能直接从缓存中获取到空对象，从而防止缓存穿透。
 */
public class HybridCache<K, V> {

    private final Cache<K, V> localCache;
    private final RedissonClient redissonClient;
    private final ExecutorService writeBehindExecutor;
    private final long redisExpiration; // Redis 缓存过期时间，单位：秒
    private static final Object NULL_PLACEHOLDER = new Object();
    // 使用 ConcurrentHashMap 来存储锁
    private final ConcurrentHashMap<K, Lock> locks = new ConcurrentHashMap<>();

    public HybridCache(RedissonClient redissonClient, long maxSize, long expireAfterWrite, TimeUnit timeUnit, long redisExpiration) {
        this.redissonClient = redissonClient;
        this.redisExpiration = redisExpiration;
        this.writeBehindExecutor = Executors.newSingleThreadExecutor(); // 用于写回策略的单线程执行器

        RemovalListener<K, V> writeBehindRemovalListener = (K key, V value, RemovalCause cause) -> {
            if (cause.wasEvicted()) {
                writeBehindExecutor.submit(() -> redissonClient.getBucket(key.toString()).set(value, redisExpiration, TimeUnit.SECONDS));
            }
        };
        this.localCache = Caffeine.newBuilder()
                .maximumSize(maxSize)
                .expireAfterWrite(expireAfterWrite, timeUnit)
                .removalListener((K key, V value, RemovalCause cause) -> {
                    System.out.printf("Key %s was removed (%s)%n", key, cause);
                })
                .recordStats()
                .build();
    }

    public V get(K key) {
        // 尝试从本地缓存获取数据
        V value = localCache.getIfPresent(key);
        if (value != null) {
            return value;
        }

        // 本地缓存没有找到，尝试从 Redis 获取
        RBucket<V> bucket = redissonClient.getBucket(key.toString());
        value = bucket.get();
        if (value != null) {
            // 将数据回填到本地缓存
            localCache.put(key, value);
        }
        return value;
    }

    /**
     * 添加了一个`ConcurrentHashMap`来存储锁对象，并在获取数据时使用了一个双重检查锁定模式。
     * 当本地缓存中没有数据时，首先获取一个锁，然后再次检查本地缓存以确保数据在获取锁的过程中没有被其他线程填充。
     * 如果本地缓存仍然没有数据，会从Redis获取数据，如果Redis也没有数据，则从数据源加载数据并更新Redis和本地缓存。
     * 这样的策略可以减少缓存击穿的风险。
     *
     * @param key
     * @param mappingFunction
     * @return
     */
    public V get4Consistency(K key, Function<? super K, ? extends V> mappingFunction) {
        // 先尝试从本地缓存获取数据
        V value = localCache.getIfPresent(key);
        if (value != null && value != NULL_PLACEHOLDER) {
            return value;
        }

        // 获取锁对象，如果不存在则创建一个新的
        Lock lock = locks.computeIfAbsent(key, k -> new ReentrantLock());
        try {
            // 锁定当前键，以便同步更新操作
            lock.lock();
            // 再次检查本地缓存，以防在获取锁的过程中数据被更新
            value = localCache.getIfPresent(key);
            if (value != null && value != NULL_PLACEHOLDER) {
                return value;
            }

            // 尝试从 Redis 获取
            RBucket<V> bucket = redissonClient.getBucket(key.toString());
            V redisValue = bucket.get();

            if (redisValue != null) {
                // 如果在 Redis 中找到了，将其回填到本地缓存
                localCache.put(key, redisValue);
                return redisValue;
            }

            // 从数据源加载数据
            V loadedValue = mappingFunction.apply(key);
            if (loadedValue == null) {
                // 存储空对象到本地缓存和 Redis 防止缓存穿透
                localCache.put(key, (V) NULL_PLACEHOLDER);
                bucket.set((V) NULL_PLACEHOLDER, redisExpiration, TimeUnit.SECONDS);
            } else {
                // 先将加载的数据写入 Redis
                bucket.set(loadedValue, redisExpiration, TimeUnit.SECONDS);
                // 然后回填到本地缓存
                localCache.put(key, loadedValue);
            }
            return loadedValue;
        } finally {
            // 释放锁
            lock.unlock();
            // 移除锁对象，避免内存泄漏
            locks.remove(key);
        }
    }

    public V get(K key, Function<? super K, ? extends V> mappingFunction) {
        // 尝试从本地缓存获取数据
        V value = localCache.get(key, k -> {
            // 尝试从 Redis 获取
            RBucket<V> bucket = redissonClient.getBucket(k.toString());
            V redisValue = bucket.get();
            if (redisValue != null) {
                return redisValue;
            }

            // 从数据源加载数据
            V loadedValue = mappingFunction.apply(k);
            if (loadedValue == null) {
                // 存储空对象到本地缓存和 Redis 防止缓存穿透
                localCache.put(k, (V) NULL_PLACEHOLDER);
                bucket.set((V) NULL_PLACEHOLDER, redisExpiration, TimeUnit.SECONDS);
                return null;
            }

            // 将加载的数据写入 Redis
            bucket.set(loadedValue, redisExpiration, TimeUnit.SECONDS);
            return loadedValue;
        });

        return value == NULL_PLACEHOLDER ? null : value;
    }

    public void put(K key, V value) {
        // 同时更新本地缓存和 Redis 缓存
        localCache.put(key, value);
        redissonClient.getBucket(key.toString()).set(value, redisExpiration, TimeUnit.SECONDS);
    }

    public void invalidate(K key) {
        // 同时移除本地缓存和 Redis 缓存中的数据
        localCache.invalidate(key);
        redissonClient.getBucket(key.toString()).delete();
    }

    public void refresh(K key, Function<? super K, ? extends V> mappingFunction) {
        // 从数据源重新加载数据并更新缓存
        V value = mappingFunction.apply(key);
        if (value != null) {
            put(key, value);
        }
    }

    // 批量获取数据
    public Map<K, V> getAll(Iterable<? extends K> keys) {
        Map<K, V> allValues = localCache.getAllPresent(keys);
        if (allValues.size() == keys.spliterator().getExactSizeIfKnown()) {
            return allValues;
        }

        // 获取缺失的键
        List<K> missingKeys = StreamSupport.stream(keys.spliterator(), false)
                .filter(key -> !allValues.containsKey(key))
                .collect(Collectors.toList());

        // 从 Redis 批量获取缺失的键
        Map<K, V> redisValues = missingKeys.stream()
                .collect(Collectors.toMap(Function.identity(), key -> redissonClient.<V>getBucket(key.toString()).get()));

        // 将 Redis 中获取的值回填到本地缓存
        localCache.putAll(redisValues);

        // 合并两个 map 并返回
        allValues.putAll(redisValues);
        return allValues;
    }

    // 批量写入数据
    public void putAll(Map<? extends K, ? extends V> map) {
        localCache.putAll(map);
        map.forEach((key, value) -> redissonClient.getBucket(key.toString()).set(value, redisExpiration, TimeUnit.SECONDS));
    }

    // 批量移除数据
    public void invalidateAll(Iterable<? extends K> keys) {
        localCache.invalidateAll(keys);
        keys.forEach(key -> redissonClient.getBucket(key.toString()).delete());
    }

    // 获取缓存统计信息
    //这里的统计信息只来自Caffeine本地缓存，因为Redisson不提供原生的缓存统计信息
    public CacheStats stats() {
        return localCache.stats();
    }

    // 关闭方法需要关闭写回策略的线程池
    public void shutdown() {
        writeBehindExecutor.shutdown();
        try {
            if (!writeBehindExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
                writeBehindExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            writeBehindExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

重要功能：

getAll：批量从缓存获取数据。首先从本地缓存获取，如果本地缓存中缺失，则从Redis中获取，并回填到本地缓存。
putAll：批量写入数据到本地和Redis缓存。
invalidateAll：批量从本地和Redis缓存中移除数据。
stats：获取Caffeine缓存的统计信息。

Note:

添加了一个ConcurrentHashMap来存储锁对象，并在获取数据时使用了一个双重检查锁定模式。当本地缓存中没有数据时，首先获取一个锁，然后再次检查本地缓存以确保数据在获取锁的过程中没有被其他线程填充。如果本地缓存仍然没有数据，会从Redis获取数据，如果Redis也没有数据，则从数据源加载数据并更新Redis和本地缓存。这样的策略可以减少缓存击穿的风险。

请注意，这种锁的使用会增加系统的复杂性，并可能导致性能开销，特别是在高并发场景下。因此，在实现这种机制时，需要仔细衡量其潜在的性能影响。

加了一个单线程的 ExecutorService 用于处理写回策略。当本地缓存中的条目因为驱逐策略被移除时，会将这个条目异步地写入 Redis。还添加了一个 shutdown 方法来关闭线程池。

对于缓存穿透保护，可以在 get 方法中加入逻辑来返回空对象或者使用布隆过滤器来预先检查键是否可能存在，一个特殊的空对象 NULL_PLACEHOLDER 存储到本地缓存和 Redis 中，这样下次查询相同的键时就能直接从缓存中获取到空对象，从而防止缓存穿透。

测试类：

使用示例：

java复制代码
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;
import java.util.function.Function;
/**
 * @Author derek_smart
 * @Date 202/4/24 15:25
 * @Description caffeine和redis 缓存组合测试类
 */
public class HybridCacheExample {

    public static void main(String[] args) {
        // 配置 Redisson
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient = Redisson.create(config);
        // 创建混合缓存实例
        HybridCache<String, String> hybridCache = new HybridCache<>(
                redissonClient,
                1000, // 本地缓存最大条目数
                10, // 本地缓存过期时间
                TimeUnit.MINUTES, // 本地缓存时间单位
                60 * 60 // Redis 缓存过期时间（秒）
        );

        // 模拟数据加载函数
        Function<String, String> dataLoader = key -> {
            // 模拟数据库或其他数据源的加载过程
            System.out.println("Loading data for key: " + key);
            return "Value for " + key;
        };

        // 尝试获取缓存数据，如果没有，则使用提供的函数从数据源加载
        String key = "key1";
        String value = hybridCache.get(key, dataLoader);
        System.out.println("Value: " + value);

        // 更新缓存数据
        String newValue = "Updated value";
        hybridCache.put(key, newValue);

        // 从缓存获取更新后的数据
        String updatedValue = hybridCache.get(key);
        System.out.println("Updated Value: " + updatedValue);

        // 刷新缓存数据
        hybridCache.refresh(key, dataLoader);

        // 使缓存的数据失效
        hybridCache.invalidate(key);

        // 关闭 Redisson 客户端和混合缓存
        hybridCache.shutdown();
        redissonClient.shutdown();
    }
}

企业微信截图_17139484338926.png

测试说明：

在这个示例中，首先配置了 Redisson 客户端并连接到本地运行的 Redis 服务器。然后，创建了一个 HybridCache 实例，设置了本地缓存的大小和过期时间，以及 Redis 缓存的过期时间。

定义了一个 dataLoader 函数，它模拟了从数据库或其他数据源加载数据的过程。接着，使用 hybridCache.get 方法尝试从缓存中获取数据。如果本地缓存和 Redis 缓存中都没有找到数据，将调用 dataLoader 函数加载数据并存入缓存。

然后，更新了缓存中的数据，并再次从缓存中获取更新后的数据。之后，使用 refresh 方法来手动刷新缓存中的数据。最后，使用 invalidate 方法使缓存中的数据失效，并关闭 Redisson 客户端和混合缓存。

Note：

请注意，在实际应用中，可能需要根据实际业务逻辑和数据源来实现数据加载函数。此外，还需要确保 Redis 服务器正在运行且可访问。

优化点1：关于老铁提出问题再次优化

在使用 Caffeine 缓存时，确实可能需要设置总大小上限或键的个数上限，并且有时候需要为每个键设置不同的过期时间。

以下是如何在配置 Caffeine 缓存时实现这两个优化：

1. 设置 Caffeine 缓存的总大小上限或键的个数上限

Caffeine 提供了两种方式来限制缓存的大小：基于权重的限制和基于最大条目数的限制。权重可以通过实现 Weigher 接口来定义，这里是一个基于键和值的大小设置权重的示例：

java复制代码Caffeine.newBuilder()
    .maximumWeight(10000) // 设置缓存的最大权重
    .weigher(new Weigher<KeyType, ValueType>() {
        @Override
        public int weigh(KeyType key, ValueType value) {
            // 自定义计算权重的逻辑
            // 例如，我们可以假设每个缓存项的权重为1
            return 1;
        }
    })
    // 其他配置...
    .build();

如果只需要限制缓存的最大条目数，可以使用 maximumSize 方法：

java复制代码Caffeine.newBuilder()
    .maximumSize(1000) // 设置缓存的最大条目数
    // 其他配置...
    .build();

2. 为每个键设置不同的过期时间

Caffeine 允许你通过实现 Expiry 接口来为每个键定义不同的过期策略。

下面是一个示例，展示如何为不同的键设置不同的过期时间：

java复制代码Caffeine.newBuilder()
    .expireAfter(new Expiry<KeyType, ValueType>() {
        @Override
        public long expireAfterCreate(KeyType key, ValueType value, long currentTime) {
            // 创建后的过期时间逻辑
            // 例如，根据键的特定属性决定过期时间
            return TimeUnit.MINUTES.toNanos(getCustomExpiryForKey(key));
        }

        @Override
        public long expireAfterUpdate(KeyType key, ValueType value, long currentTime, long currentDuration) {
            // 更新后的过期时间逻辑
            // 可以选择保留当前的过期时间或者重新计算
            return currentDuration;
        }

        @Override
        public long expireAfterRead(KeyType key, ValueType value, long currentTime, long currentDuration) {
            // 读取后的过期时间逻辑
            // 可以选择保留当前的过期时间或者重新计算
            return currentDuration;
        }
    })
    // 其他配置...
    .build();

在这个 Expiry 实现中，expireAfterCreate 方法定义了每个键在创建时的过期时间。expireAfterUpdate 和 expireAfterRead 方法允许你在键被更新或读取后调整它们的过期时间

优化点2 分布式环境本地缓存和Redis一致性问题：

在分布式环境中，确保本地缓存（如 Caffeine）与分布式缓存（如 Redis）之间的一致性是一个常见的挑战。
为了减少中间件的引入并使用 Redis 自身的消息推送功能，我们可以依赖于 Redis 的发布/订阅机制来通知各个应用实例缓存的变动。

解决方案:

本地缓存一致性: 当一个应用实例更新了 Redis 中的一个键值对时，它可以通过 Redis 的发布/订阅系统发布一个消息。其他应用实例订阅了这个消息，可以据此来清除或更新本地 Caffeine 缓存中相应的键。
键值变更处理: 如果一个键的值在 Redis 中发生变更，我们不直接在本地缓存中更新这个值，而是删除本地缓存中的这个键。下次访问这个键时，如果本地缓存中没有找到，就会从 Redis 中加载并重新放入 Caffeine 缓存。

实现步骤:

发布消息: 当一个键在 Redis 中被更新或删除时，发布一个消息到特定的频道。

1
2
3

java复制代码public void publishKeyInvalidation(String key) {
    redissonClient.getTopic("cache-invalidation-channel").publish(key);
}

订阅消息: 在应用启动时，订阅 Redis 频道，监听键失效的消息。

java复制代码public void subscribeToCacheInvalidationChannel() {
    RTopic topic = redissonClient.getTopic("cache-invalidation-channel");
    topic.addListener(String.class, (channel, key) -> {
        // 当接收到消息时，从本地缓存中移除相应的键
        caffeineCache.invalidate(key);
    });
}

处理键值变更: 当需要更新键值时，先在 Redis 中进行更新，然后发布消息。

java复制代码public void updateValueInCache(String key, ValueType value) {
    // 更新 Redis 中的值
    redissonClient.getMap("distributed-cache").fastPut(key, value);
    // 发布失效消息
    publishKeyInvalidation(key);
}

处理本地缓存: 当本地缓存尝试访问一个键时，如果缓存中没有，则从 Redis 加载。

java复制代码public ValueType getValueFromCache(String key) {
    // 尝试从本地缓存获取值
    ValueType value = caffeineCache.getIfPresent(key);
    if (value == null) {
        // 本地缓存未命中，从 Redis 加载
        value = (ValueType) redissonClient.getMap("distributed-cache").get(key);
        if (value != null) {
            // 将加载的值放入本地缓存
            caffeineCache.put(key, value);
        }
    }
    return value;
}

注意事项:

这种方案假设 Redis 的读写操作比本地缓存的操作要慢，因此在 Redis 缓存中更新数据后，我们只是删除本地缓存中的数据，而不是更新它。
在高并发环境中，可能会出现短暂的不一致，因为消息传递和处理需要时间。
为了避免不必要的 Redis 访问，可以在本地缓存中设置一个短暂的过期时间，这样即使不立即从 Redis 加载数据，本地缓存中的数据也会很快过期并被刷新。
通过以上步骤，可以构建一个既利用了本地缓存速度优势，又保持了与分布式缓存一致性的混合缓存策略。

优化点3：如果每个KEY的VALUE非常大，该如何处理

当每个键的值非常大时，仅仅通过设置 maximumSize 来限制键的数量可能是不够的，因为这样做不能有效控制占用的总内存大小。
在这种情况下，需要考虑整体的内存使用情况，并使用基于权重的缓存大小限制，
这可以通过 Caffeine 的 maximumWeight 和 weigher 配置来实现。这里是如何使用 maximumWeight 和 weigher 来限制缓存的总内存使用的：

java复制代码Caffeine.newBuilder()
    .maximumWeight(estimatedMaxMemory)
    .weigher(new Weigher<KeyType, ValueType>() {
        @Override
        public int weigh(KeyType key, ValueType value) {
            // 估算每个缓存项的内存占用大小，例如，可以通过某种方式估算对象的大小
            int valueSize = estimateSizeOf(value);
            // 如果你有键和值的大小，可以将它们加起来
            int keySize = estimateSizeOf(key);
            return keySize + valueSize;
        }
    })
    // 可以结合使用expireAfterWrite或expireAfterAccess来设置过期策略
    .expireAfterWrite(10, TimeUnit.MINUTES)
    // 其他配置...
    .build();

// 估算对象大小的方法示例
public int estimateSizeOf(Object object) {
    // 实现对象大小估算逻辑，可能需要使用一些启发式方法或者工具库
    // 例如，使用java.lang.instrument.Instrumentation来获取对象大小
    // 或者使用某些估算规则，比如一个字符串大约占用其长度 * 2字节（因为Java中的char是2字节）
}

使用 maximumWeight 和 weigher 的组合，你可以更好地控制缓存的内存占用，而不仅仅是缓存项的数量。这样可以保证即使

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有

概述：