HBase中MVCC的实现机制及应用介绍

本帖最后由 howtodown 于 2014-11-11 00:19 编辑

问题导读
1.什么是MVCC
2.MVCC基本原理的基本原理是什么？
3.hbase中通过哪个方法使用MVCC进行写操作？

MVCC（Multi-Version Concurrent Control），即多版本并发控制协议，广泛使用于数据库系统。本文将介绍HBase中对于MVCC的实现及应用情况。

MVCC基本原理

在介绍MVCC概念之前，我们先来想一下数据库系统里的一个问题：假设有多个用户同时读写数据库里的一行记录，那么怎么保证数据的一致性呢？一个基本的解决方法是对这一行记录加上一把锁，将不同用户对同一行记录的读写操作完全串行化执行，由于同一时刻只有一个用户在操作，因此一致性不存在问题。但是，它存在明显的性能问题：读会阻塞写，写也会阻塞读，整个数据库系统的并发性能将大打折扣。

MVCC（Multi-Version Concurrent Control），即多版本并发控制协议，它的目标是在保证数据一致性的前提下，提供一种高并发的访问性能。在MVCC协议中，每个用户在连接数据库时看到的是一个具有一致性状态的镜像，每个事务在提交到数据库之前对其他用户均是不可见的。当事务需要更新数据时，不会直接覆盖以前的数据，而是生成一个新的版本的数据，因此一条数据会有多个版本存储，但是同一时刻只有最新的版本号是有效的。因此，读的时候就可以保证总是以当前时刻的版本的数据可以被读到，不论这条数据后来是否被修改或删除。

更多关于MVCC基本思想的介绍，参考Wikipedia。

一个MVCC实现类

见org.apache.hadoop.hbase.regionserver.MultiVersionConsistencyControl，用于控制Memstore中读写的一致性，其中维护两个long型的变量：

1）memstoreRead：用于记录当前全局可读的readPoint，同时为了每个客户端读请求能够记录自己发起请求时刻的readPoint，还有一个ThreadLocal的perThreadReadPoint变量，以及相关的set和get方法；

2）memstoreWrite：用于记录当前全局最大的writePoint，根据它为下个事务生成新的writePoint。

MultiVersionConsistencyControl中关键的实现方法如下：

1）WriteEntry beginMemstoreInsert()：开始一个更新操作，将memstoreWrite加1，创建writeQueue并插入到writeQueue，并返回WriteEntry对象；

2）void completeMemstoreInsert(WriteEntry e)：完成当前更新操作，将WriteEntry对象标记为可读，具体分两步：

boolean advanceMemstore(WriteEntry e)：从头开始遍历writeQueue，移除所有已完成的WriteEntry对象，最后将memstoreRead更新为最新已完成的memstoreWrite；
void waitForRead(WriteEntry e)：阻塞当前线程，直到memstoreRead等于当前WriteEntry的memstoreWrite，至此表明当前WriteEntry之前的所有更新事务都已经完成。

MVCC使用场景

见org.apache.hadoop.hbase.regionserver.HRegion.java，每个Region包含一个Memstore，维护一个MultiVersionConsistencyControl对象。

写操作

见HRegion.java中的以下写操作的方法：

1）put

2）checkAndPut

3）delete

4）checkAndDelete

5）internalFlushcache

6）mutateRow

7）mutateRowsWithLocks

8）batchMutate

最终会调用到applyFamilyMapToMemstore方法使用MVCC进行写操作：

/**
   * Atomically apply the given map of family->edits to the memstore.
   * This handles the consistency control on its own, but the caller
   * should already have locked updatesLock.readLock(). This also does
   * <b>not</b> check the families for validity.
   *
   * @param familyMap Map of kvs per family
   * @param localizedWriteEntry The WriteEntry of the MVCC for this transaction.
   *        If null, then this method internally creates a mvcc transaction.
   * @return the additional memory usage of the memstore caused by the
   * new entries.
   */
  private long applyFamilyMapToMemstore(Map<byte[], List<KeyValue>> familyMap,
    MultiVersionConsistencyControl.WriteEntry localizedWriteEntry) {
    long size = 0;
    boolean freemvcc = false;

    try {
      if (localizedWriteEntry == null) {
        localizedWriteEntry = mvcc.beginMemstoreInsert();
        freemvcc = true;
      }

      for (Map.Entry<byte[], List<KeyValue>> e : familyMap.entrySet()) {
        byte[] family = e.getKey();
        List<KeyValue> edits = e.getValue();

        Store store = getStore(family);
        for (KeyValue kv: edits) {
          kv.setMemstoreTS(localizedWriteEntry.getWriteNumber());
          size += store.add(kv);
        }
      }
    } finally {
      if (freemvcc) {
        mvcc.completeMemstoreInsert(localizedWriteEntry);
      }
    }

     return size;
   }
复制代码

读操作

HRegion.java中通过private ConcurrentHashMap<RegionScanner, Long> scannerReadPoints;维护各个查询请求的readPoint。

以get或scan请求为例，最终会通过getScanner方法需要构造RegionScannerImpl对象：

org.apache.hadoop.hbase.regionserver.HRegion.RegionScannerImpl：

1）根据Scan对象构造时设置好readPoint，scan.getIsolationLevel()分为READ_UNCOMMITTED和READ_COMMITTED，只有当READ_COMMITTED时根据MultiVersionConsistencyControl.resetThreadReadPoint(mvcc);设置当前scanner线程的readPoint，并插入到scannerReadPoints维护起来。

2）根据scan需要读取的column family，创建StoreScanner（根据bloom filter、time range、ttl筛选需要的MemStoreScanner和StoreFileScanner），添加到scanners中，并最终根据scanners构造出一个KeyValueHeap。

下面看下RegionScannerImpl中的next方法是每次查询时需要调用的函数：

boolean org.apache.hadoop.hbase.regionserver.HRegion.RegionScannerImpl.next(List<KeyValue> outResults, int limit) throwsIOException

而上述方法会通过KeyValueHeap的next方法读取下一条数据：先定位到当前KeyValueScanner（即之前构造KeyValueHeap时传入的MemStoreScanner或StoreScanner），然后调用next方法。

StoreFileScanner和MemStoreScanner均为KeyValueScanner，通过其中的next()接口方法，分别调用到StoreFileScanner.java的skipKVsNewerThanReadpoint方法、Memstore.java中MemStoreScanner对象的getNext方法。

1）StoreFileScanner.java的skipKVsNewerThanReadpoint方法：

protected boolean skipKVsNewerThanReadpoint() throws IOException {
    long readPoint = MultiVersionConsistencyControl.getThreadReadPoint();

    // We want to ignore all key-values that are newer than our current
    // readPoint
    while(enforceMVCC
        && cur != null
        && (cur.getMemstoreTS() > readPoint)) {
      hfs.next();
      cur = hfs.getKeyValue();
    }

    if (cur == null) {
      close();
      return false;
    }

    // For the optimisation in HBASE-4346, we set the KV's memstoreTS to
    // 0, if it is older than all the scanners' read points. It is possible
    // that a newer KV's memstoreTS was reset to 0. But, there is an
    // older KV which was not reset to 0 (because it was
    // not old enough during flush). Make sure that we set it correctly now,
    // so that the comparision order does not change.
    if (cur.getMemstoreTS() <= readPoint) {
      cur.setMemstoreTS(0);
    }
    return true;
  }
复制代码

2） Memstore.java中MemStoreScanner对象的getNext方法：

protected KeyValue getNext(Iterator<KeyValue> it) {
      long readPoint = MultiVersionConsistencyControl.getThreadReadPoint();
    
      while (it.hasNext()) {
        KeyValue v = it.next();
        if (v.getMemstoreTS() <= readPoint) {
          return v;
        }
      }

      return null;
    }
复制代码