hbase源码系列（七）Snapshot的过程

public void takeSnapshot(SnapshotDescription snapshot) throws IOException {
    // 清空之前完成的备份和恢复的任务
    cleanupSentinels();

    // 设置snapshot的版本
    snapshot = snapshot.toBuilder().setVersion(SnapshotDescriptionUtils.SNAPSHOT_LAYOUT_VERSION)
        .build();

    // if the table is enabled, then have the RS run actually the snapshot work
    TableName snapshotTable = TableName.valueOf(snapshot.getTable());
    AssignmentManager assignmentMgr = master.getAssignmentManager();
    //根据表的状态选择snapshot的类型
    if (assignmentMgr.getZKTable().isEnabledTable(snapshotTable)) {
      snapshotEnabledTable(snapshot);
    }
    // 被禁用的表走这个方法
    else if (assignmentMgr.getZKTable().isDisabledTable(snapshotTable)) {
      snapshotDisabledTable(snapshot);
    } else {
      throw new SnapshotCreationException("Table is not entirely open or closed", tpoe, snapshot);
    }
  }

private synchronized void snapshotEnabledTable(SnapshotDescription snapshot)
      throws HBaseSnapshotException {
    // snapshot准备工作
    prepareToTakeSnapshot(snapshot);
    
    // new一个handler
    EnabledTableSnapshotHandler handler =
        new EnabledTableSnapshotHandler(snapshot, master, this);
    //通过handler线程来备份
    snapshotTable(snapshot, handler);
 }

handler.prepare();
this.executorService.submit(handler);
this.snapshotHandlers.put(TableName.valueOf(snapshot.getTable()), handler);

      // 把snapshot的信息写入到工作目录
      SnapshotDescriptionUtils.writeSnapshotInfo(snapshot, workingDir, this.fs);
      // 开一个线程去复制表信息文件
      new TableInfoCopyTask(monitor, snapshot, fs, rootDir).call();
      monitor.rethrowException();
      //查找该表相关的region和位置
      List<Pair<HRegionInfo, ServerName>> regionsAndLocations =
          MetaReader.getTableRegionsAndLocations(this.server.getCatalogTracker(),
              snapshotTable, false);

      // 开始snapshot
      snapshotRegions(regionsAndLocations);

      // 获取serverNames列表，后面的校验snapshot用到
      Set<String> serverNames = new HashSet<String>();
      for (Pair<HRegionInfo, ServerName> p : regionsAndLocations) {
        if (p != null && p.getFirst() != null && p.getSecond() != null) {
          HRegionInfo hri = p.getFirst();
          if (hri.isOffline() && (hri.isSplit() || hri.isSplitParent())) continue;
          serverNames.add(p.getSecond().toString());
        }
      }

      // 检查snapshot是否合格
      status.setStatus("Verifying snapshot: " + snapshot.getName());
      verifier.verifySnapshot(this.workingDir, serverNames);
// 备份完毕之后，把临时目录转移到正式的目录
      completeSnapshot(this.snapshotDir, this.workingDir, this.fs);

　　// 用分布式事务来备份在线的，太强悍了
    Procedure proc = coordinator.startProcedure(this.monitor, this.snapshot.getName(),
      this.snapshot.toByteArray(), Lists.newArrayList(regionServers));
    try {
      // 等待完成
      proc.waitForCompleted();
　　　　// 备份split过的region
      Path snapshotDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(snapshot, rootDir);
      for (Pair<HRegionInfo, ServerName> region : regions) {
        HRegionInfo regionInfo = region.getFirst();
        if (regionInfo.isOffline() && (regionInfo.isSplit() || regionInfo.isSplitParent())) {
          if (!fs.exists(new Path(snapshotDir, regionInfo.getEncodedName()))) {
            LOG.info("Take disabled snapshot of offline region=" + regionInfo);
            snapshotDisabledRegion(regionInfo);
          }
        }
      }

Procedure proc = createProcedure(fed, procName, procArgs,expectedMembers);
if (!this.submitProcedure(proc)) {
      LOG.error("Failed to submit procedure ‘" + procName + "‘");
      return null;
}

  final public Void call() {
    try {
      //在acquired节点下面建立实例节点
      sendGlobalBarrierStart();

      // 等待所有的rs回复
      waitForLatch(acquiredBarrierLatch, monitor, wakeFrequency, "acquired");

      //在reached节点下面建立实例节点
      sendGlobalBarrierReached();

      //等待所有的rs回复
      waitForLatch(releasedBarrierLatch, monitor, wakeFrequency, "released");

    } finally {
      sendGlobalBarrierComplete();
      completedLatch.countDown();
    }
}

　　coord.getRpcs().sendGlobalBarrierAcquire(this, args, Lists.newArrayList(this.acquiringMembers));

final public void sendGlobalBarrierAcquire(Procedure proc, byte[] info, List<String> nodeNames)
      throws IOException, IllegalArgumentException {
    String procName = proc.getName();
    // 获取abort节点的名称
    String abortNode = zkProc.getAbortZNode(procName);
    try {
      // 如果存在abort节点，就广播错误，中断该过程
      if (ZKUtil.watchAndCheckExists(zkProc.getWatcher(), abortNode)) {
        abort(abortNode);
      }
    } catch (KeeperException e) {throw new IOException("Failed while watching abort node:" + abortNode, e);
    }

    // 获得acquire节点名称
    String acquire = zkProc.getAcquiredBarrierNode(procName);try {
      // 创建acquire节点
      byte[] data = ProtobufUtil.prependPBMagic(info); 
      ZKUtil.createWithParents(zkProc.getWatcher(), acquire, data);// 监控acquire下面的节点，发现指定的节点，就报告给coordinator
      for (String node : nodeNames) {
        String znode = ZKUtil.joinZNode(acquire, node);if (ZKUtil.watchAndCheckExists(zkProc.getWatcher(), znode)) {
          coordinator.memberAcquiredBarrier(procName, node);
        }
      }
    } catch (KeeperException e) {
      throw new IOException("Failed while creating acquire node:" + acquire, e);
    }
  }

final public Void call() {
    
    try {
      // 目前是么也没干
      acquireBarrier();
      // 在acquired的实例节点下面建立rs的节点
      rpcs.sendMemberAcquired(this);
      // 等待reached的实例节点的建立
      waitForReachedGlobalBarrier();
     // 干活
      insideBarrier();
      // 活干完了
      rpcs.sendMemberCompleted(this);
     } catch (Exception e) {
     } finally {
          releasedLocalBarrier.countDown();
     }
}

for (HRegion region : regions) {
      taskManager.submitTask(new RegionSnapshotTask(region));
 }

public Void call() throws Exception {
      // 上锁，暂时不让读了
      region.startRegionOperation();
      try {
        region.flushcache();
        region.addRegionToSnapshot(snapshot, monitor);
      } finally {
        LOG.debug("Closing region operation on " + region);
        region.closeRegionOperation();
      }
      return null;
    }
}

public void addRegionToSnapshot(SnapshotDescription desc,
      ForeignExceptionSnare exnSnare) throws IOException {// 获取工作目录
    Path rootDir = FSUtils.getRootDir(this.rsServices.getConfiguration());
    Path snapshotDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(desc, rootDir);

    // 1. 在工作目录创建region目录和写入region的信息
    HRegionFileSystem snapshotRegionFs = HRegionFileSystem.createRegionOnFileSystem(conf,
        this.fs.getFileSystem(), snapshotDir, getRegionInfo());

    // 2. 为hfile创建引用
    for (Store store : stores.values()) {
      // 2.1. 分列族为store创建引用目录，每个store属于不同的列族
      Path dstStoreDir = snapshotRegionFs.getStoreDir(store.getFamily().getNameAsString());
      List<StoreFile> storeFiles = new ArrayList<StoreFile>(store.getStorefiles());// 2.2. 遍历hfile，然后创建引用
      int sz = storeFiles.size();
      for (int i = 0; i < sz; i++) {
        StoreFile storeFile = storeFiles.get(i);
        Path file = storeFile.getPath();

        Path referenceFile = new Path(dstStoreDir, file.getName());
        boolean success = true;
        if (storeFile.isReference()) {
          // 把旧的引用文件的内容写入到新的引用文件当中
          storeFile.getFileInfo().getReference().write(fs.getFileSystem(), referenceFile);
        } else {
          // 创建一个空的引用文件
          success = fs.getFileSystem().createNewFile(referenceFile);
        }
        if (!success) {
          throw new IOException("Failed to create reference file:" + referenceFile);
        }
      }
    }
  }

protected void snapshotDisabledRegion(final HRegionInfo regionInfo)
      throws IOException {
    // 创建新的region目录和region信息
    HRegionFileSystem regionFs = HRegionFileSystem.createRegionOnFileSystem(conf, fs,
      workingDir, regionInfo);

     // 把region下的recovered.edits目录的文件复制snapshot的对应region下面
    Path regionDir = HRegion.getRegionDir(rootDir, regionInfo);
    Path snapshotRegionDir = regionFs.getRegionDir();
    new CopyRecoveredEditsTask(snapshot, monitor, fs, regionDir, snapshotRegionDir).call();
    
    // 给每个列族的下面的文件创建引用，所谓引用就是一个同名的空文件
    new ReferenceRegionHFilesTask(snapshot, monitor, regionDir, fs, snapshotRegionDir).call();
}