首页 > 代码库 > hbase源码系列（二）HTable 如何访问客户端

hbase源码系列（二）HTable 如何访问客户端

2024-07-01 21:17:35   222人阅读

　　hbase的源码终于搞一个段落了，在接下来的一个月，着重于把看过的源码提炼一下，对一些有意思的主题进行分享一下。继上一篇讲了负载均衡之后，这一篇我们从client开始讲吧，从client到master再到region server，按照这个顺序来开展，网友也可以对自己感兴趣的部分给我留言或者直接联系我的QQ。

　　现在我们讲一下HTable吧，为什么讲HTable，因为这是我们最常见的一个类，这是我们对hbase中数据的操作的入口。

　　

1.Put操作

　　下面是一个很简单往hbase插入一条记录的例子。

HBaseConfiguration conf =  (HBaseConfiguration) HBaseConfiguration.create();
byte[] rowkey = Bytes.toBytes("cenyuhai");
byte[] family = Bytes.toBytes("f");
byte[] qualifier = Bytes.toBytes("name");
byte[] value = http://www.mamicode.com/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()执行调用，忽略掉失败重新提交和错误处理，客户端的提交操作到此结束。

　　

2.Delete操作

　　对于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方法。

3.Get操作

　　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);
    }
}

 

 4.批量操作

　　

　　针对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的处理的文章，敬请期待。

 

 

 

 

 

 

　　

 


联系
我们