[HBase] 写入流程之客户端处理阶段

1.前言

HBase服务器端并没有提供update,delete接口，所以这些操作在服务器端都被认作是写入操作。因此HBase中更新，删除操作的流程与写入流程完全一致。那下面就以put操作为例，进行写入流程的分析。

2.HTable类

2.1 HTable的put方法：

  @Override
  public void put(final Put put) throws IOException {
    getBufferedMutator().mutate(put);
    if (autoFlush) {
      flushCommits();
    }
  }

  @Override
  public void put(final List<Put> puts) throws IOException {
    getBufferedMutator().mutate(puts);
    if (autoFlush) {
      flushCommits();
    }
  }

<1> getBufferedMutator().mutate(put);

  BufferedMutator getBufferedMutator() throws IOException {
    // mutator:写缓存
    if (mutator == null) {
      this.mutator = (BufferedMutatorImpl) connection.getBufferedMutator(
          //创建mutator实例,传入tableName,线程池,写的buffer大小,最大的keyValue数量  
          new BufferedMutatorParams(tableName)
              .pool(pool)
              .writeBufferSize(tableConfiguration.getWriteBufferSize())
              .maxKeyValueSize(tableConfiguration.getMaxKeyValueSize())
      );
    }
    return mutator;
  }

connection.getBufferedMutator实际调用ConnectionManager的getBufferedMutator方法，最终返回new BufferedMutatorImpl构造方法构造。

这里的maxKeyValueSize是单个Cell的大小

在BufferedMutatorImpl中：

  protected ClusterConnection connection; // non-final so can be overridden in test
  private final TableName tableName;
  private volatile Configuration conf;
  @VisibleForTesting
  final ConcurrentLinkedQueue<Mutation> writeAsyncBuffer = new ConcurrentLinkedQueue<Mutation>();
  @VisibleForTesting
  AtomicLong currentWriteBufferSize = new AtomicLong(0);

  private long writeBufferSize;
  private final int maxKeyValueSize;
  private boolean closed = false;
  private final ExecutorService pool;

  @VisibleForTesting
  protected AsyncProcess ap; // non-final so can be overridden in test

调用mutate方法传入put请求，实际上就是将put请求放入writeAsyncBuffer中

   long toAddSize = 0;
    for (Mutation m : ms) {
      //如果是put请求,校验写入的每个Cell的大小是否超过限制
      if (m instanceof Put) {
        validatePut((Put) m);
      }
      //累加当前请求占据的堆内存大小
      toAddSize += m.heapSize();
    }

    // This behavior is highly non-intuitive... it does not protect us against
    // 94-incompatible behavior, which is a timing issue because hasError, the below code
    // and setter of hasError are not synchronized. Perhaps it should be removed.
    //异步提交请求发生错误
    if (ap.hasError()) {
      //累加当前的请求的大小(AtomicLong类型,具有原子性)
      currentWriteBufferSize.addAndGet(toAddSize);
      //将该请求放在缓存中(ConcurrentLinkedQueue,同步且有顺序)
      writeAsyncBuffer.addAll(ms);
      //不管当前的currentWriteBufferSize是否达到阈值,直接flush,并且同步等待结果返回
      backgroundFlushCommits(true);
    } else {
      currentWriteBufferSize.addAndGet(toAddSize);
      writeAsyncBuffer.addAll(ms);
    }

    // Now try and queue what needs to be queued.
    while (currentWriteBufferSize.get() > writeBufferSize) {
      backgroundFlushCommits(false);
    }

总结：put中，
先getBufferedMutator().mutate(put);
这个方法中在缓存达到阈值时，通过AsyncProcess异步提交（如果异步提交进程初始化异常，那么就会转为同步提交，且不必等待达到阈值）
if (autoFlush) { flushCommits(); }这个条件分支：
如果autoFlush为true，不管缓存是否达到阈值，都会直接触发AsyncProcess异步提交

2.2 backgroundFlushCommits方法

当传入backgroudFlushCommits的参数为false时执行的是异步提交，参数为true时执行的是同步提交。
极少数情况（异步提交发生异常才会转为同步）：

与此同时，可以发现无论异步提交还是同步提交，实际的提交动作是由AsyncProcess ap执行的：
其中最关键的是ap.submit(tableName, buffer, true, null, false);

locateRegion方法这里就先不赘述了，之后会写一篇专门来分析。

放到Map<ServerName, MultiAction<Row>> actionsByServer中

接下里就是多线程的RPC提交：
AsyncProcess中的submit ---》submitMultiActions ---》sendMultiAction ---》

分RegionServer对相应Actions创建多线程

---》getNewMultiActionRunnable

最终每个RegionServer一个线程

---》SingleServerRequestRunnable类中的run方法

3.总结

（1）把put操作添加到writeAsyncBuffer队列里面，符合条件（自动flush或者超过了阀值writeBufferSize）就通过AsyncProcess异步批量提交。

（2）在提交之前，我们要根据每个rowkey找到它们归属的region server，这个定位的过程是通过HConnection的locateRegion方法获得的，然后再把这些rowkey按照HRegionLocation分组。

（3）通过多线程，一个HRegionLocation构造MultiServerCallable<Row>，然后通过rpcCallerFactory.<MultiResponse> newCaller()执行调用，忽略掉失败重新提交和错误处理，客户端的提交操作到此结束。

相关博客：HBase的put流程源码分析

[HBase] 写入流程之客户端处理阶段