hbase源码系列（十）HLog与日志恢复

class Entry implements Writable {
    private WALEdit edit;
    private HLogKey key;
}

public HLogKey(final byte [] encodedRegionName, final TableName tablename,
      long logSeqNum, final long now, List clusterIds){
    init(encodedRegionName, tablename, logSeqNum, now, clusterIds);
 }

protected void init(final byte [] encodedRegionName, final TableName tablename,
      long logSeqNum, final long now, List clusterIds) {
    this.logSeqNum = logSeqNum;
    this.writeTime = now;
    this.clusterIds = clusterIds;
    this.encodedRegiOnName= encodedRegionName;
    this.tablename = tablename;
 }

public class WALEdit implements Writable, HeapSize {
  ......private final ArrayList kvs = new ArrayList();

while(!this.isInterrupted() && !closeLogSyncer.get()) {
          try {
            if (unflushedEntries.get() <= syncedTillHere) {
              synchronized (closeLogSyncer) {
                closeLogSyncer.wait(this.optionalFlushInterval);
              }
            }// 同步已经添加的entry
            sync();          } catch (IOException e) {
            LOG.error("Error while syncing, requesting close of hlog ", e);
            requestLogRoll();
            Threads.sleep(this.optionalFlushInterval);
          }
}

byte [][] regiOnsToFlush= getWAL().rollWriter(rollLog.get());
        if (regionsToFlush != null) {
          for (byte [] r: regionsToFlush) scheduleFlush(r);
}

regiOns= findMemstoresWithEditsEqualOrOlderThan(this.outputfiles.firstKey(),
          this.oldestUnflushedSeqNums);

static byte[][] findMemstoresWithEditsEqualOrOlderThan(
      final long walSeqNum, final Map<byte[], Long> regionsToSeqNums) {
    List<byte[]> regiOns= null;
    for (Map.Entry<byte[], Long> e : regionsToSeqNums.entrySet()) {
      //逐个对比，找出小于已输出为文件的最小的seq id的region
      if (e.getValue().longValue() <= walSeqNum) {
        if (regiOns== null) regiOns= new ArrayList<byte[]>();
        regions.add(e.getKey());
      }
    }
    return regiOns== null ? null : regions
        .toArray(new byte[][] { HConstants.EMPTY_BYTE_ARRAY });
}

    // 如果recovered.edits有日志的话，就恢复日志
    maxSeqId = Math.max(maxSeqId, replayRecoveredEditsIfAny(
        this.fs.getRegionDir(), maxSeqIdInStores, reporter, status));

    HLog.Reader reader = null;
    try {
      //创建reader读取hlog
      reader = HLogFactory.createReader(fs, edits, conf);
      long currentEditSeqId = -1;
      long firstSeqIdInLog = -1;
      long skippedEdits = 0;
      long editsCount = 0;
      long intervalEdits = 0;
      HLog.Entry entry;
      Store store = null;
      boolean reported_Once= false;

      try {//逐个读取
        while ((entry = reader.next()) != null) {
          HLogKey key = entry.getKey();
          WALEdit val = entry.getEdit();
          //实例化firstSeqIdInLog
          if (firstSeqIdInLog == -1) {
            firstSeqIdInLog = key.getLogSeqNum();
          }
          boolean flush = false;
          for (KeyValue kv: val.getKeyValues()) {
            // 从WALEdits里面取出kvs
            if (kv.matchingFamily(WALEdit.METAFAMILY) ||
                !Bytes.equals(key.getEncodedRegionName(),
                  this.getRegionInfo().getEncodedNameAsBytes())) {//是meta表的kv就有compaction
              CompactionDescriptor compaction = WALEdit.getCompaction(kv);
              if (compaction != null) {
                //完成compaction未完成的事情，校验输入输出文件，完成文件替换等操作
                completeCompactionMarker(compaction);
              }

              skippedEdits++;
              continue;
            }
            // 获得kv对应的store
            if (store == null || !kv.matchingFamily(store.getFamily().getName())) {
              store = this.stores.get(kv.getFamily());
            }
            if (store == null) {
              // 应该不会发生，缺少它对应的列族
              skippedEdits++;
              continue;
            }
            // seq id小，呵呵，说明已经被处理过了这个日志
            if (key.getLogSeqNum() <= maxSeqIdInStores.get(store.getFamily().getName())) {
              skippedEdits++;
              continue;
            }
            currentEditSeqId = key.getLogSeqNum();
            // 这个就是我们要处理的日志，添加到MemStore里面就ok了            flush = restoreEdit(store, kv);
            editsCount++;
          }
          //MemStore太大了，需要flush掉
          if (flush) internalFlushcache(null, currentEditSeqId, status);

         }
      } catch (IOException ioe) {
        // 就是把名字改了，然后在后面加上".时间戳"，这个有毛意思？
        if (ioe.getCause() instanceof ParseException) {          Path p = HLogUtil.moveAsideBadEditsFile(fs, edits);
          msg = "File corruption encountered!  " +
              "Continuing, but renaming " + edits + " as " + p;
        } else {// 不知道是啥错误，抛错误吧，处理不了
          throw ioe;
        }
      }
      status.markComplete(msg);
      return currentEditSeqId;
    } finally {
      status.cleanup();
      if (reader != null) {
         reader.close();
      }
    }