MyBatis緩存模塊源碼分析

優秀的ORM框架都應該提供緩存機制，MyBatis也不例外，在org.apache.ibatis.cache包下面定義了MyBatis緩存的核心模塊，需要注意的是這個包中只是MyBatis緩存的核心實現，并不涉及一級緩存和二級緩存的實現，本文同樣沒有涉及到一二級緩存的具體實現方式的講解。

在閱讀緩存模塊源碼之前，讀者們應該首先弄懂裝飾器模式的含義，因為緩存模塊的實現是裝飾器模式的一種最佳實踐，只有弄懂了裝飾器模式才能更好的理解緩存模塊的原理。

一. 初探緩存模塊源碼結構

緩存模塊源碼結構如下：

在這里插入圖片描述

decorators包：該包存放的是裝飾器，通過這些裝飾器可以為緩存添加一下核心功能，例如：防止緩存擊穿，添加緩存清空策略，日志功能、序列化功能、定時清空功能。
- BlockingCache：防止緩存擊穿的裝飾器。
- FifoCache：緩存淘汰策略裝飾器（先進先出）。
- LoggingCache：緩存的日志裝飾器，用于輸出緩存命中率。
- LruCache：緩存淘汰策略裝飾器（最近最少使用）
- ScheduledCache：緩存清空計劃裝飾器，該裝飾器目的是實現每小時清空一次緩存。
- SerializedCache：緩存值序列化裝飾器。將對象存存入緩存之前將對象序列化為字節數組存入緩存，在get時反序列化為對象。
- SoftCache：軟引用緩存裝飾器。
- SynchronizedCache：同步裝飾器，用于保證緩存的更新操作是線程安全的。
- TransactionalCache：二級緩存事務緩沖區。
- WeakCache：弱引用緩存裝飾器。
impl包：該包下只有一個PerpetualCache類，它就是緩存模塊的核心類。

一. 緩存模塊的核心接口（Cache）

/**
 * MyBatis緩存模塊采用裝飾器模式
 */
public interface Cache {

  /**
   * @return 獲取這個緩存的ID
   */
  String getId();

  void putObject(Object key, Object value);

  Object getObject(Object key);

  Object removeObject(Object key);

  void clear();

  int getSize();

  //獲取讀寫鎖。從3.2.6開始，這個方法不再被核心調用
  default ReadWriteLock getReadWriteLock() {
    return null;
  }

}

可以看到putObject方法的key不是使用的String而是Object，這是因為MyBatis涉及動態SQL的原因，緩存項的key不能僅僅通過一個String來表示，所以通過CacheKey來封裝緩存的Key。在CacheKey中封裝類多個影響緩存項的因素。

構成CacheKey的對象：

mappedStatement的id
指定查詢結果集的范圍（分頁信息）
查詢所使用的SQL語句
用戶傳遞給SQL語句的實際參數值

/**
 * 緩存key
 * MyBatis 對于其 Key 的生成采取規則為：[mappedStementId + offset + limit + SQL + queryParams + environment]生成一個哈希碼
 */
public class CacheKey implements Cloneable, Serializable {

  private static final long serialVersionUID = 1146682552656046210L;

  public static final CacheKey NULL_CACHE_KEY = new CacheKey() {

    @Override
    public void update(Object object) {
      throw new CacheException("Not allowed to update a null cache key instance.");
    }

    @Override
    public void updateAll(Object[] objects) {
      throw new CacheException("Not allowed to update a null cache key instance.");
    }
  };

  private static final int DEFAULT_MULTIPLIER = 37;
  private static final int DEFAULT_HASHCODE = 17;

  private final int multiplier;//參與hash計算的乘數
  private int hashcode; //當前CacheKey的HashCode
  private long checksum; //校驗和
  private int count; //updateList的元素個數
  // 8/21/2017 - Sonarlint flags this as needing to be marked transient. While true if content is not serializable, this
  // is not always true and thus should not be marked transient.
  private List<Object> updateList;

  public CacheKey() {
    this.hashcode = DEFAULT_HASHCODE;
    this.multiplier = DEFAULT_MULTIPLIER;
    this.count = 0;
    this.updateList = new ArrayList<>();
  }

  public CacheKey(Object[] objects) {
    this();
    updateAll(objects);
  }

  public int getUpdateCount() {
    return updateList.size();
  }

  public void update(Object object) {
    int baseHashCode = object == null ? 1 : ArrayUtil.hashCode(object);

    count++; //updateList中的size +1
    checksum += baseHashCode;//計算校驗和
    baseHashCode *= count;

    hashcode = multiplier * hashcode + baseHashCode; //更新Hash值

    updateList.add(object);
  }

  public void updateAll(Object[] objects) {
    for (Object o : objects) {
      update(o);
    }
  }

  @Override
  public boolean equals(Object object) {
    if (this == object) {
      return true;
    }
    if (!(object instanceof CacheKey)) {
      return false;
    }

    final CacheKey cacheKey = (CacheKey) object;

    /**
     * 這三個值都相同equals才有可能相同，目的是快速剔除大部分不相同的對象
     */
    if (hashcode != cacheKey.hashcode) {
      return false;
    }
    if (checksum != cacheKey.checksum) {
      return false;
    }
    if (count != cacheKey.count) {
      return false;
    }

    //經過上面三個元素的篩選，說明這兩個對象極大可能相等，這里進行最后的判斷
    for (int i = 0; i < updateList.size(); i++) {
      Object thisObject = updateList.get(i);
      Object thatObject = cacheKey.updateList.get(i);
      if (!ArrayUtil.equals(thisObject, thatObject)) {
        return false;
      }
    }
    return true;
  }

  @Override
  public int hashCode() {
    return hashcode;
  }
}

二. 緩存的核心實現（PerpetualCache）

Mybatis 緩存實現是基于HashMap實現的，所以PerpetualCache實現了實際上是線程不安全的。如果想要實現線程安全的緩存，就需要使用SynchronizedCache裝飾器裝飾PerpetualCache。

/**
 * MyBatis緩存的核心實現類。類似于IO中的FileInputStream（JDK中的IO同樣采用裝飾器模式）
 * @author Clinton Begin
 */
public class PerpetualCache implements Cache {
  //每一個緩存都有一個唯一ID
  private final String id;

  //緩存底層使用HashMap保存數據，所以單獨的PerpetualCache實例是不支持多線程的，而decorators包中提供的裝飾器可以將其包裝為Blocking
  private final Map<Object, Object> cache = new HashMap<>();

  ...

  @Override
  public void putObject(Object key, Object value) {
    cache.put(key, value);
  }

  @Override
  public Object getObject(Object key) {
    return cache.get(key);
  }

  ...
}

三. 裝飾器

由于緩存模塊的裝飾器比較多，下面抽出幾個有代表性的裝飾器講解，其他的裝飾器原理可以參考博主的中文注釋項目：

3.1 BlockingCache

BlockingCache裝飾器是為了防止緩存擊穿，保證只有一個線程根據指定的key到數據庫中查找對應的數據，使用ConcurrentHashMap（安全，不重復）去存放鎖的key（按照key加鎖的細粒度鎖）。

/**
 * 裝飾器，讓緩存擁有阻塞的功能，目的是為了防止緩存擊穿。（當在緩存中找不到元素時，它設置對緩存鍵的鎖定，這樣，其他線程將一直等待，直到該緩存鍵放入了緩存值）
 *
 * 需要注意的是，這個裝飾器并不能保證緩存操作的線程安全
 *
 * Simple and inefficient version of EhCache's BlockingCache decorator.
 * It sets a lock over a cache key when the element is not found in cache.
 * This way, other threads will wait until this element is filled instead of hitting the database.
 *
 * @author Eduardo Macarron
 *
 */
public class BlockingCache implements Cache {

  //超時時間
  private long timeout;
  //被裝飾的對象
  private final Cache delegate;
  /**
   * 這里采用分段鎖，每一個Key對應一個鎖，當在緩存中找不到元素時，它設置對緩存鍵的鎖定。
   * 這樣，其他線程將一直等待，直到該緩存鍵放入了緩存值，這也是防止緩存擊穿的典型方案。
   *
   * 需要注意的是，這里每一個Key都對應一個鎖，就并不能保證對底層Map的更新操作（主要是put操作），
   * 只由一個線程執行，那這樣多線程狀態下對底層HashMap的更新操作也是線程不安全的！
   *
   * 也就是說，BlockingCache只是為了解決緩存擊穿的問題，而不是解決緩存操作的線程安全問題，
   * 線程安全問題交由SynchronizedCache裝飾器來完成
   *
   */
  private final ConcurrentHashMap<Object, ReentrantLock> locks;

  public BlockingCache(Cache delegate) {
    this.delegate = delegate;
    this.locks = new ConcurrentHashMap<>();
  }

  @Override
  public String getId() {
    return delegate.getId();
  }

  @Override
  public int getSize() {
    return delegate.getSize();
  }

  @Override
  public void putObject(Object key, Object value) {
    try {
      delegate.putObject(key, value);
    } finally {
      //釋放鎖
      releaseLock(key);
    }
  }

  @Override
  public Object getObject(Object key) {
    //嘗試獲取鎖，獲取到鎖之前一直阻塞，如果超時則拋出異常
    acquireLock(key);
    //調用被裝飾對象的getObject方法獲取緩存
    Object value = delegate.getObject(key);
    if (value != null) {
      /**
       * 如果value不為空，則釋放鎖。
       * 這里很多人肯定會有疑問，如果沒有獲取到值難道就不釋放鎖了嗎？其實不然，當
       * MyBatis獲取緩存時沒有獲取到數據，則會真正執行SQL語句去查詢數據庫，查詢到結果后
       * 會緊接著調用緩存的putObject方法，在這個方法中會進行釋放鎖的操作
       */
      releaseLock(key);
    }
    return value;
  }

  @Override
  public Object removeObject(Object key) {
    // despite of its name, this method is called only to release locks
    releaseLock(key);
    return null;
  }

  @Override
  public void clear() {
    delegate.clear();
  }

  private ReentrantLock getLockForKey(Object key) {
    //如果存在key對應的鎖則返回已經存在的鎖，如果不存在則創建一個并返回
    return locks.computeIfAbsent(key, k -> new ReentrantLock());
  }

  /**
   * 嘗試獲取鎖
   * @param key 緩存的key，
   */
  private void acquireLock(Object key) {
    //獲取鎖對象
    Lock lock = getLockForKey(key);
    if (timeout > 0) {
      //如果鎖的獲取擁有超時時間
      try {
        //則嘗試在指定時間內獲取鎖，如果獲取到了則返回true，超時仍未獲取到則返回false
        boolean acquired = lock.tryLock(timeout, TimeUnit.MILLISECONDS);
        if (!acquired) {
          throw new CacheException("Couldn't get a lock in " + timeout + " for the key " +  key + " at the cache " + delegate.getId());
        }
      } catch (InterruptedException e) {
        throw new CacheException("Got interrupted while trying to acquire lock for key " + key, e);
      }
    } else {
      //沒有設置超時時間，則直接無限期等待鎖
      lock.lock();
    }
  }

  private void releaseLock(Object key) {
    ReentrantLock lock = locks.get(key);
    if (lock.isHeldByCurrentThread()) {
      //如果當前線程擁有此鎖，則釋放鎖
      lock.unlock();
    }
  }

  public long getTimeout() {
    return timeout;
  }

  public void setTimeout(long timeout) {
    this.timeout = timeout;
  }
}

3.2 SynchronizedCache

/**
 * 同步裝飾器，用于保證緩存的更新操作是線程安全的
 */
public class SynchronizedCache implements Cache {

  private final Cache delegate;

  public SynchronizedCache(Cache delegate) {
    this.delegate = delegate;
  }

  @Override
  public String getId() {
    return delegate.getId();
  }

  @Override
  public synchronized int getSize() {
    return delegate.getSize();
  }

  @Override
  public synchronized void putObject(Object key, Object object) {
    delegate.putObject(key, object);
  }

  @Override
  public synchronized Object getObject(Object key) {
    return delegate.getObject(key);
  }

  @Override
  public synchronized Object removeObject(Object key) {
    return delegate.removeObject(key);
  }

  @Override
  public synchronized void clear() {
    delegate.clear();
  }

  @Override
  public int hashCode() {
    return delegate.hashCode();
  }

  @Override
  public boolean equals(Object obj) {
    return delegate.equals(obj);
  }

}

同步裝飾器，用于保證緩存的更新操作是線程安全的。它對所有涉及所有對緩存的操作（讀/寫）都使用synchronized關鍵字進行同步。需要注意的是SynchronizedCache裝飾器是將整個方法進行了加鎖，這在使用的時候就應該在保證線程安全的情況下，盡量的將這個裝飾器寫在內層，這樣可以減小鎖的范圍，增加系統吞吐量。例如在LoggingCache和SynchronizedCache裝飾器同時使用的情況下,LogingCache裝飾器最好放在SynchronizedCache裝飾器外層。

new LoggingCache( new SynchronizedCache( new PerpetualCache() ) )

3.3 FifoCache

該裝飾器用于使緩存具有先進先出淘汰機制，當緩存中緩存的數量到達閾值時，就會觸發緩存淘汰，而淘汰的目標就是最先進入緩存的對象。

/**
 * 緩存淘汰策略裝飾器（先進先出）
 *
 * @author Clinton Begin
 */
public class FifoCache implements Cache {

  /**
   * 被裝飾對象
   */
  private final Cache delegate;
  /**
   * 雙端隊列
   */
  private final Deque<Object> keyList;
  /**
   * 緩存的上限個數，觸發這個值就會激活緩存淘汰策略
   */
  private int size;

  public FifoCache(Cache delegate) {
    this.delegate = delegate;
    this.keyList = new LinkedList<>();
    this.size = 1024;
  }

  ...

  @Override
  public void putObject(Object key, Object value) {
    //將緩存鍵放入隊列中，如果隊列超長，則刪除隊首的鍵，以及其對應的緩存
    cycleKeyList(key);
    delegate.putObject(key, value);
  }

  ...

  /**
   * 將緩存鍵放入隊列中，如果隊列超長，則刪除隊首的鍵，以及其對應的緩存
   * @param key
   */
  private void cycleKeyList(Object key) {
    //在雙端隊列隊尾放入緩存鍵
    keyList.addLast(key);
    //如果緩存個數大于了極限值
    if (keyList.size() > size) {
      //移除雙端隊列對數的保存的Key
      Object oldestKey = keyList.removeFirst();
      //通過這個Key刪除隊首元素
      delegate.removeObject(oldestKey);
    }
  }

}

3.4 LruCache

該裝飾器用于使緩存具有LRU（最近最少使用）淘汰機制，當緩存中緩存的數量到達閾值時，就會觸發緩存淘汰，而淘汰的目標就是最近最少使用的對象。LruCache中是借助LinkedHashMap來實現LRU淘汰算法的，想要看到LruCache的源碼，就需要讀者們先去研究一下LinkedHashMap的源碼才能理解，這里面的淘汰策略是如何實現的.

/**
 * 緩存淘汰策略裝飾器（最近最少使用）
 *
 * @author Clinton Begin
 */
public class LruCache implements Cache {

  private final Cache delegate;
  //使用LinkedHashMap維護key的使用順序，個人認為這種實現方式雖然巧妙，但并不算特別優雅
  private Map<Object, Object> keyMap;
  private Object eldestKey;//最近最久沒有使用的Key

  public LruCache(Cache delegate) {
    this.delegate = delegate;
    setSize(1024);
  }

  ...
                        
  public void setSize(final int size) {
    //匿名內部類
    keyMap = new LinkedHashMap<Object, Object>(size, .75F, true) {
      private static final long serialVersionUID = 4267176411845948333L;

      /**
       * 重寫LinkedHashMap中的removeEldestEntry方法是實現LRU的核心，
       * 這里LRU的實現機制不清楚的話可以研究一下LinkedHashMap源碼。
       *
       * 大致思路：LinkedHashMap繼承于HashMap并在HashMap的基礎上為每一個Map.Entry添加了一個before和after指針，從而
       * 實現鏈表的功能，并全局維護了一個全局的head 和tail指針，分別指向鏈表的頭和尾。當LinkedHashMap put元素時，會將
       * 這個Entry加入鏈表的末尾。這樣鏈表頭始終指向的是最近最久沒有使用的元素，在put方法中會調用removeEldestEntry方法
       * 判斷是否刪除最久沒有使用的元素，如果返回true則會刪除最久沒有會用過的元素的元素。
       * @param eldest
       * @return
       */
      @Override
      protected boolean removeEldestEntry(Map.Entry<Object, Object> eldest) {
        boolean tooBig = size() > size;
        if (tooBig) {
          //保存最近最久沒有使用的key
          eldestKey = eldest.getKey();
        }
        return tooBig;
      }
    };
  }

  @Override
  public void putObject(Object key, Object value) {
    delegate.putObject(key, value);
    //增加元素后，進行key的后處理。判斷是否觸發緩存數極限，如果觸發了則清除最老的key對應的緩存
    cycleKeyList(key);
  }

  @Override
  public Object getObject(Object key) {
    //調用LinkedHashMap.get()方法是為了讓這次使用的key移動到鏈表末尾
    keyMap.get(key); // touch
    return delegate.getObject(key);
  }

  private void cycleKeyList(Object key) {
    //將key放入keyMap，如果緩存數大于了極限值，則會刪除最老Key，并將這個key賦給eldestKey變量
    keyMap.put(key, key);
    if (eldestKey != null) {
      //eldestKey不為空，說明在調用keyMap.put()中觸發了刪除最老元素的機制，此時需要將真實緩存中對應的key-value刪除
      delegate.removeObject(eldestKey);
      eldestKey = null;
    }
  }
}

/**
   * 將連接對象歸還給連接池（實際上是將連接從active隊列中移到idle隊列中）
   *
   * @param conn
   * @throws SQLException
   */
  protected void pushConnection(PooledConnection conn) throws SQLException {

    synchronized (state) {
      //將當前連接從active隊列中移除
      state.activeConnections.remove(conn);
      if (conn.isValid()) {
        //連接是有效的
        if (state.idleConnections.size() < poolMaximumIdleConnections && conn.getConnectionTypeCode() == expectedConnectionTypeCode) {
          //空閑的數量小于最大空閑值 且 連接對象的typeCode與數據源期望的TypeCode相同
          //記錄當前連接使用的時間
          state.accumulatedCheckoutTime += conn.getCheckoutTime();
          if (!conn.getRealConnection().getAutoCommit()) {
            //如果連接對象的事務是非自動提交的，則回滾事務
            conn.getRealConnection().rollback();
          }
          //重新封裝一個新的代理連接對象
          PooledConnection newConn = new PooledConnection(conn.getRealConnection(), this);
          //將新建的代理連接對象放入空閑隊列
          state.idleConnections.add(newConn);
          //設置創建的時間戳
          newConn.setCreatedTimestamp(conn.getCreatedTimestamp());
          //設置最后使用的時間戳
          newConn.setLastUsedTimestamp(conn.getLastUsedTimestamp());
          //老的代理對象作廢
          conn.invalidate();
          if (log.isDebugEnabled()) {
            log.debug("Returned connection " + newConn.getRealHashCode() + " to pool.");
          }
          //換新在state鎖上的等待的線程
          state.notifyAll();
        } else {
          //空閑的數量大于等于最大空閑值 或者 連接對象的typeCode與數據源期望的TypeCode不相同
          state.accumulatedCheckoutTime += conn.getCheckoutTime();
          if (!conn.getRealConnection().getAutoCommit()) {
            conn.getRealConnection().rollback();
          }
          //關閉這個連接
          conn.getRealConnection().close();
          if (log.isDebugEnabled()) {
            log.debug("Closed connection " + conn.getRealHashCode() + ".");
          }
          //將代理連接作廢
          conn.invalidate();
        }
      } else {
        if (log.isDebugEnabled()) {
          log.debug("A bad connection (" + conn.getRealHashCode() + ") attempted to return to the pool, discarding connection.");
        }
        state.badConnectionCount++;
      }
    }
  }

3.5 其他裝飾器

代碼注釋請移步至：GitHub或Gitee

posted @ 2020-04-30 15:28 聽到微笑閱讀(33) 評論(0) 收藏舉報來源

刷新頁面返回頂部

聽到微笑的博客