标签：des class blog img com div code java javascript http size
hbase的源码终于搞一个段落了，在接下来的一个月，着重于把看过的源码提炼一下，对一些有意思的主题进行分享一下。继上一篇讲了负载均衡之后，这一篇我们从client开始讲吧，从client到master再到region server，按照这个顺序来开展，网友也可以对自己感兴趣的部分给我留言或者直接联系我的QQ。
现在我们讲一下HTable吧，为什么讲HTable，因为这是我们最常见的一个类，这是我们对hbase中数据的操作的入口。
下面是一个很简单往hbase插入一条记录的例子。
HBaseConfiguration conf = (HBaseConfiguration) HBaseConfiguration.create(); byte[] rowkey = Bytes.toBytes("cenyuhai"); byte[] family = Bytes.toBytes("f"); byte[] qualifier = Bytes.toBytes("name"); byte[] value = Bytes.toBytes("岑玉海"); HTable table = new HTable(conf, "test"); Put put = new Put(rowkey); put.add(family,qualifier,value); table.put(put);
我们平常就是采用这种方式提交的数据，为了提高重用性采用HTablePool，最新的API推荐使用HConnection.getTable("test")来获得HTable，旧的HTablePool已经被抛弃了。好，我们下面开始看看HTable内部是如何实现的吧，首先我们看看它内部有什么属性。
/** 实际提交数据所用的类 */
protected HConnection connection;/** 需要提交的数据的列表 */ protected List<Row> writeAsyncBuffer = new LinkedList<Row>();
/** flush的size */ private long writeBufferSize; /** 是否自动flush */ private boolean autoFlush; /** 当前的数据的size，达到指定的size就要提交 */ protected long currentWriteBufferSize; protected int scannerCaching; private int maxKeyValueSize; private ExecutorService pool; // For Multi
/** 异步提交 */ protected AsyncProcess<Object> ap;
** rpc工厂 */ private RpcRetryingCallerFactory rpcCallerFactory;
主要是靠上面的这些家伙来干活的，这里面的connection、ap、rpcCallerFactory是用来和后台通信的，HTable只是做一个操作，数据进来之后，添加到writeAsyncBuffer，满足条件就flush。
下面看看table.put是怎么执行的：
doPut(put); if (autoFlush) { flushCommits(); }
执行put操作，如果是autoFush，就提交，先看doPut的过程，如果之前的ap异步提交到有问题，就先进行后台提交，不过这次是同步的，如果没有错误，就把put添加到队列当中，然后检查一下当前的 buffer的大小，超过我们设置的内容的时候，就flush掉。
if (ap.hasError()){ backgroundFlushCommits(true); } currentWriteBufferSize += put.heapSize(); writeAsyncBuffer.add(put); while (currentWriteBufferSize > writeBufferSize) { backgroundFlushCommits(false); }
写下来，让我们看看backgroundFlushCommits这个方法吧，它的核心就这么一句ap.submit(writeAsyncBuffer, true) ，如果出错了的话，就报错了。所以网上所有关于客户端调优的方法里面无非就这么几种:
1)关闭autoFlush
2)关闭wal日志
3)把writeBufferSize设大一点，一般说是设置成5MB
经过实践，就第二条关闭日志的效果比较明显，其它的效果都不明显，因为提交的过程是异步的，所以提交的时候占用的时间并不多，提交到server端后，server还有一个写入的队列，(⊙o⊙)… 让人想起小米手机那恶心的排队了。。。所以大规模写入数据，别指望着用put来解决。。。mapreduce生成hfile，然后用bulk load的方式比较好。
不废话了，我们继续追踪ap.submit方法吧，F3进去。
int posInList = -1; Iterator<? extends Row> it = rows.iterator(); while (it.hasNext()) { Row r = it.next(); //为row定位 HRegionLocation loc = findDestLocation(r, 1, posInList); if (loc != null && canTakeOperation(loc, regionIncluded, serverIncluded)) { // loc is null if there is an error such as meta not available. Action<Row> action = new Action<Row>(r, ++posInList); retainedActions.add(action); addAction(loc, action, actionsByServer); it.remove(); } }
循环遍历r，为每个r找到它的位置loc，loc是HRegionLocation，里面记录着这行记录所在的目标region所在的位置，loc怎么获得呢，走进findDestLocation方法里面，看到了这么一句。
loc = hConnection.locateRegion(this.tableName, row.getRow());
通过表名和rowkey，使用HConnection就可以定位到它的位置，这里就先不讲定位了，稍后放一节出来讲，否则篇幅太长了，这里我们只需要记住，提交操作，是要知道它对应的region在哪里的。
定位到它的位置之后，它把loc添加到了actionsByServer，一个region server对应一组操作。（插句题外话为什么这里叫action呢，其实我们熟知的Put、Delete，以及不常用的Append、Increment都是继承自Row的，在接口传递时候，其实都是视为一种操作，到了后台之后，才做区分）。
接下来，就是多线程的rpc提交了。
MultiServerCallable<Row> callable = createCallable(loc, multiAction);
......
res = createCaller(callable).callWithoutRetries(callable);
再深挖一点，把它们的实现都扒出来吧。
protected MultiServerCallable<Row> createCallable(final HRegionLocation location, final MultiAction<Row> multi) { return new MultiServerCallable<Row>(hConnection, tableName, location, multi); } protected RpcRetryingCaller<MultiResponse> createCaller(MultiServerCallable<Row> callable) { return rpcCallerFactory.<MultiResponse> newCaller(); }
ok，看到了，先构造一个MultiServerCallable，然后再通过rpcCallerFactory做最后的call操作。
好了，到这里再总结一下put操作吧，前面写得有点儿凌乱了。
（1）把put操作添加到writeAsyncBuffer队列里面，符合条件（自动flush或者超过了阀值writeBufferSize）就通过AsyncProcess异步批量提交。
（2）在提交之前，我们要根据每个rowkey找到它们归属的region server，这个定位的过程是通过HConnection的locateRegion方法获得的，然后再把这些rowkey按照HRegionLocation分组。
（3）通过多线程，一个HRegionLocation构造MultiServerCallable<Row>，然后通过rpcCallerFactory.<MultiResponse> newCaller()执行调用，忽略掉失败重新提交和错误处理，客户端的提交操作到此结束。
对于Delete，我们也可以通过以下代码执行一个delete操作
Delete del = new Delete(rowkey); table.delete(del);
这个操作比较干脆，new一个RegionServerCallable<Boolean>,直接走rpc了，爽快啊。
RegionServerCallable<Boolean> callable = new RegionServerCallable<Boolean>(connection, tableName, delete.getRow()) { public Boolean call() throws IOException { try { MutateRequest request = RequestConverter.buildMutateRequest( getLocation().getRegionInfo().getRegionName(), delete); MutateResponse response = getStub().mutate(null, request); return Boolean.valueOf(response.getProcessed()); } catch (ServiceException se) { throw ProtobufUtil.getRemoteException(se); } } }; rpcCallerFactory.<Boolean> newCaller().callWithRetries(callable, this.operationTimeout);
这里面注意一下这行MutateResponse response = getStub().mutate(null, request);
getStub()返回的是一个ClientService.BlockingInterface接口，实现这个接口的类是HRegionServer，这样子我们就知道它在服务端执行了HRegionServer里面的mutate方法。
get操作也和delete一样简单
Get get = new Get(rowkey); Result row = table.get(get);
get操作也没几行代码，还是直接走的rpc
public Result get(final Get get) throws IOException { RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection, getName(), get.getRow()) { public Result call() throws IOException { return ProtobufUtil.get(getStub(), getLocation().getRegionInfo().getRegionName(), get); } }; return rpcCallerFactory.<Result> newCaller().callWithRetries(callable, this.operationTimeout);
}
注意里面的ProtobufUtil.get操作，它其实是构建了一个GetRequest，需要的参数是regionName和get，然后走HRegionServer的get方法，返回一个GetResponse
public static Result get(final ClientService.BlockingInterface client, final byte[] regionName, final Get get) throws IOException { GetRequest request = RequestConverter.buildGetRequest(regionName, get); try { GetResponse response = client.get(null, request); if (response == null) return null; return toResult(response.getResult()); } catch (ServiceException se) { throw getRemoteException(se); }
}
针对put、delete、get都有相应的操作的方式：
1.Put(list)操作，很多童鞋以为这个可以提高写入速度，其实无效。。。为啥？因为你构造了一个list进去，它再遍历一下list，执行doPut操作。。。。反而还慢点。
2.delete和get的批量操作走的都是connection.processBatchCallback(actions, tableName, pool, results, callback)，具体的实现在HConnectionManager的静态类HConnectionImplementation里面，结果我们惊人的发现：
AsyncProcess<?> asyncProcess = createAsyncProcess(tableName, pool, cb, conf); asyncProcess.submitAll(list); asyncProcess.waitUntilDone();
它走的还是put一样的操作，既然是一样的，何苦代码写得那么绕呢？
okay，HTable到现在就告一段落了，后续会有在Region Server这端关于put、delete、get的处理的文章，敬请期待。
hbase源码系列（二）HTable 如何访问客户端,码迷,mamicode.com
标签：des class blog img com div code java javascript http size
原文地址：http://www.cnblogs.com/cenyuhai/p/3694725.html
