hbase源码系列（十一）Put、Delete在服务端是如何处理？

responseProto = getStub().multi(controller, requestProto);

regionMutationBuilder.setAtomic(true);
getStub().multi(null, request);

doBatchOp(builder, region, mutations, cellScanner);

long addedSize = doMiniBatchMutation(batchOp, isReplay);
//MemStore的大小到了阀值，就要flush到文件了
if (isFlushSize(newSize)) {
   requestFlush();
}

//日志，isInReplay是否支持重做，这里是false
WALEdit walEdit = new WALEdit(isInReplay);
//控制多版本的MemStore flush的结果，每次flush的w都是一样的，就好像同一批号的食品
MultiVersionConsistencyControl.WriteEntry w = null;
long txid = 0;
//日志同步是否成功
boolean walSyncSuccessful = false;
boolean locked = false;

//加锁，现在要上锁了，这段时间内不允许读
lock(this.updatesLock.readLock(), numReadyToWrite);
locked = true;

//等待时间
final long waitTime = Math.min(maxBusyWaitDuration,
        busyWaitDuration * Math.min(numReadyToWrite, maxBusyWaitMultiplier));
      if (!lock.tryLock(waitTime, TimeUnit.MILLISECONDS)) {
        throw new RegionTooBusyException(
          "failed to get a lock in " + waitTime + "ms");
}

//为这次添加进MemStore的数据添加一个批次号
w = mvcc.beginMemstoreInsert();

//这是批次号的计算方式，nextWriteNumber就等于memstore的写的次数+1
public WriteEntry beginMemstoreInsert() {
    synchronized (writeQueue) {
      long nextWriteNumber = ++memstoreWrite;
      WriteEntry e = new WriteEntry(nextWriteNumber);
      writeQueue.add(e);
      return e;
    }
}

//把kv们写入memstore
long addedSize = 0;
for (int i = firstIndex; i < lastIndexExclusive; i++) {
    if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN) {
        continue;
    }
    addedSize += applyFamilyMapToMemstore(familyMaps[i], w);
} 

private long applyFamilyMapToMemstore(Map<byte[], List<Cell>> familyMap,
    MultiVersionConsistencyControl.WriteEntry localizedWriteEntry) {
    long size = 0;try {for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
        byte[] family = e.getKey();
        List<Cell> cells = e.getValue();
        //把kv添加到memstore当中
        Store store = getStore(family);
        for (Cell cell: cells) {
          KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
          kv.setMvccVersion(localizedWriteEntry.getWriteNumber());
          size += store.add(kv);
        }
      }
    } 
return size;
   }

      Durability durability = Durability.USE_DEFAULT;
      for (int i = firstIndex; i < lastIndexExclusive; i++) {
        // 跳过状态不对的
        if (batchOp.retCodeDetails[i].getOperationStatusCode()
            != OperationStatusCode.NOT_RUN) {
          continue;
        }
        //标志状态为成功
        batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;

        Mutation m = batchOp.operations[i];
        //获取自定义的日志同步方式
        Durability tmpDur = getEffectiveDurability(m.getDurability());
        if (tmpDur.ordinal() > durability.ordinal()) {
          durability = tmpDur;
        }
        if (tmpDur == Durability.SKIP_WAL) {
          //记录日志的kv的大小，但是不写入到日志当中
          recordMutationWithoutWal(m.getFamilyCellMap());
          continue;
        }
//把列族里面的kv全部添加到walEdit当中
        addFamilyMapToWALEdit(familyMaps[i], walEdit);
      }

Mutation mutation = batchOp.operations[firstIndex];
if (walEdit.size() > 0) {
   //异步添加日志
   txid = this.log.appendNoSync(this.getRegionInfo(), this.htableDescriptor.getTableName(),
           walEdit, mutation.getClusterIds(), now, this.htableDescriptor);
}

//释放相关的锁
if (locked) {
    this.updatesLock.readLock().unlock();
    locked = false;
}
releaseRowLocks(acquiredRowLocks);

if (walEdit.size() > 0) {
    syncOrDefer(txid, durability);
}
walSyncSuccessful = true;

if (w != null) {
   mvcc.completeMemstoreInsert(w);
   w = null;
}

finally {
  if (!walSyncSuccessful) {
     //如果日志没有成功，
     rollbackMemstore(batchOp, familyMaps, firstIndex, lastIndexExclusive);
  }
  ......  
}

KeyValue found = this.snapshot.get(kv);
if (found != null && found.getMvccVersion() == kv.getMvccVersion()) {
 　this.snapshot.remove(kv);
}
// 比较一下mvcc，相同就删除.
found = this.kvset.get(kv);
if (found != null && found.getMvccVersion() == kv.getMvccVersion()) {
   removeFromKVSet(kv);
   long s = heapSizeChange(kv, true);
   this.size.addAndGet(-s);
}

void prepareDeleteTimestamps(Map<byte[], List<Cell>> familyMap, byte[] byteNow)
      throws IOException {
    for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
      byte[] family = e.getKey();
      List<Cell> cells = e.getValue();
      //列和count的映射
      Map<byte[], Integer> kvCount = new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);

      for (Cell cell: cells) {
        KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
        // 如果是时间戳是最新的话就执行下面这些操作
        if (kv.isLatestTimestamp() && kv.isDeleteType()) {
          //new一个Get从Store里面去搜索
         } else {
          kv.updateLatestStamp(byteNow);
        }
      }
    }
  }

public Delete(byte [] row) {
    this(row, HConstants.LATEST_TIMESTAMP);
}

public Delete(byte [] row, long timestamp) {
    this(row, 0, row.length, timestamp);
}

byte[] qual = kv.getQualifier();
          if (qual == null) qual = HConstants.EMPTY_BYTE_ARRAY;
          //想到相同列的每次+1
          Integer count = kvCount.get(qual);
          if (count == null) {
            kvCount.put(qual, 1);
          } else {
            kvCount.put(qual, count + 1);
          }
          //更新之后把最新的count数量
          count = kvCount.get(qual);

          Get get = new Get(kv.getRow());
          get.setMaxVersions(count);
          get.addColumn(family, qual);
          //从store当中取出相应的result来
          List<Cell> result = get(get, false);

          if (result.size() < count) {
            // Nothing to delete 数量不够。。 更新最新的时间戳为现在的时间
            kv.updateLatestStamp(byteNow);
            continue;
          }
          //数量超过了也不行
          if (result.size() > count) {
            throw new RuntimeException("Unexpected size: " + result.size());
          }
          //取最后一个的时间戳
          KeyValue getkv = KeyValueUtil.ensureKeyValue(result.get(count - 1));
          //更新kv的时间戳为getkv的时间戳
          Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(),
              getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG);