Hbase 批量提交源码分析

桃扇骨 2023-05-30 08:53 5阅读 0赞

### 文章目录 ###

*  Hbase batch()方法源码学习
 *   *  背景
     *   *  hbase版本
     *  方法介绍
     *   *   *  AsyncRequestFutureImpl类的组成结构
             *   *  参考链接

# Hbase batch()方法源码学习 #

## 背景 ##

> 项目中，常常会向hbase中存储东西，put()也是会向hbase中存储，但是batch可以批处理方式进行存储，效率更快，所以自己决定学习和分析下batch()源码

### hbase版本 ###

> hbase1.3.1, 不同版本可能会有些许差距

## 方法介绍 ##

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-snDAxomo-1573199896561)(13CCA816219641719762E11945FECCB7)\]

在Table.java的接口文件中具体源码如下:

void batch(final List<? extends Row> actions, final Object[] results) throws IOException,
        InterruptedException;

因为接口文件中没有具体实现，跳转到HTable.java文件中的batch()方法， 具体代码如下:

/**
       * {@inheritDoc}
       */
      @Override
      public void batch(final List<? extends Row> actions, final Object[] results)
          throws InterruptedException, IOException {
        AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, actions, null, results);
        ars.waitUntilDone();
        if (ars.hasError()) {
          throw ars.getErrors();
        }
      }

其中:

1.  multiAp --是一个异步操作进程

> 表示批量提交操作在hbase中是异步进行的

protected AsyncProcess multiAp;

1.  通过submitAll()函数进行提交任务

> 底层中使用了AsyncProcess.java类中的submitAll()方法进行提交任务  
> \[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aqQZt6vv-1573199896562)(F0A21E01080B4404984DBEE85DF46D79)\]

具体代码如下：

/**
       * Submit immediately the list of rows, whatever the server status. Kept for backward
       * compatibility: it allows to be used with the batch interface that return an array of objects.
       *
       * @param pool ExecutorService to use.
       * @param tableName name of the table for which the submission is made.
       * @param rows the list of rows.
       * @param callback the callback.
       * @param results Optional array to return the results thru; backward compat.
       */
      public <CResult> AsyncRequestFuture submitAll(ExecutorService pool, TableName tableName,
          List<? extends Row> rows, Batch.Callback<CResult> callback, Object[] results,
          PayloadCarryingServerCallable callable, int operationTimeout, int rpcTimeout) {
        List<Action<Row>> actions = new ArrayList<Action<Row>>(rows.size());//构造时直接初始化大小，值得学习
    
        // The position will be used by the processBatch to match the object array returned.
        int posInList = -1;
        //this.connection: 是这个成员变量 protected final ClusterConnection connection;
        NonceGenerator ng = this.connection.getNonceGenerator();
        for (Row r : rows) {
          posInList++;
          if (r instanceof Put) { //判断如果是否是Put类型
            Put put = (Put) r;
            if (put.isEmpty()) { //如果Put操作为空，直接抛出异常
              throw new IllegalArgumentException("No columns to insert for #" + (posInList+1)+ " item");
            }
          }
          Action<Row> action = new Action<Row>(r, posInList); //将put操作转变为Action
          setNonce(ng, r, action);
          actions.add(action); //添加到List中
        }
        //通过创建AsyncRequestFutureImpl进行具体的任务执行
        AsyncRequestFutureImpl<CResult> ars = createAsyncRequestFuture(
            tableName, actions, ng.getNonceGroup(), getPool(pool), callback, results, results != null,
            callable, operationTimeout, rpcTimeout);
        ars.groupAndSendMultiAction(actions, 1); // 对每个区域服务器的操作列表进行分组，并发送操作列表
        return ars;
      }

createAsyncRequestFuture()方法具体代码如下：

//创建AsyncRequestFutureImpl对象完成
      protected <CResult> AsyncRequestFutureImpl<CResult> createAsyncRequestFuture(
          TableName tableName, List<Action<Row>> actions, long nonceGroup, ExecutorService pool,
          Batch.Callback<CResult> callback, Object[] results, boolean needResults,
          PayloadCarryingServerCallable callable, int operationTimeout, int rpcTimeout) {
        return new AsyncRequestFutureImpl<CResult>(
            tableName, actions, nonceGroup, getPool(pool), needResults,
            results, callback, callable, operationTimeout, rpcTimeout);
      }

#### AsyncRequestFutureImpl类的组成结构 ####

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CfbuuHl1-1573199896562)(7AE7C3718CB34295AFAA8CD2DF14988E)\]

可以看出类中有两个runnable子类

AsyncRequestFutureImpl类的构造函数

public AsyncRequestFutureImpl(TableName tableName, List<Action<Row>> actions, long nonceGroup,
            ExecutorService pool, boolean needResults, Object[] results,
            Batch.Callback<CResult> callback, PayloadCarryingServerCallable callable,
            int operationTimeout, int rpcTimeout) {
         //...
        }

然后程序就会执行groupAndSendMultiAction()这个函数, 对每个区域服务器的操作列表进行分组，并发送操作列表

1.  首先会查找region的位置
2.  获取对应的HRegionLocation
3.  添加到队列中
4.  调用sendMultiAction()进行发送

private void groupAndSendMultiAction(List<Action<Row>> currentActions, int numAttempt) {
          Map<ServerName, MultiAction<Row>> actionsByServer =
              new HashMap<ServerName, MultiAction<Row>>();
    
          boolean isReplica = false;
          List<Action<Row>> unknownReplicaActions = null;
          for (Action<Row> action : currentActions) {
            RegionLocations locs = findAllLocationsOrFail(action, true);
            if (locs == null) continue;
            boolean isReplicaAction = !RegionReplicaUtil.isDefaultReplica(action.getReplicaId());
            if (isReplica && !isReplicaAction) {
              // This is the property of the current implementation, not a requirement.
              throw new AssertionError("Replica and non-replica actions in the same retry");
            }
            isReplica = isReplicaAction;
            HRegionLocation loc = locs.getRegionLocation(action.getReplicaId());
            if (loc == null || loc.getServerName() == null) {
              if (isReplica) {
                if (unknownReplicaActions == null) {
                  unknownReplicaActions = new ArrayList<Action<Row>>();
                }
                unknownReplicaActions.add(action);
              } else {
                // TODO: relies on primary location always being fetched
                manageLocationError(action, null);
              }
            } else {
              byte[] regionName = loc.getRegionInfo().getRegionName();
              addAction(loc.getServerName(), regionName, action, actionsByServer, nonceGroup);
            }
          }
          boolean doStartReplica = (numAttempt == 1 && !isReplica && hasAnyReplicaGets);
          boolean hasUnknown = unknownReplicaActions != null && !unknownReplicaActions.isEmpty();
    
          if (!actionsByServer.isEmpty()) {
            // If this is a first attempt to group and send, no replicas, we need replica thread.
            sendMultiAction(actionsByServer, numAttempt, (doStartReplica && !hasUnknown)
                ? currentActions : null, numAttempt > 1 && !hasUnknown);
          }
    
          if (hasUnknown) {
            actionsByServer = new HashMap<ServerName, MultiAction<Row>>();
            for (Action<Row> action : unknownReplicaActions) {
              HRegionLocation loc = getReplicaLocationOrFail(action);
              if (loc == null) continue;
              byte[] regionName = loc.getRegionInfo().getRegionName();
              addAction(loc.getServerName(), regionName, action, actionsByServer, nonceGroup);
            }
            if (!actionsByServer.isEmpty()) {
              sendMultiAction(
                  actionsByServer, numAttempt, doStartReplica ? currentActions : null, true);
            }
          }
        }

##### 参考链接 #####

http://cxy7.com/articles/2018/07/01/1530451774179.html