Caffeine Cache（本地缓存限流）

pavel随笔 2021-11-05

3403

实际工程中的缓存可分为本地缓存和分布式缓存。分布式缓存有Redis和MongoDB；本地缓存一般以map形式保存在本地内存中，最常用的是Guava cache（Google开源的Java重用工具集库Guava里的一款缓存工具），而这篇文章会提及另一个本地缓存框架：Caffeine Cache，站在Guava Cache的肩膀上优化算法而产生。

Caffine Cache框架介绍
Caffine Cache 在算法上的优点
实际使用：构造本地缓存限流器（附代码）

1. Caffine Cache框架介绍

在数据一致性的文章中我提到过，业务操作会对DB和缓存两者数据同时改动。如：新增数据时，先插入DB，再删除原来的缓存；查询数据时，先查缓存，命中则返回，没有命中时，需要查询DB，再把查询结果放入缓存中。如果访问量大，必须考虑缓存的线程安全问题和回收策略。

为了较好解决上述的问题，Guava Cache框架继承了ConcurrentHashMap的思路，使用多个segments方式的细粒度锁，封装了get和put操作；提供线程安全的缓存操作；提供过期策略；提供回收策略；缓存监控。当缓存的数据超过最大值时，使用LRU算法替换。Caffine Cache 则更强，在保留了Guava各种优点的基础上优化了缓存淘汰算法——W-TinyLFU算法，提供了近乎最佳的命中率。具体比较如下：

可以看到Guava Cache完败，所以还是和我一起用Caffine吧

官方数据地址：https://github.com/ben-manes/caffeine/wiki/Benchmarks-zh-CN，有兴趣的可以自己去查。

Caffeine提供了灵活的构造器去创建一个拥有下列特性的缓存：

自动加载元素到缓存当中，异步加载的方式也可供选择
当达到最大容量的时候可以使用基于就近度和频率的算法进行基于容量的驱逐
将根据缓存中的元素上一次访问或者被修改的时间进行基于过期时间的驱逐
当向缓存中一个已经过时的元素进行访问的时候将会进行异步刷新
key将自动被弱引用所封装
value将自动被弱引用或者软引用所封装
驱逐(或移除)缓存中的元素时将会进行通知
写入传播到一个外部数据源当中
持续计算缓存的访问统计指标

2. Caffine Cache 在算法上的优点

常见的的缓存淘汰算法：FIFO，LRU，LFU（面试必手撕代码）

FIFO：先进先出，在这种淘汰算法中，先进入缓存的会先被淘汰，会导致命中率很低。
LRU：最近最少使用算法，每次访问数据都会将其放在我们的队尾，如果需要淘汰数据，就只需要淘汰队首即可。存在局限：如果有个数据在 1 分钟访问了 1000次，再后 1 分钟没有访问这个数据，但是有其他的数据访问，就导致了我们这个热点数据被淘汰。
LFU：最近最少频率使用，记录每个数据的使用频率，然后选出频率最低进行淘汰。这样就避免了 LRU 不能处理时间段的问题。也存在局限：比如有部新剧出来了，我们使用 LFU 给他缓存下来，这部新剧在这几天大概访问了几亿次，这个访问频率也在我们的 LFU 中记录了几亿次。但是新剧总会过气的，比如一个月之后这个新剧的前几集其实已经过气了，但是他的访问量的确是太高了，其他的电视剧根本无法淘汰这个新剧，所以在这种模式下是有局限性。

Caffeine的缓存淘汰算法——Window TinyLfu算法：

W-TinyLfu通过LRU将小范围窗口中的数据数据驱逐到更大的范围中进行Segmented LRU进行驱逐。TinyLfu依赖一个频率sketch 来预估元素的历史访问频率。小范围的数据窗口的设计使该策略在新数据大量加入的时候具有较高的命中率。小范围的时间窗口大小和大范围的主要空间大小将根据hill climbing算法进行动态优化。这允许缓存能以较低的开销估计元素的访问频率和就近度。该算法提供了近乎最佳的命中率。同时实现相对简单，并不会要求非常驻元素的存在，并且要求更低的内存开销。具体比较地址：

https://github.com/ben-manes/caffeine/wiki/Efficiency-zh-CN。

项目常用本地缓存的比较：

3. 实际使用：构造本地缓存限流器

引入依赖

<dependency>
    <groupId>com.github.ben-manes.caffeine</groupId>
    <artifactId>caffeine</artifactId>
    <version>2.3.5</version>
</dependency>

限流器配置参数

public class LocalRateLimiterConfig {


    //缓存key数量最大值及key的过期时间
    private int size = 10000;
    private int expireAfterAccess = 6000;


    //二级Map的key数量最大值及key的过期时间
    private int sizePerkey = 10;
    private int expireAfterAccessPerkey = 1000;


    //全局限流数值及单个key的限流数值
    private int globalPermitPerSecond = 1;
    private Map<Object, Integer> permitsMap;


    //限流次数报警值
    private int warnGlobalPermitPerSecond = 1;


    //开关：off关闭，其它均为打开。
    private String limitSwitch = "open";


}

使用Caffein构建限流器

public class LocalRatetLimiter<K> {


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


    private LocalRateLimiterConfig limitConfig = new LocalRateLimiterConfig();


    private LoadingCache<K, LoadingCache<Long, AtomicInteger>> limitStore;


    private void init() {
        limitStore =
                Caffeine.newBuilder().maximumSize(limitConfig.getSize()).expireAfterWrite(limitConfig.getExpireAfterAccess(), TimeUnit.SECONDS)
                        .build(key -> {
                            return Caffeine.newBuilder().maximumSize(limitConfig.getSizePerkey())
                                    .expireAfterWrite(limitConfig.getExpireAfterAccessPerkey(), TimeUnit.SECONDS).build(key0 -> {
                                        return new AtomicInteger(0);
                                    });
                        });
    }




    public boolean limit(K k) {
        try {
            if ("off".equals(limitConfig.getLimitSwitch())) {
                return false;
            }
            int threshold = limitConfig.getGlobalPermitPerSecond();
            Map<Object, Integer> permitsPerSecondMap = limitConfig.getPermitsMap();
            if (permitsPerSecondMap != null) {
                Integer value = permitsPerSecondMap.get(k);
                if (value != null && value.intValue() != 0) {
                    threshold = value.intValue();
                }
            }
            int thresholdOfWarn = limitConfig.getWarnGlobalPermitPerSecond();
            // 固定窗口的令牌桶限流
            int currentValue = limitStore.get(k).get(System.currentTimeMillis() / 1000).incrementAndGet();


            if (currentValue > thresholdOfWarn && currentValue < threshold) {
                logger.error("请求key已经超过阀值请关注。key = " + k);
                return false;
            }


            boolean ret = currentValue > threshold;
            if (ret) {
                logger.error("请求key已经被拦截请关注。key = " + k);
            }
            return ret;
        } catch (Exception e) {
        }
        return false;
    }


    private static void sleepUninterruptibly(long sleepFor, TimeUnit unit) {
        boolean interrupted = false;
        try {
            long remainingNanos = unit.toNanos(sleepFor);
            long end = System.nanoTime() + remainingNanos;
            while (true) {
                try {
                    // TimeUnit.sleep() treats negative timeouts just like zero.
                    NANOSECONDS.sleep(remainingNanos);
                    return;
                } catch (InterruptedException e) {
                    interrupted = true;
                    remainingNanos = end - System.nanoTime();
                }
            }
        } finally {
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        }
    }


    public void change(LocalRateLimiterConfig config) {
        if (limitStore != null) {
            int size = config.getSize();
            if (size > 0) {
                limitStore.policy().eviction().ifPresent(eviction -> {
                    eviction.setMaximum(size);
                });
            }
            int expireAfterAccess = config.getExpireAfterAccess();
            if (expireAfterAccess > 0) {
                limitStore.policy().expireAfterAccess().ifPresent(consumer -> {
                    consumer.setExpiresAfter(expireAfterAccess, TimeUnit.SECONDS);
                });
            }
        }
    }


    public  boolean isLimit(String key) {
        if (limitStore == null) {
            synchronized (LocalRatetLimiter.class) {
                if (limitStore == null) {
                   init();
                }
            }
        }
        return limit((K) key);
    }




}

使用

if(new LocalRatetLimiter().isLimit(key)){
    return this.failed("request too many ！");
}

数据库

文章转载自pavel随笔，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Caffeine Cache（本地缓存限流）

Caffine Cache 在算法上的优点

实际使用：构造本地缓存限流器（附代码）

1. Caffine Cache框架介绍

2. Caffine Cache 在算法上的优点

常见的的缓存淘汰算法：FIFO，LRU，LFU（面试必手撕代码）

Caffeine的缓存淘汰算法——Window TinyLfu算法：

评论