張開濤：應用級緩存之緩存使用模式實踐—
前面已經介紹了Java緩存的使用。對于我們來說如果有人總結一些緩存使用模式/模板的話，我們在使用時直接照著模式寫即可。而實際確實已經有總結好的模式，主要分兩大類：Cache-Aside和Cache-As-SoR(Read-through、Write-through、Write-behind)。
首先，同步兩個名詞。
- SoR(system-of-record)：記錄系統，或者可以叫做數據源，即實際存儲原始數據的系統。
- Cache：緩存，是SoR的快照數據，Cache的訪問速度比SoR要快，放入Cache的目的是提升訪問速度，減少回源到SoR的次數。
- 回源：即回到數據源頭獲取數據，Cache沒有***時，需要從SoR讀取數據，這叫做回源。
本文主要以Guava Cache和Ehcache3.x作為實踐框架來講解。
一、Cache-Aside
Cache-Aside即業務代碼圍繞著Cache寫，是由業務代碼直接維護緩存，示例代碼如下所示。
讀場景，先從緩存獲取數據，如果沒有***，則回源到SoR并將源數據放入緩存供下次讀取使用。
- //1、先從緩存中獲取數據
- value = myCache.getIfPresent(key);
- if(value == null) {
- //2.1、如果緩存沒有***，則回源到SoR獲取源數據
- value = loadFromSoR(key);
- //2.2、將數據放入緩存，下次即可從緩存中獲取數據
- myCache.put(key, value);
- }
寫場景，先將數據寫入SoR，寫入成功后立即將數據同步寫入緩存。
- //1、先將數據寫入SoR
- writeToSoR(key,value);
- //2、執行成功后立即同步寫入緩存
- myCache.put(key, value);
或者先將數據寫入SoR，寫入成功后將緩存數據過期，下次讀取時再加載緩存。
- //1、先將數據寫入SoR
- writeToSoR(key,value);
- //2、失效緩存，然后下次讀時再加載緩存
- myCache.invalidate(key);
Cache-Aside適合使用AOP模式去實現，可以參考筆者的博客《Spring Cache抽象詳解》去實現。
對于Cache-Aside可能存在并發更新情況，即如果多個應用實例同時更新，那么緩存怎么辦?
● 如果是用戶維度的數據(如訂單數據、用戶數據)，則出現這種幾率非常小，因為并發的情況很少，可以不考慮這個問題，加上過期時間來解決即可。
● 對于如商品這種基礎數據，可以考慮使用canal訂閱binlog進行增量更新分布式緩存，這樣不會存在緩存數據不一致的情況，但是，緩存更新會存在延遲。而本地緩存根據不一致容忍度設置合理的過期時間。
● 讀服務場景，可以考慮使用一致性哈希，將相同的操作負載均衡到同一個實例，從而減少并發幾率。或者設置比較短的過期時間，可參考“第17章 京東商品詳情頁服務閉環實踐”。
二、Cache-As-SoR
Cache-As-SoR即把Cache看作為SoR，所有操作都是對Cache進行，然后Cache再委托給SoR進行真實的讀/寫。即業務代碼中只看到Cache的操作，看不到關于SoR相關的代碼。有三種實現：read-through、write-through、write-behind。
1. Read-Through
Read-Through，業務代碼首先調用Cache，如果Cache不***由Cache回源到SoR，而不是業務代碼(即由Cache讀SoR)。使用Read-Through模式，需要配置一個CacheLoader組件用來回源到SoR加載源數據。Guava Cache和Ehcache 3.x都支持該模式。
Guava Cache實現
- 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用來加載緩存，操作流程如下。
- 應用業務代碼直接調用getCache.get(sortId)。
- 首先查詢Cache，如果緩存中有，則直接返回緩存數據。
- 如果緩存沒有***，則委托給CacheLoader，CacheLoader會回源到SoR查詢源數據(返回值必須不為null，可以包裝為Null對象)，然后寫入緩存。
使用CacheLoader后有幾個好處。
● 應用業務代碼更簡潔了，不需要像Cache-Aside模式那樣緩存查詢代碼和SoR代碼交織在一起。如果緩存使用邏輯散落在多處，則使用這種方式很簡單的消除了重復代碼。
● 解決Dog-pile effect，即當某個緩存失效時，又有大量相同的請求沒***緩存，從而同時請求到后端，導致后端壓力太大，此時限制一個請求去拿即可。
- if (firstCreateNewEntry) {//***個請求加載緩存的線程去SoR加載源數據
- try {
- synchronized (e) {
- returnloadSync(key, hash, loadingValueReference, loader);
- }
- } finally{
- statsCounter.recordMisses(1);
- }
- } else {//其他并發線程等待“***個線程”加載的數據
- return waitForLoadingValue(e, key,valueReference);
- }
- Guava Cache還支持get(K key, Callable<? extends V> valueLoader)方法，傳入一個Callable實例，當緩存沒***時，會調用Callable#call來查詢SoR加載源數據。
- Ehcache 3.x實現
- 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來實現，通過load(K key)和loadAll(Iterable keys)分別來加載單個KEY和批量KEY。Ehcache 3.1沒有自己去解決Dog-pile effect。
2. Write-Through
Write-Through，稱之為穿透寫模式/直寫模式，業務代碼首先調用Cache寫(新增/修改)數據，然后由Cache負責寫緩存和寫SoR，而不是業務代碼。使用Write-Through模式需要配置一個CacheWriter組件用來回寫SoR。Guava Cache沒有提供支持。Ehcache 3.x支持該模式。Ehcache需要配置一個CacheLoaderWriter，CacheLoaderWriter知道如何去寫SoR。當Cache需要寫(新增/修改)數據時，首先調用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來實現，通過write(String key, String value)、writeAll(Iterable> entries)和delete(String key)、deleteAll(Iterable keys)分別來支持單個寫、批量寫和單個刪除、批量刪除操作。
操作流程如下。
- 當我們調用myCache.put("e","123")或者myCache.putAll(map)時，寫緩存。
- 首先，Cache會將寫操作立即委托給CacheLoaderWriter#write和#writeAll，然后由CacheLoaderWriter負責立即去寫SoR。
- 當寫SoR成功后，再寫入Cache。
3. Write-Behind
Write-Behind，也叫Write-Back，稱之為回寫模式，不同于Write-Through是同步寫SoR和Cache，Write-Behind是異步寫。異步之后可以實現批量寫、合并寫、延時和限流。
(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會進行如下幾個配置。
- newUnBatchedWriteBehindConfiguration：表示不進行批量處理，那么所有批量操作都將會轉換成單個操作，即CacheLoaderWriter只需要實現write和delete即可。
- queueSize(int size)：因為操作是異步回寫SoR，需要將操作先放入寫操作等待隊列，因此，使用queue size定義寫操作等待隊列***大小，即線程池隊列大小。內部使用NonBatchingLocalHeapWriteBehindQueue。
- concurrencyLevel(int concurrency)：配置使用多少個并發線程和隊列進行WriteBehind。因為我們只傳入一個線程池，這是如何實現該模式的呢?首先看如下代碼片段。
- 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));
- }
- }
可以看到會創建concurrencyLevel個隊列NonBatchingLocalHeapWriteBehindQueue，其又通過如下代碼片段創建線程池和線程池隊列。
- 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方法會創建PartitionedOrderedExecutor實例。
- PartitionedOrderedExecutor(BlockingQueue<Runnable> queue,ExecutorService executor) {
- this.delegate= new PartitionedUnorderedExecutor(queue, executor, 1);
- }
其使用maxWorkers=1創建了PartitionedUnorderedExecutor，然后Partitioned UnorderedExecutor通過this.runnerPermit = newSemaphore(maxWorkers)來控制并發，即maxWorkers=1就實現了一個并發。
因此，Ehcache實際能寫的***隊列大小為concurrency level *queue size。
因為內部使用線程池去寫，因此就實現了異步寫，又因為使用了隊列，因此控制了總的吞吐量(此處有注意根據實際場景給線程池配置Rejected Policy)，接下來看下如何實現批量寫。
(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進行批量配置。
● newBatchedWriteBehindConfiguration(longmaxDelay, TimeUnit maxDelayUnit, int batchSize)：設置批處理大小和***延遲。batchSize用于定義批處理大小，當寫操作數量等于批處理大小時，將把這一批數據發給CacheLoaderWriter進行處理。Ehcache使用BatchingLocalHeapWriteBehindQueue實現批量隊列，其中操作批量的代碼如下。
- if (openBatch.add(operation)) {//往batch里添加操作，添加的數量等于批處理大小時
- submit(openBatch);//異步提交批處理操作
- openBatch= null;
- }
因此，Ehcache實際能寫的***隊列大小為concurrency level * queue size * batch size。
maxDelay用于配置未完成的批處理***延遲，比如，我們設置批處理大小為3，而我們實際只寫入了兩個數據，當寫第3個數據時，會觸發提交批處理操作。但是，如果我們不寫第3個，那么將造成這2個數據一直等待，我們可以設置maxDelay，當超時時也會將這兩個數據提交批處理。
● enableCoalescing：是否需要合并寫，即對于相同的Key只記錄***一次數據。
● CacheLoaderWriter：write和delete會轉換為writeAll和deleteAll，即批處理。
三、Copy Pattern
有兩種Copy Pattern，Copy-On-Read(在讀時復制)和Copy-On-Write(在寫時復制)，對于Guava Cache和Ehcache中堆緩存都是基于引用的，這樣如果有人拿到緩存數據并修改了它，則可能發生不可預測的問題，筆者就見過因為這種情況造成數據錯誤。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專欄作者張開濤的原創文章，作者微信公眾號：開濤的博客( kaitao-1234567)】
