Caffeine本地緩存示例詳解

更新時間：2023年07月14日 14:37:19 作者：2021不再有雨

Caffeine是一種高性能的緩存庫，是基于Java 8的最佳（最優(yōu)）緩存框架，這篇文章主要介紹了Caffeine本地緩存相關(guān)知識,需要的朋友可以參考下

一. 概述

Caffeine是一種高性能的緩存庫，是基于Java 8的最佳（最優(yōu)）緩存框架。

基于Google的Guava Cache，Caffeine提供一個性能卓越的本地緩存(local cache) 實現(xiàn), 也是SpringBoot內(nèi)置的本地緩存實現(xiàn)。(Caffeine性能是Guava Cache的6倍)

Caffeine提供了靈活的結(jié)構(gòu)來創(chuàng)建緩存，并且有以下特性：

自動加載條目到緩存中，可選異步方式
可以基于大小剔除
可以設(shè)置過期時間，時間可以從上次訪問或上次寫入開始計算
異步刷新
keys自動包裝在弱引用中
values自動包裝在弱引用或軟引用中
條目剔除通知
緩存訪問統(tǒng)計

二. 數(shù)據(jù)加載

Caffeine提供以下四種類型的加載策略：

1. Manual手動

public static void demo(){
		Cache<String,String> cache = Caffeine.newBuilder()
			.expireAfterWrite(20, TimeUnit.SECONDS)
			.maximumSize(5000)
			.build();
		// 1.Insert or update an entry
		cache.put("hello","world");
		// 2. Lookup an entry, or null if not found
		String val1 = cache.getIfPresent("hello");
		// 3. Lookup and compute an entry if absent, or null if not computable
		cache.get("msg", k -> createExpensiveGraph(k));
		// 4. Remove an entry
		cache.invalidate("hello");
	}
	private static String createExpensiveGraph(String key){
		System.out.println("begin to query db..."+Thread.currentThread().getName());
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
		}
		System.out.println("success to query db...");
		return UUID.randomUUID().toString();
	}

Cache接口可以顯式地控制檢索、更新和刪除Entry

2. Loading自動

private static void demo() {
		LoadingCache<String, String> cache = Caffeine.newBuilder()
			.expireAfterWrite(5, TimeUnit.SECONDS)
			.maximumSize(500)
			.build(new CacheLoader<String, String>() {
				@Override
				public String load(String key) throws Exception {
					return createExpensiveGraph(key);
				}
				@Override
				public Map<String, String>  loadAll(Iterable<? extends String> keys) {
					System.out.println("build keys");
					Map<String,String> map = new HashMap<>();
					for(String k : keys){
						map.put(k,k+"-val");
					}
					return map;
				}
			});
		String val1 = cache.get("hello");
		Map<String,String> values = cache.getAll(Lists.newArrayList("key1", "key2"));
	}
	private static String createExpensiveGraph(String key){
		System.out.println("begin to query db..."+Thread.currentThread().getName());
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
		}
		System.out.println("success to query db...");
		return UUID.randomUUID().toString();
	}

LoadingCache通過關(guān)聯(lián)一個CacheLoader來構(gòu)建Cache, 當(dāng)緩存未命中會調(diào)用CacheLoader的load方法生成V

還可以通過LoadingCache的getAll方法批量查詢, 當(dāng)CacheLoader未實現(xiàn)loadAll方法時, 會批量調(diào)用load方法聚合會返回.

當(dāng)CacheLoader實現(xiàn)loadAll方法時, 則直接調(diào)用loadAll返回.

public interface CacheLoader<K, V>{
    V load(@NonNull K var1) throws Exception;
    Map<K, V> loadAll(@NonNull Iterable<? extends K> keys);
}

3. Asynchronous Manual異步手動

private static void demo() throws ExecutionException, InterruptedException {
		AsyncCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.expireAfterWrite(10, TimeUnit.SECONDS)
			.buildAsync();
		// Lookup and asynchronously compute an entry if absent
		CompletableFuture<String> future = cache.get("hello", k -> createExpensiveGraph(k));
		System.out.println(future.get());
	}
	private static String createExpensiveGraph(String key){
		System.out.println("begin to query db..."+Thread.currentThread().getName());
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
		}
		System.out.println("success to query db...");
		return UUID.randomUUID().toString();
	}

AsyncCache是另一種Cache，它基于Executor計算Entry，并返回一個CompletableFuture

和Cache的區(qū)別是, AsyncCache計算Entry的線程是ForkJoinPool線程池. 手動Cache緩存是調(diào)用線程進行計算

4. Asynchronously Loading異步自動

public static void demo() throws ExecutionException, InterruptedException {
		AsyncLoadingCache<String,String> cache = Caffeine.newBuilder()
			.expireAfterWrite(10, TimeUnit.SECONDS)
			.maximumSize(500)
			.buildAsync(k -> createExpensiveGraph(k));
		CompletableFuture<String> future = cache.get("hello");
		System.out.println(future.get());
	}
	private static String createExpensiveGraph(String key){
		System.out.println("begin to query db..."+Thread.currentThread().getName());
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
		}
		System.out.println("success to query db...");
		return UUID.randomUUID().toString();
	}

AsyncLoadingCache 是關(guān)聯(lián)了 AsyncCacheLoader 的 AsyncCache

三. 數(shù)據(jù)驅(qū)逐

Caffeine提供以下幾種剔除方式：基于大小、基于權(quán)重、基于時間、基于引用

1. 基于容量

又包含兩種, 基于size和基于weight權(quán)重

基于size

LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.recordStats()
			.build( k -> UUID.randomUUID().toString());
		for (int i = 0; i < 600; i++) {
			cache.get(String.valueOf(i));
			if(i> 500){
				CacheStats stats = cache.stats();
				System.out.println("evictionCount:"+stats.evictionCount());
				System.out.println("stats:"+stats.toString());
			}
		}

如果緩存的條目數(shù)量不應(yīng)該超過某個值，那么可以使用Caffeine.maximumSize(long)。如果超過這個值，則會剔除很久沒有被訪問過或者不經(jīng)常使用的那個條目。

上述測試并不是i=500時, 而是稍微延遲于i的增加, 說明驅(qū)逐是另外一個線程異步進行的

基于權(quán)重

LoadingCache<Integer,String> cache = Caffeine.newBuilder()
			.maximumWeight(300)
			.recordStats()
			.weigher((Weigher<Integer, String>) (key, value) -> {
				if(key % 2 == 0){
					return 2;
				}
				return 1;
			})
			.build( k -> UUID.randomUUID().toString());
		for (int i = 0; i < 300; i++) {
			cache.get(i);
			if(i> 200){
				System.out.println(cache.stats().toString());
			}
		}

如果，不同的條目有不同的權(quán)重值的話(不同的實例占用空間大小不一樣)，那么你可以用Caffeine.weigher(Weigher)來指定一個權(quán)重函數(shù)，并且使用Caffeine.maximumWeight(long)來設(shè)定最大的權(quán)重值。

上述測試并不是i=200時, 而是稍微延遲于i的增加, 說明驅(qū)逐是另外一個線程異步進行的

簡單的來說，要么限制緩存條目的數(shù)量，要么限制緩存條目的權(quán)重值，二者取其一。

2. 基于時間

基于時間又分為四種: expireAfterAccess、expireAfterWrite、refreshAfterWrite、expireAfter

expireAfterAccess

超時未訪問則失效: 訪問包括讀和寫

private static LoadingCache<String,String> cache = Caffeine.newBuilder()
		.expireAfterAccess(1, TimeUnit.SECONDS)
		.build(key -> UUID.randomUUID().toString());

特征:

訪問包括讀和寫入
數(shù)據(jù)失效后不會主動重新加載, 必須依賴下一次訪問. (言外之意: 失效和回源是兩個動作)
key超時失效或不存在，若多個線程并發(fā)訪問, 只有1個線程回源數(shù)據(jù)，其他線程阻塞等待數(shù)據(jù)返回
對同一數(shù)據(jù)一直訪問, 且間隔小于失效時間, 則不會去load數(shù)據(jù), 一直讀到的是臟數(shù)據(jù)

expireAfterWrite

寫后超時失效

private static LoadingCache<String,String> cache = Caffeine.newBuilder()
		.expireAfterWrite(1, TimeUnit.SECONDS)
		.build(key -> UUID.randomUUID().toString());

特征:

數(shù)據(jù)失效后不會主動重新加載, 必須依賴下一次訪問. (言外之意: 失效和回源是兩個動作)

key超時失效或不存在，若多個線程并發(fā)訪問, 只有1個線程回源數(shù)據(jù)，其他線程阻塞等待數(shù)據(jù)返回

expire后來訪問一定能保證拿到最新的數(shù)據(jù)

refreshAfterWrite

private static LoadingCache<String,String> cache = Caffeine.newBuilder()
		.refreshAfterWrite(1, TimeUnit.SECONDS)
		.build(key -> UUID.randomUUID().toString());

和expireAfterWrite類似基于寫后超時驅(qū)逐, 區(qū)別是重新load的操作不一樣.

特征:

數(shù)據(jù)失效后不會主動重新加載, 必須依賴下一次訪問. (言外之意: 失效和回源是兩個動作)
當(dāng)cache命中未命中時, 若多個線程并發(fā)訪問時, 只有1個線程回源數(shù)據(jù)，其他線程阻塞等待數(shù)據(jù)返回
當(dāng)cache命中失效數(shù)據(jù)時, 若多個線程并發(fā)訪問時, 第一個訪問的線程提交一個load數(shù)據(jù)的任務(wù)到公共線程池，然后和所有其他訪問線程一樣直接返回舊值

實際通過LoadingCache.refresh(K)進行異步刷新, 如果想覆蓋默認的刷新行為, 可以實現(xiàn)CacheLoader.reload(K, V)方法

expireAfter

比較少用

public static void demo(){
		MyTicker ticker = new MyTicker();
		LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.ticker(ticker)
			//此時的效果為expireAfterWrite(5,TimeUnit.SECONDS)
			.expireAfter(new Expiry<String, String>() {
				//1.如果寫入key時是第一次創(chuàng)建，則調(diào)用該方法返回key剩余的超時時間, 單位納秒ns
				//currentTime為當(dāng)前put時Ticket的時間，單位ns
				@Override
				public long expireAfterCreate(String key,String value, long currentTime) {
					System.out.println("write first currentTime:"+currentTime/1_000_000_000L);
					return 5_000_000_000L;//5s
				}
				//2.如果寫入key時已經(jīng)存在即更新key時，則調(diào)用該方法返回key剩余的超時時間, 單位納秒ns
				//currentTime為當(dāng)前put時Ticket的時間，單位ns，durationTime為舊值(上次設(shè)置)剩余的存活時間，單位是ns
				@Override
				public long expireAfterUpdate(String key,String value, long currentTime,long durationTime) {
					System.out.println("update currentTime:"+currentTime/1_000_000_000L+",leftTime:"+durationTime/1_000_000_000L);
					return 5_000_000_000L;//5s
				}
				//3.如果key被訪問時,則調(diào)用該方法返回key剩余的超時時間, 單位納秒ns
				//currentTime為read時Ticket的時間，單位ns，durationTime為舊值(上次設(shè)置)剩余的存活時間，單位是ns
				@Override
				public long expireAfterRead(String key,String value, long currentTime,long durationTime) {
					System.out.println("read currentTime:"+currentTime/1_000_000_000L+",leftTime:"+durationTime/1_000_000_000L);
					return durationTime;
				}
			})
			.build(k ->  UUID.randomUUID().toString());
		cache.get("key1");//觸發(fā)expireAfterCreate
		ticker.advance(1, TimeUnit.SECONDS);//模擬時間消逝
		cache.get("key1");//觸發(fā)expireAfterRead,剩余生存時間4s
		ticker.advance(2, TimeUnit.SECONDS);//模擬時間消逝
		cache.put("key1","value1");//觸發(fā)expireAfterUpdate,重置生存時間為5s
		ticker.advance(3, TimeUnit.SECONDS);//模擬時間消逝
		cache.get("key1");//觸發(fā)expireAfterCreate,剩余生存時間為2s
	}
public class MyTicker implements Ticker {
	private final AtomicLong nanos = new AtomicLong();
	//模擬時間消逝
	public void advance(long time, TimeUnit unit) {
		this.nanos.getAndAdd(unit.toNanos(time));
	}
	@Override
	public long read() {
		return this.nanos.get();
	}
}

上述實現(xiàn)了Expiry接口, 分別重寫了expireAfterCreate、expireAfterUpdate、expireAfterRead方法, 當(dāng)?shù)谝淮螌懭霑r、更新時、讀訪問時會分別調(diào)用這三個方法有機會重新設(shè)置剩余的失效時間, 上述案例模擬了expireAfterWrite(5,TimeUnit.SECONDS)的效果.

注意點:

以上基于時間驅(qū)逐, 數(shù)據(jù)超時失效和回源是兩個動作, 必須依賴下一次訪問. 為了避免服務(wù)啟動時大量緩存穿透, 可以通過提前項目啟動時手動預(yù)熱
一般expireAfterWrite和refreshAfterWrite結(jié)合使用, expire的時間t1大于refresh的時間t2, 在t2~t1內(nèi)數(shù)據(jù)更新允許臟數(shù)據(jù), t1之后必須要重新同步加載新數(shù)據(jù)

3. 基于弱/軟引用

/**
	 * 允許GC時回收keys或values
	 */
	public static void demo(){
		LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.expireAfterWrite(10, TimeUnit.SECONDS)
			.weakKeys()
			.weakValues()
			.build(k -> UUID.randomUUID().toString());
	}

Caffeine.weakKeys() 使用弱引用存儲key。如果沒有強引用這個key，則GC時允許回收該條目

Caffeine.weakValues() 使用弱引用存儲value。如果沒有強引用這個value，則GC時允許回收該條目

Caffeine.softValues() 使用軟引用存儲value, 如果沒有強引用這個value，則GC內(nèi)存不足時允許回收該條目

public static void demo(){
		/**
		 * 使用軟引用存儲value,GC內(nèi)存不夠時會回收
		 */
		LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.expireAfterWrite(10, TimeUnit.SECONDS)
			.softValues()//注意沒有softKeys方法
			.build(k -> UUID.randomUUID().toString());
	}

Java4種引用的級別由高到低依次為：強引用 > 軟引用 > 弱引用 > 虛引用

引用類型	被垃圾回收時間	用途	生存時間
強引用	從來不會	對象的一般狀態(tài)	JVM停止運行時終止
軟引用	在內(nèi)存不足時	對象緩存	內(nèi)存不足時終止
弱引用	在垃圾回收時	對象緩存	gc運行后終止
虛引用	Unknown	Unknown	Unknown

四. 驅(qū)逐監(jiān)聽

eviction 指受策略影響而被刪除
invalidation 值被調(diào)用者手動刪除
removal 值因eviction或invalidation而發(fā)生的一種行為

1. 手動觸發(fā)刪除

// individual key
cache.invalidate(key)
// bulk keys
cache.invalidateAll(keys)
// all keys
cache.invalidateAll()

2. 被驅(qū)逐的原因

EXPLICIT：如果原因是這個，那么意味著數(shù)據(jù)被我們手動的remove掉了
REPLACED：就是替換了，也就是put數(shù)據(jù)的時候舊的數(shù)據(jù)被覆蓋導(dǎo)致的移除
COLLECTED：這個有歧義點，其實就是收集，也就是垃圾回收導(dǎo)致的，一般是用弱引用或者軟引用會導(dǎo)致這個情況
EXPIRED：數(shù)據(jù)過期，無需解釋的原因。
SIZE：個數(shù)超過限制導(dǎo)致的移除

3. 監(jiān)聽器

public static void demo(){
		LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(5)
			.recordStats()
			.expireAfterWrite(2, TimeUnit.SECONDS)
			.removalListener((String key, String value, RemovalCause cause) -> {
				System.out.printf("Key %s was removed (%s)%n", key, cause);
			})
			.build(key -> UUID.randomUUID().toString());
		for (int i = 0; i < 15; i++) {
			cache.get(i+"");
			try {
				Thread.sleep(200);
			} catch (InterruptedException e) {
			}
		}
		//因為evict是異步線程去執(zhí)行，為了看到效果稍微停頓一下
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
		}
	}

日志打印如下:

Key 0 was removed (SIZE)
Key 1 was removed (SIZE)
Key 6 was removed (SIZE)
Key 7 was removed (SIZE)
Key 8 was removed (SIZE)
Key 9 was removed (SIZE)
Key 10 was removed (SIZE)
Key 2 was removed (EXPIRED)
Key 3 was removed (EXPIRED)
Key 4 was removed (EXPIRED)

五. 統(tǒng)計

public static void demo(){
		LoadingCache<Integer,String> cache = Caffeine.newBuilder()
			.maximumSize(10)
			.expireAfterWrite(10, TimeUnit.SECONDS)
			.recordStats()
			.build(key -> {
				if(key % 6 == 0 ){
					return null;
				}
				return  UUID.randomUUID().toString();
			});
		for (int i = 0; i < 20; i++) {
			cache.get(i);
			printStats(cache.stats());
		}
		for (int i = 0; i < 10; i++) {
			cache.get(i);
			printStats(cache.stats());
		}
	}
	private static void printStats(CacheStats stats){
		System.out.println("---------------------");
		System.out.println("stats.hitCount():"+stats.hitCount());//命中次數(shù)
		System.out.println("stats.hitRate():"+stats.hitRate());//緩存命中率
		System.out.println("stats.missCount():"+stats.missCount());//未命中次數(shù)
		System.out.println("stats.missRate():"+stats.missRate());//未命中率
		System.out.println("stats.loadSuccessCount():"+stats.loadSuccessCount());//加載成功的次數(shù)
		System.out.println("stats.loadFailureCount():"+stats.loadFailureCount());//加載失敗的次數(shù),返回null
		System.out.println("stats.loadFailureRate():"+stats.loadFailureRate());//加載失敗的百分比
		System.out.println("stats.totalLoadTime():"+stats.totalLoadTime());//總加載時間,單位ns
		System.out.println("stats.evictionCount():"+stats.evictionCount());//驅(qū)逐次數(shù)
		System.out.println("stats.evictionWeight():"+stats.evictionWeight());//驅(qū)逐的weight值總和
		System.out.println("stats.requestCount():"+stats.requestCount());//請求次數(shù)
		System.out.println("stats.averageLoadPenalty():"+stats.averageLoadPenalty());//單次load平均耗時
	}

六. 其他

1. Ticker

時鐘, 方便測試模擬時間流逝

public static void demo(){
		MyTicker ticker = new MyTicker();
		LoadingCache<String,String> cache = Caffeine.newBuilder()
			.maximumSize(500)
			.ticker(ticker)
			.expireAfterWrite(1, TimeUnit.SECONDS)
			.build(k ->  UUID.randomUUID().toString());
		cache.get("key1");//觸發(fā)expireAfterCreate
		ticker.advance(1, TimeUnit.SECONDS);//模擬時間消逝
		cache.get("key1");//觸發(fā)expireAfterRead,剩余生存時間4s
		ticker.advance(2, TimeUnit.SECONDS);//模擬時間消逝
		cache.put("key1","value1");//觸發(fā)expireAfterUpdate,重置生存時間為5s
	}
public class MyTicker implements Ticker {
	private final AtomicLong nanos = new AtomicLong();
	//模擬時間消逝
	public void advance(long time, TimeUnit unit) {
		this.nanos.getAndAdd(unit.toNanos(time));
	}
	@Override
	public long read() {
		return this.nanos.get();
	}
}