国产成人亚洲精品蜜芽影院,亚洲日本一区二区一本一道,久久亚洲国产中v天仙www

MyBatis 緩存機制深度解剖 / 自定義二級緩存

http://denger.iteye.com/blog/1126423

緩存概述

正如大多數(shù)持久層框架一樣，MyBatis 同樣提供了一級緩存和二級緩存的支持；
一級緩存基于 PerpetualCache 的 HashMap 本地緩存，其存儲作用域為 Session，當 Session flush 或 close 之后，該Session中的所有 Cache 就將清空。
二級緩存與一級緩存其機制相同，默認也是采用 PerpetualCache，HashMap存儲，不同在于其存儲作用域為 Mapper(Namespace)，并且可自定義存儲源，如 Ehcache、Hazelcast等。
對于緩存數(shù)據(jù)更新機制，當某一個作用域(一級緩存Session/二級緩存Namespaces)的進行了 C/U/D 操作后，默認該作用域下所有 select 中的緩存將被clear。
MyBatis 的緩存采用了delegate機制及裝飾器模式設計，當put、get、remove時，其中會經(jīng)過多層 delegate cache 處理，其Cache類別有：BaseCache(基礎緩存)、EvictionCache(排除算法緩存) 、DecoratorCache(裝飾器緩存)：

BaseCache ：

EvictionCache ：

DecoratorCache：

一般緩存框架的數(shù)據(jù)結構基本上都是 Key-Value 方式存儲，MyBatis 對于其 Key 的生成采取規(guī)則為：[hashcode : checksum : mappedStementId : offset : limit : executeSql : queryParams]。
對于并發(fā) Read/Write 時緩存數(shù)據(jù)的同步問題，MyBatis 默認基于 JDK/concurrent中的ReadWriteLock，使用ReentrantReadWriteLock 的實現(xiàn)，從而通過 Lock 機制防止在并發(fā) Write Cache 過程中線程安全問題。

源碼剖解
接下來將結合 MyBatis 序列圖進行源碼分析。在分析其Cache前，先看看其整個處理過程。
執(zhí)行過程:

① 通常情況下，我們需要在 Service 層調(diào)用 Mapper Interface 中的方法實現(xiàn)對數(shù)據(jù)庫的操作，上述根據(jù)產(chǎn)品 ID 獲取 Product 對象。
② 當調(diào)用 ProductMapper 時中的方法時，其實這里所調(diào)用的是 MapperProxy 中的方法，并且 MapperProxy已經(jīng)將將所有方法攔截，其具體原理及分析，參考 MyBatis+Spring基于接口編程的原理分析，其 invoke 方法代碼為：

Java代碼  
//當調(diào)用 Mapper 所有的方法時，將都交由Proxy 中的 invoke 處理：  
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {  
    try {  
      if (!OBJECT_METHODS.contains(method.getName())) {  
        final Class declaringInterface = findDeclaringInterface(proxy, method);  
        // 最終交由 MapperMethod 類處理數(shù)據(jù)庫操作，初始化 MapperMethod 對象  
        final MapperMethod mapperMethod = new MapperMethod(declaringInterface, method, sqlSession);  
        // 執(zhí)行 mapper method，返回執(zhí)行結果   
        final Object result = mapperMethod.execute(args);  
        ....  
        return result;  
      }  
    } catch (SQLException e) {  
      e.printStackTrace();  
    }  
    return null;  
  }  

③其中的 mapperMethod 中的 execute 方法代碼如下：

Java代碼  
public Object execute(Object[] args) throws SQLException {  
    Object result;  
    // 根據(jù)不同的操作類別，調(diào)用 DefaultSqlSession 中的執(zhí)行處理  
    if (SqlCommandType.INSERT == type) {  
      Object param = getParam(args);  
      result = sqlSession.insert(commandName, param);  
    } else if (SqlCommandType.UPDATE == type) {  
      Object param = getParam(args);  
      result = sqlSession.update(commandName, param);  
    } else if (SqlCommandType.DELETE == type) {  
      Object param = getParam(args);  
      result = sqlSession.delete(commandName, param);  
    } else if (SqlCommandType.SELECT == type) {  
      if (returnsList) {  
        result = executeForList(args);  
      } else {  
        Object param = getParam(args);  
        result = sqlSession.selectOne(commandName, param);  
      }  
    } else {  
      throw new BindingException("Unkown execution method for: " + commandName);  
    }  
    return result;  
  }  

由于這里是根據(jù) ID 進行查詢，所以最終調(diào)用為 sqlSession.selectOne函數(shù)。也就是接下來的的 DefaultSqlSession.selectOne 執(zhí)行；
④ ⑤ 可以在 DefaultSqlSession 看到，其 selectOne 調(diào)用了 selectList 方法：

Java代碼  
public Object selectOne(String statement, Object parameter) {  
    List list = selectList(statement, parameter);  
    if (list.size() == 1) {  
      return list.get(0);  
    }   
    ...  
}  
  
public List selectList(String statement, Object parameter, RowBounds rowBounds) {  
    try {  
      MappedStatement ms = configuration.getMappedStatement(statement);  
      // 如果啟動用了Cache 才調(diào)用 CachingExecutor.query，反之則使用 BaseExcutor.query 進行數(shù)據(jù)庫查詢   
      return executor.query(ms, wrapCollection(parameter), rowBounds, Executor.NO_RESULT_HANDLER);  
    } catch (Exception e) {  
      throw ExceptionFactory.wrapException("Error querying database.  Cause: " + e, e);  
    } finally {  
      ErrorContext.instance().reset();  
    }  
}  

⑥到這里，已經(jīng)執(zhí)行到具體數(shù)據(jù)查詢的流程，在分析 CachingExcutor.query 前，先看看 MyBatis 中 Executor 的結構及構建過程。

執(zhí)行器(Executor):
Executor: 執(zhí)行器接口。也是最終執(zhí)行數(shù)據(jù)獲取及更新的實例。其類結構如下：

BaseExecutor: 基礎執(zhí)行器抽象類。實現(xiàn)一些通用方法，如createCacheKey 之類。并且采用模板模式將具體的數(shù)據(jù)庫操作邏輯(doUpdate、doQuery)交由子類實現(xiàn)。另外，可以看到變量 localCache: PerpetualCache，在該類采用 PerpetualCache 實現(xiàn)基于 Map 存儲的一級緩存，其 query 方法如下：

Java代碼  
public List query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {  
    ErrorContext.instance().resource(ms.getResource()).activity("executing a query").object(ms.getId());  
    // 執(zhí)行器已關閉  
    if (closed) throw new ExecutorException("Executor was closed.");  
    List list;  
    try {  
      queryStack++;   
      // 創(chuàng)建緩存Key  
      CacheKey key = createCacheKey(ms, parameter, rowBounds);   
      // 從本地緩存在中獲取該 key 所對應 的結果集  
      final List cachedList = (List) localCache.getObject(key);   
      // 在緩存中找到數(shù)據(jù)  
      if (cachedList != null) {   
        list = cachedList;  
      } else { // 未從本地緩存中找到數(shù)據(jù)，開始調(diào)用數(shù)據(jù)庫查詢  
        //為該 key 添加一個占位標記  
        localCache.putObject(key, EXECUTION_PLACEHOLDER);   
        try {  
          // 執(zhí)行子類所實現(xiàn)的數(shù)據(jù)庫查詢 操作  
          list = doQuery(ms, parameter, rowBounds, resultHandler);   
        } finally {  
          // 刪除該 key 的占位標記  
          localCache.removeObject(key);  
        }  
        // 將db中的數(shù)據(jù)添加至本地緩存中  
        localCache.putObject(key, list);  
      }  
    } finally {  
      queryStack--;  
    }  
    // 刷新當前隊列中的所有 DeferredLoad實例，更新 MateObject  
    if (queryStack == 0) {   
      for (DeferredLoad deferredLoad : deferredLoads) {  
        deferredLoad.load();  
      }  
    }  
    return list;  
  }  

BatchExcutor、ReuseExcutor、 SimpleExcutor: 這幾個就沒什么好說的了，繼承了 BaseExcutor 的實現(xiàn)其 doQuery、doUpdate 等方法，同樣都是采用 JDBC 對數(shù)據(jù)庫進行操作；三者區(qū)別在于，批量執(zhí)行、重用 Statement 執(zhí)行、普通方式執(zhí)行。具體應用及場景在Mybatis 的文檔上都有詳細說明。

CachingExecutor: 二級緩存執(zhí)行器。個人覺得這里設計的不錯，靈活地使用 delegate機制。其委托執(zhí)行的類是 BaseExcutor。當無法從二級緩存獲取數(shù)據(jù)時，同樣需要從 DB 中進行查詢，于是在這里可以直接委托給 BaseExcutor 進行查詢。其大概流程為：

流程為：從二級緩存中進行查詢 -> [如果緩存中沒有，委托給 BaseExecutor] -> 進入一級緩存中查詢 -> [如果也沒有] -> 則執(zhí)行 JDBC 查詢，其 query 代碼如下：

Java代碼  
public List query(MappedStatement ms, Object parameterObject, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException {  
    if (ms != null) {  
      // 獲取二級緩存實例  
      Cache cache = ms.getCache();  
      if (cache != null) {  
        flushCacheIfRequired(ms);  
        // 獲取 讀鎖( Read鎖可由多個Read線程同時保持)  
        cache.getReadWriteLock().readLock().lock();  
        try {  
          // 當前 Statement 是否啟用了二級緩存  
          if (ms.isUseCache()) {  
            // 將創(chuàng)建 cache key 委托給 BaseExecutor 創(chuàng)建  
            CacheKey key = createCacheKey(ms, parameterObject, rowBounds);  
            final List cachedList = (List) cache.getObject(key);  
            // 從二級緩存中找到緩存數(shù)據(jù)  
            if (cachedList != null) {  
              return cachedList;  
            } else {  
              // 未找到緩存，很委托給 BaseExecutor 執(zhí)行查詢  
              List list = delegate.query(ms, parameterObject, rowBounds, resultHandler);  
              tcm.putObject(cache, key, list);  
              return list;  
            }  
          } else { // 沒有啟動用二級緩存，直接委托給 BaseExecutor 執(zhí)行查詢   
            return delegate.query(ms, parameterObject, rowBounds, resultHandler);  
          }  
        } finally {  
          // 當前線程釋放 Read 鎖  
          cache.getReadWriteLock().readLock().unlock();  
        }  
      }  
    }  
    return delegate.query(ms, parameterObject, rowBounds, resultHandler);  
}  

至此，已經(jīng)完完了整個緩存執(zhí)行器的整個流程分析，接下來是對緩存的緩存數(shù)據(jù)管理實例進行分析，也就是其 Cache 接口，用于對緩存數(shù)據(jù) put 、get及remove的實例對象。

Cache 委托鏈構建:
正如最開始的緩存概述所描述道，其緩存類的設計采用裝飾模式，基于委托的調(diào)用機制。
緩存實例構建：
緩存實例的構建，Mybatis 在解析其 Mapper 配置文件時就已經(jīng)將該實現(xiàn)初始化，在 org.apache.ibatis.builder.xml.XMLMapperBuilder 類中可以看到：

Java代碼  
private void cacheElement(XNode context) throws Exception {  
    if (context != null) {  
      // 基礎緩存類型  
      String type = context.getStringAttribute("type", "PERPETUAL");  
      Class typeClass = typeAliasRegistry.resolveAlias(type);  
      // 排除算法緩存類型  
      String eviction = context.getStringAttribute("eviction", "LRU");  
      Class evictionClass = typeAliasRegistry.resolveAlias(eviction);  
      // 緩存自動刷新時間  
      Long flushInterval = context.getLongAttribute("flushInterval");  
      // 緩存存儲實例引用的大小  
      Integer size = context.getIntAttribute("size");  
      // 是否是只讀緩存  
      boolean readWrite = !context.getBooleanAttribute("readOnly", false);  
      Properties props = context.getChildrenAsProperties();  
      // 初始化緩存實現(xiàn)  
      builderAssistant.useNewCache(typeClass, evictionClass, flushInterval, size, readWrite, props);  
    }  
  }  

以下是 useNewCache 方法實現(xiàn)：

Java代碼  
public Cache useNewCache(Class typeClass,  
                           Class evictionClass,  
                           Long flushInterval,  
                           Integer size,  
                           boolean readWrite,  
                           Properties props) {  
    typeClass = valueOrDefault(typeClass, PerpetualCache.class);  
    evictionClass = valueOrDefault(evictionClass, LruCache.class);  
    // 這里構建 Cache 實例采用 Builder 模式，每一個 Namespace 生成一個  Cache 實例  
    Cache cache = new CacheBuilder(currentNamespace)  
        // Builder 前設置一些從XML中解析過來的參數(shù)  
        .implementation(typeClass)  
        .addDecorator(evictionClass)  
        .clearInterval(flushInterval)  
        .size(size)  
        .readWrite(readWrite)  
        .properties(props)  
        // 再看下面的 build 方法實現(xiàn)  
        .build();  
    configuration.addCache(cache);  
    currentCache = cache;  
    return cache;  
}  
  
public Cache build() {  
    setDefaultImplementations();  
    // 創(chuàng)建基礎緩存實例  
    Cache cache = newBaseCacheInstance(implementation, id);  
    setCacheProperties(cache);  
    // 緩存排除算法初始化，并將其委托至基礎緩存中  
    for (Class<? extends Cache> decorator : decorators) {  
      cache = newCacheDecoratorInstance(decorator, cache);  
      setCacheProperties(cache);  
    }  
    // 標準裝飾器緩存設置，如LoggingCache之類，同樣將其委托至基礎緩存中  
    cache = setStandardDecorators(cache);  
    // 返回最終緩存的責任鏈對象  
    return cache;  
}  

最終生成后的緩存實例對象結構：

可見，所有構建的緩存實例已經(jīng)通過責任鏈方式將其串連在一起，各 Cache 各負其責、依次調(diào)用，直到緩存數(shù)據(jù)被 Put 至基礎緩存實例中存儲。

Cache 實例解剖:
實例類：SynchronizedCache
說明：用于控制 ReadWriteLock，避免并發(fā)時所產(chǎn)生的線程安全問題。
解剖：
對于 Lock 機制來說，其分為 Read 和 Write 鎖，其 Read 鎖允許多個線程同時持有，而 Write 鎖，一次能被一個線程持有，如果當 Write 鎖沒有釋放，其它需要 Write 的線程只能等待其釋放才能去持有。
其代碼實現(xiàn)：

Java代碼  
public void putObject(Object key, Object object) {  
    acquireWriteLock();  // 獲取 Write 鎖  
    try {  
      delegate.putObject(key, object); // 委托給下一個 Cache 執(zhí)行 put 操作  
    } finally {  
      releaseWriteLock(); // 釋放 Write 鎖  
    }  
  }  

對于 Read 數(shù)據(jù)來說，也是如此，不同的是 Read 鎖允許多線程同時持有 :

Java代碼  
public Object getObject(Object key) {  
    acquireReadLock();  
    try {  
      return delegate.getObject(key);  
    } finally {  
      releaseReadLock();  
    }  
  }  

其具體原理可以看看 jdk concurrent 中的 ReadWriteLock 實現(xiàn)。

實例類：LoggingCache
說明：用于日志記錄處理，主要輸出緩存命中率信息。
解剖：
說到緩存命中信息的統(tǒng)計，只有在 get 的時候才需要統(tǒng)計命中率：

Java代碼  
public Object getObject(Object key) {  
    requests++; // 每調(diào)用一次該方法，則獲取次數(shù)+1  
    final Object value = delegate.getObject(key);  
    if (value != null) {  // 命中！ 命中+1  
      hits++;  
    }  
    if (log.isDebugEnabled()) {  
      // 輸出命中率。計算方法為： hits / requets 則為命中率  
      log.debug("Cache Hit Ratio [" + getId() + "]: " + getHitRatio());  
    }  
    return value;  
}  

實例類：SerializedCache
說明：向緩存中 put 或 get 數(shù)據(jù)時的序列化及反序列化處理。
解剖：
序列化在Java里面已經(jīng)是最基礎的東西了，這里也沒有什么特殊之處：

Java代碼  
public void putObject(Object key, Object object) {  
     // PO 類需要實現(xiàn) Serializable 接口  
    if (object == null || object instanceof Serializable) {  
      delegate.putObject(key, serialize((Serializable) object));   
    } else {  
      throw new CacheException("SharedCache failed to make a copy of a non-serializable object: " + object);  
    }  
  }  
  
  public Object getObject(Object key) {  
    Object object = delegate.getObject(key);  
    // 獲取數(shù)據(jù)時對 二進制數(shù)據(jù)進行反序列化  
    return object == null ? null : deserialize((byte[]) object);  
  }  

其 serialize 及 deserialize 代碼就不必要貼了。

實例類：LruCache
說明：最近最少使用的:移除最長時間不被使用的對象，基于LRU算法。
解剖：
這里的 LRU 算法基于 LinkedHashMap 覆蓋其 removeEldestEntry 方法實現(xiàn)。好象之前看過 XMemcached 的 LRU 算法也是這樣實現(xiàn)的。
初始化 LinkedHashMap，默認為大小為 1024 個元素：

Java代碼  
public LruCache(Cache delegate) {  
    this.delegate = delegate;  
    setSize(1024); // 設置 map 默認大小  
}  
public void setSize(final int size) {  
    // 設置其 capacity 為size, 其 factor 為.75F  
    keyMap = new LinkedHashMap(size, .75F, true) {  
      // 覆蓋該方法，當每次往該map 中put 時數(shù)據(jù)時，如該方法返回 True，便移除該map中使用最少的Entry  
      // 其參數(shù)  eldest 為當前最老的  Entry  
      protected boolean removeEldestEntry(Map.Entry eldest) {  
        boolean tooBig = size() > size;  
        if (tooBig) {  
          eldestKey = eldest.getKey(); //記錄當前最老的緩存數(shù)據(jù)的 Key 值，因為要委托給下一個 Cache 實現(xiàn)刪除  
        }  
        return tooBig;  
      }  
    };  
  }  
  
public void putObject(Object key, Object value) {  
    delegate.putObject(key, value);  
    cycleKeyList(key);  // 每次 put 后，調(diào)用移除最老的 key  
}  
// 看看當前實現(xiàn)是否有 eldestKey, 有的話就調(diào)用 removeObject ，將該key從cache中移除  
private void cycleKeyList(Object key) {  
    keyMap.put(key, key); // 存儲當前 put 到cache中的 key 值  
    if (eldestKey != null) {  
      delegate.removeObject(eldestKey);  
      eldestKey = null;  
    }  
  }  
  
public Object getObject(Object key) {  
    keyMap.get(key); // 便于 該 Map 統(tǒng)計 get該key的次數(shù)  
    return delegate.getObject(key);  
  }  

實例類：PerpetualCache
說明：這個比較簡單，直接通過一個 HashMap 來存儲緩存數(shù)據(jù)。所以沒什么說的，直接看下面的 MemcachedCache 吧。

自定義二級緩存/Memcached
其自定義二級緩存也較為簡單，它本身默認提供了對 Ehcache 及 Hazelcast 的緩存支持：Mybatis-Cache，我這里參考它們的實現(xiàn)，自定義了針對 Memcached 的緩存支持，其代碼如下:

Java代碼  
package com.xx.core.plugin.mybatis;  
  
import java.util.LinkedList;  
import java.util.concurrent.locks.ReadWriteLock;  
import java.util.concurrent.locks.ReentrantReadWriteLock;  
  
import org.apache.ibatis.cache.Cache;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
  
import com.xx.core.memcached.JMemcachedClientAdapter;  
import com.xx.core.memcached.service.CacheService;  
import com.xx.core.memcached.service.MemcachedService;  
  
/** 
 * Cache adapter for Memcached. 
 *  
 * @author denger 
 */  
public class MemcachedCache implements Cache {  
  
    // Sf4j logger reference  
    private static Logger logger = LoggerFactory.getLogger(MemcachedCache.class);  
  
    /** The cache service reference. */  
    protected static final CacheService CACHE_SERVICE = createMemcachedService();  
  
    /** The ReadWriteLock. */  
    private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();  
  
    private String id;  
    private LinkedList<String> cacheKeys = new LinkedList<String>();  
  
    public MemcachedCache(String id) {  
        this.id = id;  
    }  
    // 創(chuàng)建緩存服務類，基于java-memcached-client  
    protected static CacheService createMemcachedService() {  
        JMemcachedClientAdapter memcachedAdapter;  
  
        try {  
            memcachedAdapter = new JMemcachedClientAdapter();  
        } catch (Exception e) {  
            String msg = "Initial the JMmemcachedClientAdapter Error.";  
            logger.error(msg, e);  
            throw new RuntimeException(msg);  
        }  
        return new MemcachedService(memcachedAdapter);  
    }  
  
    @Override  
    public String getId() {  
        return this.id;  
    }  
  
    // 根據(jù) key 從緩存中獲取數(shù)據(jù)  
    @Override  
    public Object getObject(Object key) {  
        String cacheKey = String.valueOf(key.hashCode());  
        Object value = CACHE_SERVICE.get(cacheKey);  
        if (!cacheKeys.contains(cacheKey)){  
            cacheKeys.add(cacheKey);  
        }  
        return value;  
    }  
  
    @Override  
    public ReadWriteLock getReadWriteLock() {  
        return this.readWriteLock;  
    }  
  
    // 設置數(shù)據(jù)至緩存中  
    @Override  
    public void putObject(Object key, Object value) {  
        String cacheKey = String.valueOf(key.hashCode());  
  
        if (!cacheKeys.contains(cacheKey)){  
            cacheKeys.add(cacheKey);  
        }  
        CACHE_SERVICE.put(cacheKey, value);  
    }  
    // 從緩存中刪除指定 key 數(shù)據(jù)  
    @Override  
    public Object removeObject(Object key) {  
        String cacheKey = String.valueOf(key.hashCode());  
  
        cacheKeys.remove(cacheKey);  
        return CACHE_SERVICE.delete(cacheKey);  
    }  
    //清空當前 Cache 實例中的所有緩存數(shù)據(jù)  
    @Override  
    public void clear() {  
        for (int i = 0; i < cacheKeys.size(); i++){  
            String cacheKey = cacheKeys.get(i);  
            CACHE_SERVICE.delete(cacheKey);  
        }  
        cacheKeys.clear();  
    }  
  
    @Override  
    public int getSize() {  
        return cacheKeys.size();  
    }  
}  

在 ProductMapper 中增加配置：

Xml代碼  
<cache eviction="LRU" type="com.xx.core.plugin.mybatis.MemcachedCache" />  

啟動Memcached:

Shell代碼  
memcached -c 2000 -p 11211 -vv -U 0 -l 192.168.1.2 -v  

執(zhí)行Mapper 中的查詢、修改等操作，Test:

Java代碼  
@Test  
    public void testSelectById() {  
        Long pid = 100L;  
  
        Product dbProduct = productMapper.selectByID(pid);  
        Assert.assertNotNull(dbProduct);  
  
        Product cacheProduct = productMapper.selectByID(pid);  
        Assert.assertNotNull(cacheProduct);  
  
        productMapper.updateName("IPad", pid);  
  
        Product product = productMapper.selectByID(pid);  
        Assert.assertEquals(product.getName(), "IPad");  
    }  

Memcached Loging:

看上去沒什么問題~ OK了。

posted on 2014-05-30 13:53 SIMONE 閱讀(602) 評論(1) 編輯收藏所屬分類: JAVA

常用鏈接

留言簿(46)

隨筆分類(476)

隨筆檔案(495)

最新隨筆

搜索

積分與排名

最新評論

閱讀排行榜

評論排行榜


只有注冊用戶登錄后才能發(fā)表評論。




網(wǎng)站導航: 博客園 IT新聞 Chat2DB C++博客博問管理
相關文章: java keytool證書工具使用小結 Java 和 HTTP 的那些事（二）使用代理使用embeded tomcat進行嵌入式javaee開發(fā)－啟動tomcat java修改static final常量值 Max MQTT connections HDFS配置Kerberos認證 Benchmarking Apache Kafka: 2 Million Writes Per Second (On Three Cheap Machines) Kafka 高性能吞吐揭秘 JAVA實現(xiàn)gif圖片縮放與剪切功能基于Redis實現(xiàn)分布式鎖