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;

}