張開濤：應(yīng)用級緩存之緩存使用模式實(shí)踐—
前面已經(jīng)介紹了Java緩存的使用。對于我們來說如果有人總結(jié)一些緩存使用模式/模板的話，我們在使用時直接照著模式寫即可。而實(shí)際確實(shí)已經(jīng)有總結(jié)好的模式，主要分兩大類：Cache-Aside和Cache-As-SoR(Read-through、Write-through、Write-behind)。
首先，同步兩個名詞。
- SoR(system-of-record)：記錄系統(tǒng)，或者可以叫做數(shù)據(jù)源，即實(shí)際存儲原始數(shù)據(jù)的系統(tǒng)。
- Cache：緩存，是SoR的快照數(shù)據(jù)，Cache的訪問速度比SoR要快，放入Cache的目的是提升訪問速度，減少回源到SoR的次數(shù)。
- 回源：即回到數(shù)據(jù)源頭獲取數(shù)據(jù)，Cache沒有***時，需要從SoR讀取數(shù)據(jù)，這叫做回源。
本文主要以Guava Cache和Ehcache3.x作為實(shí)踐框架來講解。
一、Cache-Aside
Cache-Aside即業(yè)務(wù)代碼圍繞著Cache寫，是由業(yè)務(wù)代碼直接維護(hù)緩存，示例代碼如下所示。
讀場景，先從緩存獲取數(shù)據(jù)，如果沒有***，則回源到SoR并將源數(shù)據(jù)放入緩存供下次讀取使用。
- //1、先從緩存中獲取數(shù)據(jù)
- value = myCache.getIfPresent(key);
- if(value == null) {
- //2.1、如果緩存沒有***，則回源到SoR獲取源數(shù)據(jù)
- value = loadFromSoR(key);
- //2.2、將數(shù)據(jù)放入緩存，下次即可從緩存中獲取數(shù)據(jù)
- myCache.put(key, value);
- }
寫場景，先將數(shù)據(jù)寫入SoR，寫入成功后立即將數(shù)據(jù)同步寫入緩存。
- //1、先將數(shù)據(jù)寫入SoR
- writeToSoR(key,value);
- //2、執(zhí)行成功后立即同步寫入緩存
- myCache.put(key, value);
或者先將數(shù)據(jù)寫入SoR，寫入成功后將緩存數(shù)據(jù)過期，下次讀取時再加載緩存。
- //1、先將數(shù)據(jù)寫入SoR
- writeToSoR(key,value);
- //2、失效緩存，然后下次讀時再加載緩存
- myCache.invalidate(key);
Cache-Aside適合使用AOP模式去實(shí)現(xiàn)，可以參考筆者的博客《Spring Cache抽象詳解》去實(shí)現(xiàn)。
對于Cache-Aside可能存在并發(fā)更新情況，即如果多個應(yīng)用實(shí)例同時更新，那么緩存怎么辦?
● 如果是用戶維度的數(shù)據(jù)(如訂單數(shù)據(jù)、用戶數(shù)據(jù))，則出現(xiàn)這種幾率非常小，因?yàn)椴l(fā)的情況很少，可以不考慮這個問題，加上過期時間來解決即可。
● 對于如商品這種基礎(chǔ)數(shù)據(jù)，可以考慮使用canal訂閱binlog進(jìn)行增量更新分布式緩存，這樣不會存在緩存數(shù)據(jù)不一致的情況，但是，緩存更新會存在延遲。而本地緩存根據(jù)不一致容忍度設(shè)置合理的過期時間。
● 讀服務(wù)場景，可以考慮使用一致性哈希，將相同的操作負(fù)載均衡到同一個實(shí)例，從而減少并發(fā)幾率?；蛘咴O(shè)置比較短的過期時間，可參考“第17章 京東商品詳情頁服務(wù)閉環(huán)實(shí)踐”。
二、Cache-As-SoR
Cache-As-SoR即把Cache看作為SoR，所有操作都是對Cache進(jìn)行，然后Cache再委托給SoR進(jìn)行真實(shí)的讀/寫。即業(yè)務(wù)代碼中只看到Cache的操作，看不到關(guān)于SoR相關(guān)的代碼。有三種實(shí)現(xiàn)：read-through、write-through、write-behind。
1. Read-Through
Read-Through，業(yè)務(wù)代碼首先調(diào)用Cache，如果Cache不***由Cache回源到SoR，而不是業(yè)務(wù)代碼(即由Cache讀SoR)。使用Read-Through模式，需要配置一個CacheLoader組件用來回源到SoR加載源數(shù)據(jù)。Guava Cache和Ehcache 3.x都支持該模式。
Guava Cache實(shí)現(xiàn)
- LoadingCache<Integer,Result<Category>> getCache =
- CacheBuilder.newBuilder()
- .softValues()
- .maximumSize(5000).expireAfterWrite(2, TimeUnit.MINUTES)
- .build(new CacheLoader<Integer,Result<Category>>() {
- @Override
- public Result<Category> load(final Integer sortId) throwsException {
- return categoryService.get(sortId);
- }
- });
在build Cache時，傳入一個CacheLoader用來加載緩存，操作流程如下。
- 應(yīng)用業(yè)務(wù)代碼直接調(diào)用getCache.get(sortId)。
- 首先查詢Cache，如果緩存中有，則直接返回緩存數(shù)據(jù)。
- 如果緩存沒有***，則委托給CacheLoader，CacheLoader會回源到SoR查詢源數(shù)據(jù)(返回值必須不為null，可以包裝為Null對象)，然后寫入緩存。
使用CacheLoader后有幾個好處。
● 應(yīng)用業(yè)務(wù)代碼更簡潔了，不需要像Cache-Aside模式那樣緩存查詢代碼和SoR代碼交織在一起。如果緩存使用邏輯散落在多處，則使用這種方式很簡單的消除了重復(fù)代碼。
● 解決Dog-pile effect，即當(dāng)某個緩存失效時，又有大量相同的請求沒***緩存，從而同時請求到后端，導(dǎo)致后端壓力太大，此時限制一個請求去拿即可。
- if (firstCreateNewEntry) {//***個請求加載緩存的線程去SoR加載源數(shù)據(jù)
- try {
- synchronized (e) {
- returnloadSync(key, hash, loadingValueReference, loader);
- }
- } finally{
- statsCounter.recordMisses(1);
- }
- } else {//其他并發(fā)線程等待“***個線程”加載的數(shù)據(jù)
- return waitForLoadingValue(e, key,valueReference);
- }
- Guava Cache還支持get(K key, Callable<? extends V> valueLoader)方法，傳入一個Callable實(shí)例，當(dāng)緩存沒***時，會調(diào)用Callable#call來查詢SoR加載源數(shù)據(jù)。
- Ehcache 3.x實(shí)現(xiàn)
- CacheManager cacheManager = CacheManagerBuilder. newCacheManagerBuilder(). build(true);
- org.ehcache.Cache<String, String> myCache =cacheManager. createCache ("myCache",
- CacheConfigurationBuilder.newCacheConfigurationBuilder(String.class,String.class,
- ResourcePoolsBuilder.newResourcePoolsBuilder().heap(100,MemoryUnit.MB))
- .withDispatcherConcurrency(4)
- .withExpiry(Expirations.timeToLiveExpiration(Duration.of(10,TimeUnit.SECONDS)))
- .withLoaderWriter(newDefaultCacheLoaderWriter<String, String> () {
- @Override
- public String load(String key) throws Exception {
- return readDB(key);
- }
- @Override
- public Map<String, String> loadAll(Iterable<? extendsString> keys) throws BulkCacheLoadingException, Exception {
- return null;
- }
- }));
Ehcache 3.x使用CacheLoaderWriter來實(shí)現(xiàn)，通過load(K key)和loadAll(Iterable keys)分別來加載單個KEY和批量KEY。Ehcache 3.1沒有自己去解決Dog-pile effect。
2. Write-Through
Write-Through，稱之為穿透寫模式/直寫模式，業(yè)務(wù)代碼首先調(diào)用Cache寫(新增/修改)數(shù)據(jù)，然后由Cache負(fù)責(zé)寫緩存和寫SoR，而不是業(yè)務(wù)代碼。使用Write-Through模式需要配置一個CacheWriter組件用來回寫SoR。Guava Cache沒有提供支持。Ehcache 3.x支持該模式。Ehcache需要配置一個CacheLoaderWriter，CacheLoaderWriter知道如何去寫SoR。當(dāng)Cache需要寫(新增/修改)數(shù)據(jù)時，首先調(diào)用CacheLoaderWriter來同步(立即)到SoR，成功后會更新緩存。
- CacheManager cacheManager = CacheManagerBuilder.newCacheManagerBuilder().build(true);
- org.ehcache.Cache<String, String> myCache =cacheManager.createCache ("myCache",
- CacheConfigurationBuilder.newCacheConfigurationBuilder(String.class,String.class,
- ResourcePoolsBuilder.newResourcePoolsBuilder().heap(100,MemoryUnit.MB))
- .withDispatcherConcurrency(4)
- .withExpiry(Expirations.timeToLiveExpiration(Duration.of(10,TimeUnit.SECONDS)))
- .withLoaderWriter(newDefaultCacheLoaderWriter<String, String> () {
- @Override
- public void write(String key, String value) throws Exception{
- //write
- }
- @Override
- public void writeAll(Iterable<? extends Map.Entry<? extendsString, ? extends String>> entries) throws BulkCacheWritingException,Exception {
- for(Object entry: entries) {
- //batch write
- }
- }
- @Override
- public void delete(Stringkey) throws Exception {
- //delete
- }
- @Override
- public void deleteAll(Iterable<? extends String>keys) throws BulkCacheWritingException, Exception {
- for(Object key :keys) {
- //batch delete
- }
- }
- }).build());
Ehcache 3.x還是使用CacheLoaderWriter來實(shí)現(xiàn)，通過write(String key, String value)、writeAll(Iterable> entries)和delete(String key)、deleteAll(Iterable keys)分別來支持單個寫、批量寫和單個刪除、批量刪除操作。
操作流程如下。
- 當(dāng)我們調(diào)用myCache.put("e","123")或者myCache.putAll(map)時，寫緩存。
- 首先，Cache會將寫操作立即委托給CacheLoaderWriter#write和#writeAll，然后由CacheLoaderWriter負(fù)責(zé)立即去寫SoR。
- 當(dāng)寫SoR成功后，再寫入Cache。
3. Write-Behind
Write-Behind，也叫Write-Back，稱之為回寫模式，不同于Write-Through是同步寫SoR和Cache，Write-Behind是異步寫。異步之后可以實(shí)現(xiàn)批量寫、合并寫、延時和限流。
(1) 異步寫
- CacheManager cacheManager = CacheManagerBuilder. newCacheManagerBuilder()
- .using(PooledExecutionServiceConfigurationBuilder
- .newPooledExecutionServiceConfigurationBuilder()
- .pool("writeBehindPool", 1, 5)
- .build())
- .build(true);
- org.ehcache.Cache<String, String> myCache =cacheManager. createCache ("myCache",
- CacheConfigurationBuilder.newCacheConfigurationBuilder(String.class,String.class,
- ResourcePoolsBuilder.newResourcePoolsBuilder().heap(100,MemoryUnit.MB))
- .withDispatcherConcurrency(4)
- .withExpiry(Expirations.timeToLiveExpiration(Duration.of(10,TimeUnit.SECONDS)))
- .withLoaderWriter(new DefaultCacheLoaderWriter<String,String >() {
- @Override
- public void write(String key, String value) throws Exception{
- //write
- }
- @Override
- public void delete(String key) throws Exception {
- //delete
- }
- })
- .add(WriteBehindConfigurationBuilder
- .newUnBatchedWriteBehindConfiguration()
- .queueSize(5)
- .concurrencyLevel(2
- .useThreadPool("writeBehindPool")
- .build()));
幾個重要配置如下。
- hreadPool：使用PooledExecutionServiceConfigurationBuilder配置線程池;然后WriteBehindConfigurationBuilder通過useThreadPool配置使用哪一個線程池;
- WriteBehindConfigurationBuilder：配置WriteBehind策略;
- CacheLoaderWriter：配置WriteBehind如何操作SoR。
WriteBehindConfigurationBuilder會進(jìn)行如下幾個配置。
- newUnBatchedWriteBehindConfiguration：表示不進(jìn)行批量處理，那么所有批量操作都將會轉(zhuǎn)換成單個操作，即CacheLoaderWriter只需要實(shí)現(xiàn)write和delete即可。
- queueSize(int size)：因?yàn)椴僮魇钱惒交貙慡oR，需要將操作先放入寫操作等待隊(duì)列，因此，使用queue size定義寫操作等待隊(duì)列***大小，即線程池隊(duì)列大小。內(nèi)部使用NonBatchingLocalHeapWriteBehindQueue。
- concurrencyLevel(int concurrency)：配置使用多少個并發(fā)線程和隊(duì)列進(jìn)行WriteBehind。因?yàn)槲覀冎粋魅胍粋€線程池，這是如何實(shí)現(xiàn)該模式的呢?首先看如下代碼片段。
- for (int i = 0; i < writeBehindConcurrency; i++) {
- if (config.getBatchingConfiguration()== null) {
- this.stripes.add(newNonBatchingLocalHeapWriteBehindQueue<K, V>(executionService,defaultThreadPool, config, cacheLoaderWriter));
- } else {
- this.stripes.add(newBatchingLocalHeapWriteBehindQueue<K, V>(executionService, defaultThreadPool,config, cacheLoaderWriter));
- }
- }
可以看到會創(chuàng)建concurrencyLevel個隊(duì)列NonBatchingLocalHeapWriteBehindQueue，其又通過如下代碼片段創(chuàng)建線程池和線程池隊(duì)列。
- this.executorQueue = new LinkedBlockingQueue<Runnable>(config.getMaxQueueSize());
- if (config.getThreadPoolAlias() == null) {
- this.executor= executionService.getOrderedExecutor(defaultThreadPool, executorQueue);
- } else {
- this.executor= executionService.getOrderedExecutor(config. getThreadPoolAlias(), executorQueue);
- }
- ● CacheLoaderWriter：此處我們只配置了write和delete，而writeAll和deleteAll將會把批量操作委托給write和delete。
- PooledExecutionService#getOrderedExecutor方法會創(chuàng)建PartitionedOrderedExecutor實(shí)例。
- PartitionedOrderedExecutor(BlockingQueue<Runnable> queue,ExecutorService executor) {
- this.delegate= new PartitionedUnorderedExecutor(queue, executor, 1);
- }
其使用maxWorkers=1創(chuàng)建了PartitionedUnorderedExecutor，然后Partitioned UnorderedExecutor通過this.runnerPermit = newSemaphore(maxWorkers)來控制并發(fā)，即maxWorkers=1就實(shí)現(xiàn)了一個并發(fā)。
因此，Ehcache實(shí)際能寫的***隊(duì)列大小為concurrency level *queue size。
因?yàn)閮?nèi)部使用線程池去寫，因此就實(shí)現(xiàn)了異步寫，又因?yàn)槭褂昧岁?duì)列，因此控制了總的吞吐量(此處有注意根據(jù)實(shí)際場景給線程池配置Rejected Policy)，接下來看下如何實(shí)現(xiàn)批量寫。
(2) 批量寫
- .withLoaderWriter(new DefaultCacheLoaderWriter<String,String>() {
- @Override
- publicvoid writeAll(Iterable<? extends Map.Entry<? extends String,? extends String>> entries) throws BulkCacheWritingException,Exception {
- for(Objectentry : entries) {
- //batchwrite
- }
- }
- @Override
- publicvoid deleteAll(Iterable<? extends String> keys) throws BulkCacheWritingException,Exception {
- for(Objectkey : keys) {
- //batchdelete
- }
- }
- })
- .add(WriteBehindConfigurationBuilder
- .newBatchedWriteBehindConfiguration(3,TimeUnit.SECONDS, 2)
- .queueSize(5)
- .concurrencyLevel(1)
- .enableCoalescing()
- .useThreadPool("writeBehindPool")
- .build()));
和上一個示例不同的地方是使用了newBatchedWriteBehindConfiguration進(jìn)行批量配置。
● newBatchedWriteBehindConfiguration(longmaxDelay, TimeUnit maxDelayUnit, int batchSize)：設(shè)置批處理大小和***延遲。batchSize用于定義批處理大小，當(dāng)寫操作數(shù)量等于批處理大小時，將把這一批數(shù)據(jù)發(fā)給CacheLoaderWriter進(jìn)行處理。Ehcache使用BatchingLocalHeapWriteBehindQueue實(shí)現(xiàn)批量隊(duì)列，其中操作批量的代碼如下。
- if (openBatch.add(operation)) {//往batch里添加操作，添加的數(shù)量等于批處理大小時
- submit(openBatch);//異步提交批處理操作
- openBatch= null;
- }
因此，Ehcache實(shí)際能寫的***隊(duì)列大小為concurrency level * queue size * batch size。
maxDelay用于配置未完成的批處理***延遲，比如，我們設(shè)置批處理大小為3，而我們實(shí)際只寫入了兩個數(shù)據(jù)，當(dāng)寫第3個數(shù)據(jù)時，會觸發(fā)提交批處理操作。但是，如果我們不寫第3個，那么將造成這2個數(shù)據(jù)一直等待，我們可以設(shè)置maxDelay，當(dāng)超時時也會將這兩個數(shù)據(jù)提交批處理。
● enableCoalescing：是否需要合并寫，即對于相同的Key只記錄***一次數(shù)據(jù)。
● CacheLoaderWriter：write和delete會轉(zhuǎn)換為writeAll和deleteAll，即批處理。
三、Copy Pattern
有兩種Copy Pattern，Copy-On-Read(在讀時復(fù)制)和Copy-On-Write(在寫時復(fù)制)，對于Guava Cache和Ehcache中堆緩存都是基于引用的，這樣如果有人拿到緩存數(shù)據(jù)并修改了它，則可能發(fā)生不可預(yù)測的問題，筆者就見過因?yàn)檫@種情況造成數(shù)據(jù)錯誤。Guava Cache沒有提供支持，Ehcache 3.x提供了支持。
- public interface Copier<T> {
- TcopyForRead(T obj); //Copy-On-Read，比如myCache.get()
- TcopyForWrite(T obj); //Copy-On-Write，比如myCache.put()
- }
通過如下方法來配置Key和Value的Copier。
- CacheConfigurationBuilder.withKeyCopier()
- CacheConfigurationBuilder.withValueCopier()
【本文是51CTO專欄作者張開濤的原創(chuàng)文章，作者微信公眾號：開濤的博客( kaitao-1234567)】
