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处理的