hbase写数据过程

// 单条写API

public void put( final Put put) throws IOException {

    doPut(put);

    if (autoFlush) {

      flushCommits();

    }

  }

//批量写API

public void put( final List<Put> puts) throws IOException {

    for (Put put : puts) {

      doPut(put);

    }

    if (autoFlush) {

      flushCommits();

    }

  }

//具体的put实现

private void doPut(Put put) throws IOException{

    validatePut(put);

    writeBuffer.add(put);

    currentWriteBufferSize += put.heapSize();

    if (currentWriteBufferSize > writeBufferSize) {

      flushCommits();

    }

  }

public void close() throws IOException {

    if ( this.closed) {

      return;

    }

    flushCommits();

    ….

  }

public void flushCommits() throws IOException {

    try {

      Object[] results = new Object[writeBuffer.size()];

       try {

        this.connection.processBatch(writeBuffer, tableName, pool, results);

      } catch (InterruptedException e) {

        throw new IOException(e);

      } finally {

        …

    } finally {

      …

    }

  }

public void processBatch(List<? extends Row> list, final byte[] tableName, ExecutorService pool,

        Object[] results) throws IOException, InterruptedException {

      …

      processBatchCallback(list, tableName, pool, results, null);

    }

public <R> void processBatchCallback(List<? extends Row> list, byte[] tableName, ExecutorService pool,

        Object[] results, Batch.Callback<R> callback) throws IOException, InterruptedException {

      ….

      HRegionLocation [] lastServers = new HRegionLocation[results.length];

      for ( int tries = 0; tries < numRetries && retry; ++tries) {

…

        // step 1: break up into regionserver-sized chunks and build the data structs

        Map<HRegionLocation, MultiAction<R>> actionsByServer =

          new HashMap<HRegionLocation, MultiAction<R>>();

for ( int i = 0; i < workingList.size(); i++) {

          Row row = workingList.get(i);

          if (row != null) {

            HRegionLocation loc = locateRegion(tableName, row.getRow());

            byte[] regionName = loc.getRegionInfo().getRegionName();

            MultiAction<R> actions = actionsByServer.get(loc);

            if (actions == null) {

              actions = new MultiAction<R>();

              actionsByServer.put(loc, actions);  //每一个region对应一个MultiAction对象，每个MultiAction对象持有该region所有的put Action

            }

            Action<R> action = new Action<R>(row, i);

            lastServers[i] = loc;

            actions.add(regionName, action);

          }

        }

        // step 2: make the requests ，每个region 开启一个线程

        Map<HRegionLocation, Future<MultiResponse>> futures =

            new HashMap<HRegionLocation, Future<MultiResponse>>(actionsByServer.size());

        for (Entry<HRegionLocation, MultiAction<R>> e: actionsByServer.entrySet()) {

          futures.put(e.getKey(), pool.submit(createCallable(e.getKey(), e.getValue(), tableName)));

        }

        // step 3: collect the failures and successes and prepare for retry

        …

        // step 4: identify failures and prep for a retry (if applicable).

        …

}

…

}

// 从缓存中定位region，通过NavigableMap实现，如果没有缓存则需查询.META.表

HRegionLocation getCachedLocation( final byte [] tableName,

        final byte [] row) {

      SoftValueSortedMap< byte [], HRegionLocation> tableLocations =

        getTableLocations(tableName);

      …

      // 找到小于 rowKey 并且最接近rowKey 的 startKey 对应 的region ，通过NavigableMap 实现

      possibleRegion = tableLocations.lowerValueByKey(row);

      if (possibleRegion == null) {

        return null;

      }

      // 表的最末一个region的endKey是空字符串，如果不是最末一个region，则只有当rowKey小于endKey才返回region。

      byte[] endKey = possibleRegion.getRegionInfo().getEndKey();

      if (Bytes. equals(endKey, HConstants. EMPTY_END_ROW) ||

          KeyValue. getRowComparator(tableName).compareRows(

              endKey, 0, endKey.length, row, 0, row.length) > 0) {

        return possibleRegion;

      }

      return null;

}

//单条put

public void put( final byte[] regionName, final Put put) throws IOException {

    HRegion region = getRegion(regionName);

if (!region.getRegionInfo().isMetaTable()) {

// 检查HRegionServer的memstore总内存占用量是否已经超过了hbase.regionserver.global.memstore.upperLimit（默认值是0.4）或者hbase.regionserver.global.memstore.lowerLimit（默认值是0.35）的限制，如果超过了则会在flush队列中添加一个任务，其中如果是超过了upper的限制则会阻塞所有的写memstore的操作，直到内存降至lower限制以下。

       this.cacheFlusher.reclaimMemStoreMemory();

    }

boolean writeToWAL = put.getWriteToWAL();

//region会调用Store的add()方法把数据保存到相关Store的memstore中

//region在保存完数据后，会检查是否需要flush memstore，如果需要则发出flush请求，由HRegionServer的flush守护线程异步执行。

    region.put(put, getLockFromId(put.getLockId()), writeToWAL);

  }

// 批量put

public int put( final byte[] regionName, final List<Put> puts) throws IOException {

    region = getRegion(regionName);

    if (!region.getRegionInfo().isMetaTable()) {

       this.cacheFlusher.reclaimMemStoreMemory();

    }

   OperationStatus codes[] = region.batchMutate(putsWithLocks);

    for (i = 0; i < codes.length; i++) {

        if (codes[i].getOperationStatusCode() != OperationStatusCode. SUCCESS) {

          return i;

        }

     }

     return -1;

  }

// threadWakeFrequency 默认值是10*1000 ， multiplier 默认值是1000 ，单位:毫秒

this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency * multiplier, this);

//chore是CompactionChecker定时执行的方法，定时进行minor和major的compcat合并，如果hbase.hregion.majorcompaction配置为0则不执行major合并，minor升级为major除外。

protected void chore() {

      for (HRegion r : this.instance.onlineRegions.values()) {

        if (r == null)

          continue;

        for (Store s : r.getStores().values()) {

          try {

            if ( s.needsCompaction()) {

              // 如果整个store 下的storeFile 文件均需要合并，则会自动升级到major 合并

              this.instance.compactSplitThread.requestCompaction(r, s, getName()

                  + " requests compaction", null);

            } else if (s.isMajorCompaction()) {

              if (majorCompactPriority == DEFAULT_PRIORITY

                  || majorCompactPriority > r.getCompactPriority()) {

                this.instance.compactSplitThread.requestCompaction(r, s, getName()

                    + " requests major compaction; use default priority", null);

              } else {

                this.instance.compactSplitThread.requestCompaction(r, s, getName()

                    + " requests major compaction; use configured priority",

                  this.majorCompactPriority, null);

              }

            }

          } catch (IOException e) {

            LOG.warn("Failed major compaction check on " + r, e);

          }

        }

      }

}

// store 内除去正在执行compact 的storeFile 后剩余的storeFile 数如果大于配置的最小可合并数，则可以进行compact 合并，最小的可合并数通过hbase.hstore.compactionThreshold 配置，默认是3 ，最小值为2 。

public boolean needsCompaction() {

    return (storefiles.size() - filesCompacting.size()) > minFilesToCompact;

}

// 是否是major 合并

private boolean isMajorCompaction(final List<StoreFile> filesToCompact) throws IOException {

boolean result = false;

// 根据hbase.hregion.majorcompaction 配置的major 合并周期计算下次进行major 合并的时间，如果设置为0则不进行major合并

long mcTime = getNextMajorCompactTime();

if (filesToCompact == null || filesToCompact.isEmpty() || mcTime == 0) {

      return result;

    }

// TODO: Use better method for determining stamp of last major (HBASE-2990)

// store中最久没有被修改过的 storeFile文件的 时间，作为上次major合并的时间进行判断下次应该进行major合并的时间，这种做法并不合理，可能会导致延后执行major 合并，极端情况下会导致永远不进行major 合并。

    long lowTimestamp = getLowestTimestamp(filesToCompact);

    long now = System. currentTimeMillis();

    // 只有当达到了major 合并时间才可能进行major 合并

    if (lowTimestamp > 0l && lowTimestamp < (now - mcTime)) { 

      // Major compaction time has elapsed.

      if (filesToCompact.size() == 1) {

        StoreFile sf = filesToCompact.get(0);

        // store 中最久的时间与当前时间的时间差

        long oldest = (sf.getReader().timeRangeTracker == null) ?

                Long. MIN_VALUE :

                now - sf.getReader().timeRangeTracker.minimumTimestamp;

        if (sf.isMajorCompaction() && ( this.ttl == HConstants. FOREVER || oldest < this.ttl)) {

          //如果 列簇没有设置过期时间( 通过HColumnDescriptor.setTimeToLive() 设置) ，因此无需通过major 合并删除过期数据。

          }

        } else if ( this.ttl != HConstants. FOREVER && oldest > this.ttl) {

            result = true;

        }

      } else {

        result = true;

      }

    }

    return result;

  }

//CompactSplitThread类，只由HRegionServer类持有，在以下几个地方被调用：

//1、HRegionServer的compact守护线程

//2、MemStoreFlusher的flushRegion

//3、CompactingRequest的run方法

public synchronized CompactionRequest requestCompaction(final HRegion r, final Store s,

      final String why, int priority, CompactionRequest request) throws IOException {

…

    CompactionRequest cr = s.requestCompaction(priority, request);

…

      cr.setServer(server);

…

// 是否是large合并，只与参与合并的文件的总大小有关，超过一定值后就会通过large合并的线程池，

//注意与major合并的区别，large线程池执行的任务可能是一个minor合并也可能是major合并。

//默认的large和small线程数是1，可以通过hbase.regionserver.thread.compaction.large和hbase.regionserver.thread.compaction.small配置

ThreadPoolExecutor pool = s.throttleCompaction(cr.getSize())? largeCompactions : smallCompactions;

      pool.execute(cr);

…

    return cr;

  }

// Store 类

public CompactionRequest requestCompaction( int priority, CompactionRequest request)

      throws IOException {

   …

    this.lock.readLock().lock();

    try {

      synchronized (filesCompacting) {

        // candidates = all storefiles not already in compaction queue

        List<StoreFile> candidates = Lists. newArrayList(storefiles);

        if (!filesCompacting.isEmpty()) {

          // exclude all files older than the newest file we're currently

          // compacting. this allows us to preserve contiguity (HBASE-2856)

          StoreFile last = filesCompacting.get(filesCompacting.size() - 1);

          int idx = candidates.indexOf(last);

          Preconditions. checkArgument(idx != -1);

          candidates.subList(0, idx + 1).clear();

        }

        boolean override = false;

        if (region.getCoprocessorHost() != null) {

          override = region.getCoprocessorHost().preCompactSelection( this, candidates, request);

        }

        CompactSelection filesToCompact;

        if (override) {

          // coprocessor is overriding normal file selection

          filesToCompact = new CompactSelection(conf, candidates);

        } else {

          filesToCompact = compactSelection(candidates, priority);

        }

        if (region.getCoprocessorHost() != null) {

          region.getCoprocessorHost().postCompactSelection( this,

            ImmutableList. copyOf(filesToCompact.getFilesToCompact()), request);

        }

        // no files to compact

        if (filesToCompact.getFilesToCompact().isEmpty()) {

          return null;

        }

        // basic sanity check: do not try to compact the same StoreFile twice.

        if (!Collections. disjoint(filesCompacting, filesToCompact.getFilesToCompact())) {

          // TODO: change this from an IAE to LOG.error after sufficient testing

          Preconditions. checkArgument( false, "%s overlaps with %s",

              filesToCompact, filesCompacting);

        }

        filesCompacting.addAll(filesToCompact.getFilesToCompact());

        Collections. sort(filesCompacting, StoreFile.Comparators. FLUSH_TIME);

        // major compaction iff all StoreFiles are included

        boolean isMajor = (filesToCompact.getFilesToCompact().size() == this.storefiles.size());

        if (isMajor) {

          // since we're enqueuing a major, update the compaction wait interval

          this.forceMajor = false;

        }

        // everything went better than expected. create a compaction request

        int pri = getCompactPriority(priority);

        //not a special compaction request, so we need to make one

        if(request == null){

          request = new CompactionRequest(region, this, filesToCompact, isMajor, pri);

        } else {

          // update the request with what the system thinks the request should be

          // its up to the request if it wants to listen

          request.setSelection(filesToCompact);

          request.setIsMajor(isMajor);

          request.setPriority(pri);

        }

      }

    } finally {

      this.lock.readLock().unlock();

    }

    if (request != null) {

      CompactionRequest. preRequest(request);

    }

    return request;

  }

// 如果合并的总文件大小超过2 * this.minFilesToCompact * this.region.memstoreFlushSize则会通过大合并的线程池进行合并，总共有两个合并的线程池

ThreadPoolExecutor pool = s.throttleCompaction(cr.getSize())? largeCompactions : smallCompactions;

// minFilesToCompact默认值为3， memstoreFlushSize默认值128M

boolean throttleCompaction( long compactionSize) {

    long throttlePoint = conf.getLong(

        "hbase.regionserver.thread.compaction.throttle",

        2 * this.minFilesToCompact * this.region.memstoreFlushSize);

    return compactionSize > throttlePoint;

  }