hbase的客户端代码并不想hive一样用java编写,shell调用,而是使用ruby编写。
在admin.rb文件中方法create,其中接受两个参数,其中第二个参数类型为变长参数。
而在create方法的最后,调用了admin.createTable,其中的admin是hbaes.rb初始化时通过调用java代码ConnectionFactory.createConnection创建的connection调用getAdmin而获得的。
下面简单分析一下ConnectionFactory.createConnection流程。
默认的hbase.client.connection.impl实现类ConnectionImplementation.class,因此,该方法其实就相当于初始化了ConnectionImplementation。而在ConnectionImplementation中,最主要的构建了类型为NettyRpcClient的rpc客户端。
接着,根据源码我们可以发现,然后调用了HBaseAdmin.createTable。在该方法调用的时候,有一些hbase通用的架构,我们接下来一一道来。
首先,调用createTableAsync方法,其中构建了一个MasterCallable类型的匿名对象,其复写的rpcCall方法真正的调用了客户端方法。
接下来调用executeCallable方法,然后构建RpcRetryingCaller对象,并调用该对象的callWithRetries方法。在其唯一实现RpcRetryingCallerImpl中我们可以看到
首先调用了传入的callable.prepare方法。由于此时我们分析的callable类型为MasterCallable,因此,我们可以追踪到MasterCallable.prepare方法。在这里,调用了ConnectionImplementation.getMaster方法。接着调用了ConnectionImplementation.getKeepAliveMasterService。接下来,返回rpc调用的本地stub。
然后调用了callable.call方法,而这个call方法最后恰恰调用了匿名对象复写的rpcCall方法。也就是说,他调用了本地stub的createTable方法。
而接下来的调用流程就正如我在上篇博文中所讲的。会调用BlockingRpcChannelImplementation.callBlockingMethod,AbstractRpcClient.callBlockingMethod,AbstractRpcClient.callMethod,NettyRpcConnection.sendRequest,HBaseRpcControllerImpl.notifyOnCancel等一系列方法。
在NettyRpcConnection.sendRequest方法中,我们将着重进行分析。以纠正之前所犯的错误。
在这里首先执行了connect方法,如下图所示,我们可以发现,这里添加了一个ChannelFutureListener。
通过operationComplete里面的established方法,我们可以看到,通道的pipeline中添加了NettyRpcDuplexHandler。
然后执行了write方法,ch.writeAndFlush,学过Netty大家都清楚,下一步就会调用刚刚加入的NettyRpcDuplexHandler.write方法。然后就调用该方法,向服务端发送信息。
接着,等待服务端的返回。
服务端接收到客户端后(具体流程可以参考我的上一篇博文[Hbase之rpc调用流程简介]),将响应返回。
并调用下图所示的readResponse方法。
(而在此之前,在AbstractRpcClient.callBlockingMethod的方法中BlockingRpcCallback.get方法已经开始调用this.wait()。
在BlockingRpcConnection.run方法中,会调用readResponse。(在客户端的实现为BlockingRpcConnection,才会调用。)。
而我们都知道,在调用的实际过程中,hbase的默认客户端实现是NettyRpcConnection。
而在readResponse方法中,类似hadoop中rpc的阻塞一样,调用in.readInt,也就是说等待到服务端的返回后,该方法会继续向下执行。一直到call.setResponse,接着就是call.callComplete,)
在readResponse方法的最后,我们可以看到调用了call.setResponse,接着就是callComplete, callback.run,而这里的callback恰恰就是下图中的匿名对象。
接着呢,就是调用BlockingRpcCallback.run方法。调用this.notify.。然后将AbstractRpcClient.callBlockingMethod中的阻塞打开,获得server端的返回值。
当然,这只是获得了createTable的服务端返回值。接下来会创建CreateTableFuture对象,其中封装了刚刚获得的服务端返回值。
而接下来会继续调用到ProcedureFuture.get(long timeout, TimeUnit unit)方法。在该方法内部,会继续调用waitProcedureResult,getProcedureResult等等一系列方法。其流程与上面所叙述的大体一致,我们就不在这里一一介绍了。所不同的是,这里调用的方法是getProcedureResult。
hbase createTable的流程答题时这样,如果感觉对你的理解有帮助,欢迎你的赞赏,如果解答不了你的疑问,可以发送邮件至[email protected],期待你的来信。
你的赞赏是我前进的动力。