HBase1.0.0源码分析之请求处理流程分析以Put操作为例（二）

HBase1.0.0源码分析之请求处理流程分析以Put操作为例（二）

1.通过mutate(put)操作，将单个put操作添加到缓冲操作中，这些缓冲操作其实就是Put的父类的一个List的集合。如下：

    private List<Row> writeAsyncBuffer = new LinkedList<>();
    writeAsyncBuffer.add(m);

当writeAsyncBuffer满了之后或者是人为的调用backgroundFlushCommits操作促使缓冲池中的操作被执行。

2.backgroundFlushCommits(boolean synchronous)执行操作缓冲池中的操作，其实他也并不是自己去处理响应的操作，而是委托给一个AsyncProcess具体进行响应操作的执行，该类是模拟一个异步处理持续请求流的类。这其中主要发生以下几个主要操作：

    ap.submit(tableName, writeAsyncBuffer, true, null, false);

    Map<ServerName, MultiAction<Row>> actionsByServer =
        new HashMap<ServerName, MultiAction<Row>>();
   List<Action<Row>> retainedActions = new  ArrayList<Action<Row>>(rows.size());

   addAction(loc.getServerName(), regionName, action, actionsByServer, nonceGroup

    return submitMultiActions(tableName, retainedActions, nonceGroup, callback, null, needResults,
      locationErrors, locationErrorRows, actionsByServer, pool);

总结以下这里主要干了以下几件事：
* 对Put操作进行封装，封装成Action
* 找出每个操作对应的regionServer形成ServerName - > MultiAction的键值对，然后继续submit
3. 在submitMultiActions方法里面，作者使用了一个AsyncRequestFutureImpl的实现来保存结果数据

 AsyncRequestFutureImpl<CResult> ars = createAsyncRequestFuture(
      tableName, retainedActions, nonceGroup, pool, callback, results, needResults);

4.类ars中的sendMultiAction函数sendMultiAction才是最后真正的逻辑层调用的地方，完整的函数如下：

    private void sendMultiAction(Map<ServerName, MultiAction<Row>> actionsByServer,
        int numAttempt, List<Action<Row>> actionsForReplicaThread, boolean reuseThread) {
      // Run the last item on the same thread if we are already on a send thread.
      // We hope most of the time it will be the only item, so we can cut down on threads.
      int actionsRemaining = actionsByServer.size();
      // This iteration is by server (the HRegionLocation comparator is by server portion only).
      for (Map.Entry<ServerName, MultiAction<Row>> e : actionsByServer.entrySet()) {
        ServerName server = e.getKey();
        MultiAction<Row> multiAction = e.getValue();
        incTaskCounters(multiAction.getRegions(), server);
        Collection<? extends Runnable> runnables = getNewMultiActionRunnable(server, multiAction,
            numAttempt);
        // make sure we correctly count the number of runnables before we try to reuse the send
        // thread, in case we had to split the request into different runnables because of backoff
        if (runnables.size() > actionsRemaining) {
          actionsRemaining = runnables.size();
        }

        // run all the runnables
        for (Runnable runnable : runnables) {
          if ((--actionsRemaining == 0) && reuseThread) {
            runnable.run();
          } else {
            try {
              pool.submit(runnable);
            } catch (RejectedExecutionException ree) {
              // This should never happen. But as the pool is provided by the end user, let‘s secure
              //  this a little.
              decTaskCounters(multiAction.getRegions(), server);
              LOG.warn("#" + id + ", the task was rejected by the pool. This is unexpected." +
                  " Server is " + server.getServerName(), ree);
              // We‘re likely to fail again, but this will increment the attempt counter, so it will
              //  finish.
              receiveGlobalFailure(multiAction, server, numAttempt, ree);
            }
          }
        }
      }

      if (actionsForReplicaThread != null) {
        startWaitingForReplicaCalls(actionsForReplicaThread);
      }
    }

函数蛮长，但是做的事情的逻辑也是比较清晰的，从代码中可以看出，程序的处理逻辑是按照regionServer进行区分的，将每个操作封装成可运行的任务（SingleServerRequestRunnable），然后用现成池pool依次执行。
这里有必要将他们所构造的任务的run 函数贴出来研究以下，从run函数我们才能够看到请求的真正逻辑：

    MultiResponse res;
    MultiServerCallable<Row> callable = createCallable(server, tableName, multiAction);
    res = createCaller(callable).callWithoutRetries(callable, timeout);
    receiveMultiAction(multiAction, server, res, numAttempt);

每个任务都是通过HBase的RPC框架与服务器进行通信，并获取返回的结果。至此，客户端的Put操作的流程也就结束了，至于具体的RPC端如何执行相关内容，后续博客继续关注。

总结

1.客户端有个缓存机制，批量处理
2.默认的Put的处理是异步进行的
3.Put最后是按照RegionServer分别RPC处理的