Hadoop3.2.1 【 HDFS 】源码分析 : BPServiceActor 解析

快来打我* 2022-12-22 00:51 156阅读 0赞

# 一.前言 #

一个HDFS集群可以配置多个命名空间(Namespace） ， 每个Datanode都会存储多个块池的数据块。 所以在Datanode实现中， 定义了BlockPoolManager类来管理Datanode上的所有块池， Datanode的其他模块对块池的操作都必须通过BlockPoolManager执行， 每个Datanode都有一个BlockPoolManager的实例。

由于在HDFS Federation部署中， **一个Datanode会保存多个块池的数据块， 所以BlockPoolManager会拥有多个BPOfferService对  
象， 每个BPOfferService对象都封装了对单个块池操作的API**。 同时， 由于在HDFS HA部署中， 每个命名空间又会**同时拥有两个Namenode**， 一个作为活动的（Active) Namenode，另一个作为热备的（Standby) Namenode， 所以**每个BPOfferService都会包含两个BPServiceActor对象**， 每个BPServiceActor对象都封装了与该命名空间中单个Namenode的操作， 包括定时向这个Namenode发送心跳（heartbeat） 、 增量块汇报（blockReceivedAndDeleted） 、 全量块汇报（blockreport） 、 缓存块汇报（cacheReport） ， 以及执行Namenode通过心跳／ 块汇报响应传回的名字节点指令等操作。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

# 二.BPServiceActor #

在BlockPoolManager的实现中， 使用BPServiceActor类负责与一个Namenode通信。 每个BPServiceActor的实例都是一个独立运行的线程， 这个线程主要实现了以下功能。  
■ 与Namenode进行第一次握手， 获取命名空间的信息。  
■ 向Namenode注册当前Datanode。  
■ 定期向Namenode发送心跳、 增量块汇报、 全量块汇报以及缓存块汇报。  
■ 执行Namenode传回的名字节点指令

## 2.1.BPServiceActor ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]

■ nnAddr： 当前BPServiceActor对应的Namenode的地址。  
■ state： 当前BPServiceActor对应的Namenode的状态。 \[ initializing ,active,standby,observer,stopping \]  
■ bpos： 管理当前BPServiceActor对象的BPOfferSerice对象的引用。  
■Scheduler \[ lastBlockReportTime、 lastDeletedReport、 lastHeartbeatTime\]： 记录上一次数据块汇报、 增量块汇报、 缓存块汇报以及心跳的时间。  
■ bpThread： 当前的工作线程， 是BPServiceActor主逻辑的执行线程。  
■ bpNamenode： 向Namenode发送RPC请求的代理， Datanode向Namenode发起DatanodeProtocolRPC调用都是通过这个引用进行的。  
■ dn和dnConf: Datanode对象的引用， 以及Datanode的配置。  
■ bpRegistration： 用于记录Datanode的注册信息。  
■ shouldServiceRun： 标志位， 用来指明当前BPServiceActor的状态， true为运行状态， false为停止状态， 这个标志位是用于控制bpThread工作线程的。  
■ runningState： 当前BPServiceActor的状态， BPServiceActor的状态有CONNECTING, INIT\_FAILED, RUNNING, EXITED, FAILED ， 初始状态是CONNECTNG。

## 2.2.构造方法 ##

BPServiceActor是由BPOfferService的构造方法构建的.

BPServiceActor(InetSocketAddress nnAddr, InetSocketAddress lifelineNnAddr,
          BPOfferService bpos) {
    
    
        this.bpos = bpos;
    
    
        this.dn = bpos.getDataNode();
    
        // localhost/127.0.0.1:8020
        this.nnAddr = nnAddr;
    
        // null
        this.lifelineSender = lifelineNnAddr != null ?
            new LifelineSender(lifelineNnAddr) : null;
    
        // null
        this.initialRegistrationComplete = lifelineNnAddr != null ?
            new CountDownLatch(1) : null;
    
        // DNConf@3736
        this.dnConf = dn.getDnConf();
    
        // IncrementalBlockReportManager
        this.ibrManager = new IncrementalBlockReportManager(
            dnConf.ibrInterval,
            dn.getMetrics());
        // -2554285107120895793
        prevBlockReportId = ThreadLocalRandom.current().nextLong();
    
        fullBlockReportLeaseId = 0;
    
    
        // heartBeatInterval : 3s
        // getLifelineIntervalMs : 9s
        // blockReportInterval : 21600 s  =   6h
        // outliersReportIntervalMs : 30min
        scheduler = new Scheduler(dnConf.heartBeatInterval,
            dnConf.getLifelineIntervalMs(), dnConf.blockReportInterval,
            dnConf.outliersReportIntervalMs);
    
        // get the value of maxDataLength.   67108864  == 64mb
        this.maxDataLength = dnConf.getMaxDataLength();
    
      }

## 2.3.start()、 stop()、 join() 方法 ##

start()、 stop()、 join()方法用于启动、 停止与等待BPServiceActor的bpThread工作线程线程。

//This must be called only by BPOfferService
      void start() {
        if ((bpThread != null) && (bpThread.isAlive())) {
          //Thread is started already
          //bpThread线程已经启动， 则直接返回
          return;
        }
        //构造bpThread并以守护线程的方式启动
        bpThread = new Thread(this);
        bpThread.setDaemon(true); // needed for JUnit testing
        bpThread.start();
    
        if (lifelineSender != null) {
          lifelineSender.start();
        }
      }
    
    
      //This must be called only by blockPoolManager.
      void stop() {
        //将shouldServiceRun赋值为false， 并中断bpThread线程的执行
        shouldServiceRun = false;
        if (lifelineSender != null) {
          lifelineSender.stop();
        }
        if (bpThread != null) {
          bpThread.interrupt();
        }
      }
      
      //This must be called only by blockPoolManager
      void join() {
        try {
          if (lifelineSender != null) {
            //直接调用bpThread的join()方法
            lifelineSender.join();
          }
          if (bpThread != null) {
            bpThread.join();
          }
        } catch (InterruptedException ie) { }
      }

## 2.4.run() 方法 ##

BPServiceActor的核心处理逻辑都是在bpThread工作线程中实现的， 也就是在BPServiceActor.run()方法中.

1.  首先调用BPServiceActor.connectToNNAndHandshake()方法与Namenode握手并初始化Datanode上该命名空间对应块池（ BlockPool） 的存储，
2.  然后在该Namenode上注册当前Datanode.
3.  执行完初始化操作之后,run()方法会循环调BPServiceActor.offerService()方法定期向Namenode发送心跳、 块汇报、 缓存汇报以及增量汇报。
4.  offerService()方法还会处理Namenode通过响应带回的名字节点指令。

offerService()方法会一直循环执行， 无论是否抛出异常， 直到shouldRunService字段被赋值为false时才会停止执行。

shouldRunService字段在两种情况下会被赋值为false：  
当Datanode关闭时 或者 客户端触发刷新Namenode操作时(Namenode列表的改变， 需要在当前命名空间中添加或者删除Namenode） 。

BPServiceActor.run()方法的代码如下：

/**
       * No matter what kind of exception we get, keep retrying to offerService().
       * That's the loop that connects to the NameNode and provides basic DataNode
       * functionality.
       *
       * Only stop when "shouldRun" or "shouldServiceRun" is turned off, which can
       * happen either at shutdown or due to refreshNamenodes.
       */
      @Override
      public void run() {
        LOG.info(this + " starting to offer service");
    
        try {
          while (true) {
            // init stuff
            try {
              // setup storage
              //与Namenode握手， 初始化Datanode存储并注册
              connectToNNAndHandshake();
              break;
            } catch (IOException ioe) {
              // Initial handshake, storage recovery or registration failed
              runningState = RunningState.INIT_FAILED;
              if (shouldRetryInit()) {
                //进行重试搡作， 直到BPOfferService停止初始化操作
                // Retry until all namenode's of BPOS failed initialization
                LOG.error("Initialization failed for " + this + " "
                    + ioe.getLocalizedMessage());
                sleepAndLogInterrupts(5000, "initializing");
              } else {
                  //初始化失败， 将运行状态改为FAILED
                runningState = RunningState.FAILED;
                LOG.error("Initialization failed for " + this + ". Exiting. ", ioe);
                return;
              }
            }
          }
    
          //初始化成功， 将运行状态设置为RUNNING
          runningState = RunningState.RUNNING;
          if (initialRegistrationComplete != null) {
            initialRegistrationComplete.countDown();
          }
    
          //线程的主循环，
          // 循环调用offerService()方法向Namenode发送心跳、 块汇报、 增量块汇报以及缓存块汇报，
          // 并执行Namenode返回的名字节点指令
          while (shouldRun()) {
            try {
              offerService();
            } catch (Exception ex) {
              //收到异常也不理会，休眠5秒,  继续循环直到BPServiceActor停止或者Datanode停止
              LOG.error("Exception in BPOfferService for " + this, ex);
              sleepAndLogInterrupts(5000, "offering service");
            }
          }
          //BPServiceActor停止以后， 将线程状态改为EXITED
          runningState = RunningState.EXITED;
        } catch (Throwable ex) {
          LOG.warn("Unexpected exception in block pool " + this, ex);
          runningState = RunningState.FAILED;
        } finally {
          //调用cleanup ()方法进行清理操作
          LOG.warn("Ending block pool service for: " + this);
          cleanUp();
        }
      }

## 2.5.BPServiceActor的状态机 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]

BPServiceActor的初始状态为**CONNECTING**， 当BPServiceActor成功与Namenode连接并握手完成后， 状态会转变为**RUNNING**。  
如果连接和握手过程当中出现异常， 并且有重试机制时， BPServiceActor转变为**INIT\_FAILED**状态。 当调用stop()方法将shouldServiceRun标志位设置为false后，BPServiceActor更改为**EXITED**状态。

## 2.6.connectToNNAndHandshake() ##

connectToNNAndHandshake()方法用于与Namenode握手， 初始化Datanode存储并注册当前Datnaode。

■ 获取Namenode的RPC代理。  
■ 第一阶段的握手， 通过调用RPC方法DatanodeProtocol.versionRequest()， 获取当前块池对应的命名空间的信息。  
■ 确认获取的命名空间信息与命名空间中另一个Namenode获取的信息匹配。 同时，如果这是第一个Namenode的连接， 则在Datanode上初始化命名空间对应块池的存储。  
■ 第二阶段的握手， 调用BPServiceActor.register()方法向Namenode注册当前Datanode。

private void connectToNNAndHandshake() throws IOException {
        // 获取与namenode通讯的代理对象
        // get NN proxy
        bpNamenode = dn.connectToNN(nnAddr);
    
        // First phase of the handshake with NN - get the namespace info.
        // 1. 获取NamespaceInfo
        NamespaceInfo nsInfo = retrieveNamespaceInfo();
    
        // Verify that this matches the other NN in this HA pair.
        // This also initializes our block pool in the DN if we are
        // the first NN connection for this BP.
        // 验证是否和namenode匹配, 如果是第一个Namenode 连接这个BP会初始化block pool
        bpos.verifyAndSetNamespaceInfo(this, nsInfo);
    
        /* set thread name again to include NamespaceInfo when it's available. */
        // 设置线程名
        this.bpThread.setName(formatThreadName("heartbeating", nnAddr));
    
        // Second phase of the handshake with the NN.
        //  向namenode注册该datanode的信息
        register(nsInfo);
      }

## 2.7.BPOfferService.verifyAndSetNaшespaceInfo() ##

这个方法是在BPOfferService上调用的， BPOfferService通过bpNSInfo字段保存当前块池对应的命名空间信息。 如果这个字段为空， 则证明当前BPServiceActor对应的Namenode是第一个响应握手请求的Namenode。 verifyAndSetNamespaceInfo()方法会首先对bpNSInfo赋值， 然后调用Datanode.initBLockPool()初始化命名空间对应块池的本地存储。如果这个字段不为空， 则证明命名空间中定义的另一个Namenode已经提前注册并且初始化了本地存储， 这时只需要将当前注册信息与已有的注册信息比较， 确认blockPoolId、namespaceId、 clusterId等字段相等即可。

BPOfferService.verifyAndSetNamespaceInfo()方法的代码如下：

/**
       * 
       * 由BPServiceActors在与NN握手时调用。
       * 如果这是第一个NN连接，则设置此BPOfferService的命名空间信息。
       * 如果是到新NN的连接，它将验证这个名称空间是否匹配（例如，为了防止在指定来自不同集群的备用节点时发生配置错误）
       * 
       * Called by the BPServiceActors when they handshake to a NN.
       * If this is the first NN connection, this sets the namespace info
       * for this BPOfferService. If it's a connection to a new NN, it
       * verifies that this namespace matches (eg to prevent a misconfiguration
       * where a StandbyNode from a different cluster is specified)
       */
      void verifyAndSetNamespaceInfo(BPServiceActor actor, NamespaceInfo nsInfo)
        throws IOException {
        writeLock();
    
        if(nsInfo.getState() == HAServiceState.ACTIVE
            && bpServiceToActive == null) {
          LOG.info("Acknowledging ACTIVE Namenode during handshake" + actor);
          bpServiceToActive = actor;
        }
    
        try {
    
          if (setNamespaceInfo(nsInfo) == null) {
            boolean success = false;
            //第一个Namenode的响应， 这时我们已经知道命名空间id，
            // 就可以通过调用Datanode.initBLockPool()方法初始化Datanode的本地存储了
    
            // Now that we know the namespace ID, etc, we can pass this to the DN.
            // The DN can now initialize its local storage if we are the
            // first BP to handshake, etc.
            try {
              dn.initBlockPool(this);
              success = true;
            } finally {
              if (!success) {
                // The datanode failed to initialize the BP. We need to reset
                // the namespace info so that other BPService actors still have
                // a chance to set it, and re-initialize the datanode.
    
                //如果初始化失败， 则将bpNSInfo置为空， 等待下一个Namenode的响应
                setNamespaceInfo(null);
              }
            }
          }
        } finally {
          writeUnlock();
        }
      }

这里的Datanode.initBlockPool()操作初始化了命名空间对应块池的本地存储， 在检查clusterId的一致性之后， 会在BlockPoolManager中注册当前块池对应的BPOfferService对象。 initBlockPool()方法还会初始化块池对应的DataStorage对象， 初始化FsDatasetImpl对象， 以及BlockScanner和DirectoryScanner对象.

## 2.8.register() ##

第二阶段的握手， 调用DatanodeProtocol.registerDatanode()方法在Namenode上注册当前的Datanode。  
注册成功后调用BPOfferService.registrationSucceeded()方法确认namespaceId、 clusterId与命名空间中定义的另一个Namenode返回的信息一致， 同时确认DatanodeUUid与Datanode本地存储一致。  
BPOfferService.registrationSucceeded()方法的实现请参考BPOfferService的方法小节。

/**
       *
       * 构造Datanode注册请求，
       * 这里StorageInfo以及DatanodeUUid已经通过上一阶段的握手获得并且持久化在磁盘上，
       * createRegistration()根据这些信息构造Datanode注册请求
       *
       *
       * Register one bp with the corresponding NameNode
       * <p>
       * The bpDatanode needs to register with the namenode on startup in order
       * 1) to report which storage it is serving now and 
       * 2) to receive a registrationID
       *  
       * issued by the namenode to recognize registered datanodes.
       * 
       * @param nsInfo current NamespaceInfo
       * @see FSNamesystem#registerDatanode(DatanodeRegistration)
       * @throws IOException
       */
      void register(NamespaceInfo nsInfo) throws IOException {
        // The handshake() phase loaded the block pool storage
        // off disk - so update the bpRegistration object from that info
        DatanodeRegistration newBpRegistration = bpos.createRegistration();
    
        LOG.info(this + " beginning handshake with NN");
    
        while (shouldRun()) {
          try {
            // 调用 DatanodeProtocol.registerDatanode()注册Datanode
            // Use returned registration from namenode with updated fields
            newBpRegistration = bpNamenode.registerDatanode(newBpRegistration);
            newBpRegistration.setNamespaceInfo(nsInfo);
            bpRegistration = newBpRegistration;
            break;
          } catch(EOFException e) {
              //Namenode忙碌， 则等待1秒后重试
              // namenode might have just restarted
            LOG.info("Problem connecting to server: " + nnAddr + " :"
                + e.getLocalizedMessage());
            sleepAndLogInterrupts(1000, "connecting to server");
          } catch(SocketTimeoutException e) {  // namenode is busy
            LOG.info("Problem connecting to server: " + nnAddr);
            sleepAndLogInterrupts(1000, "connecting to server");
          }
        }
    
        //  调用registrationSucceeded()方法确认namespaceId、 clusterId
        //  与其他Namenode返回的信息一致
        //  并且确认DatanodeUUid与Datanode本地存储一致
        LOG.info("Block pool " + this + " successfully registered with NN");
        bpos.registrationSucceeded(this, bpRegistration);
    
        // reset lease id whenever registered to NN.
        // ask for a new lease id at the next heartbeat.
        fullBlockReportLeaseId = 0;
    
        // 对块汇汇报做一个延迟  :  随机短延迟-有助于分散所有DNs的BR
        // random short delay - helps scatter the BR from all DNs
        scheduler.scheduleBlockReport(dnConf.initialBlockReportDelayMs);
      }

## 2.9.offerService() ##

offerService()方法是BPServiceActor的主循环方法， 它用于向Namenode发送心跳、 块汇报、 缓存汇报以及增量汇报。 offerService()方法会一直循环运行， 直到Datanode关闭或者客户端调用ClientDatanodeProtocol.refreshNodes()重新加载Namenode配置。

**定期发送心跳**  
offerService()方法会以dnConf.heartBeatInterval（默认配置是3秒） 间隔向Namenode发送心跳.  
offerService()是通过调用DatanodeProtocol.sendHeartbeat()方法向Namenode发送心跳的。 Datanode向Namenode发送的心跳信息主要包括： Datanode注册信息、 Datanode存储信息（使用容量， 剩余容量等） 、 缓存信息、 当前Datanode写文件的连接数， 以及读写数据使用的线程数等。

HeartbeatResponse response = bpNamenode.sendHeartbeat(bpRegistration,
            reports,
            dn.getFSDataset().getCacheCapacity(),
            dn.getFSDataset().getCacheUsed(),
            dn.getXmitsInProgress(),
            dn.getXceiverCount(),
            numFailedVolumes,
            volumeFailureSummary,
            requestBlockReportLease,
            slowPeers,
            slowDisks);

Namenode收到Datanode的心跳之后， 会返回一个心跳响应HeartbeatResponse。 这个心跳响应中包含一个DatanodeCommand的数组， 用来携带Namenode对Datanode的名字节点指令。 同时心跳响应中还包含一个NNHAStatusHeartbeat对象， 用来标识当前Namenode的  
HA状态。 Datanode会使用这个字段来确定BPOfferService当中的哪一个BPServiceActor对应的Namenode是Active状态的。

/**
     * Response to {@link DatanodeProtocol#sendHeartbeat}
     */
    public class HeartbeatResponse {
      /** Commands returned from the namenode to the datanode */
      private final DatanodeCommand[] commands;
      
      /** Information about the current HA-related state of the NN */
      private final NNHAStatusHeartbeat haStatus;
    
      private final RollingUpgradeStatus rollingUpdateStatus;
    
      private final long fullBlockReportLeaseId;
    
      // 代码略....
    }

offerService()方法中发送心跳流程部分的代码。 offerService()方法首先调用DatanodeProtocol.sendHeartbeat()方法向Namenode发送一个心跳。 然后调用bpos.updateActorStatesFromHeartbeat()方法处理心跳响应当中的HA状态字段NNHAStatusHeartbeat， 也就是更新BPOfferService中的Active Namenode的引用。 最后调用processCommand()方法执行Namenode通过心跳返回的名字节点指令。

offerService()方法就会调用processCommand()方法执行Namenode返回的名字节点指令。 processCommand()方法首先遍历心跳响应中携带的指令， 然后调用BPOfferService.processCommandFromActor()方法处理指令。

/**
       * Update the BPOS's view of which NN is active, based on a heartbeat
       * response from one of the actors.
       * 
       * @param actor the actor which received the heartbeat
       * @param nnHaState the HA-related heartbeat contents
       */
      void updateActorStatesFromHeartbeat(
          BPServiceActor actor,
          NNHAStatusHeartbeat nnHaState) {
        writeLock();
        try {
          // Namenode携带的txid
          final long txid = nnHaState.getTxId();
    
    
          //当前Namenode是否声明自己为Active Namenode
          final boolean nnClaimsActive =
              nnHaState.getState() == HAServiceState.ACTIVE;
    
    
          final boolean bposThinksActive = bpServiceToActive == actor;
    
          //当前Namenode携带的txid是否大于原Active Namenode携带的txid
          final boolean isMoreRecentClaim = txid > lastActiveClaimTxId;
    
          //原来的Standby Namenode声明自己为Active Namenode， 发生状态切换
          if (nnClaimsActive && !bposThinksActive) {
            LOG.info("Namenode " + actor + " trying to claim ACTIVE state with " +
                "txid=" + txid);
            if (!isMoreRecentClaim) {
              // Split-brain scenario - an NN is trying to claim active
              // state when a different NN has already claimed it with a higher
              // txid.
              LOG.warn("NN " + actor + " tried to claim ACTIVE state at txid=" +
                  txid + " but there was already a more recent claim at txid=" +
                  lastActiveClaimTxId);
              return;
            } else {
              if (bpServiceToActive == null) {
                LOG.info("Acknowledging ACTIVE Namenode " + actor);
              } else {
                LOG.info("Namenode " + actor + " taking over ACTIVE state from " +
                    bpServiceToActive + " at higher txid=" + txid);
              }
              bpServiceToActive = actor;
            }
          } else if (!nnClaimsActive && bposThinksActive) {
            LOG.info("Namenode " + actor + " relinquishing ACTIVE state with " +
                "txid=" + nnHaState.getTxId());
            bpServiceToActive = null;
          }
    
          //将bpServiceToActive指向当前Namenode对应的BPServiceActor
          if (bpServiceToActive == actor) {
            assert txid >= lastActiveClaimTxId;
            lastActiveClaimTxId = txid;
          }
        } finally {
          writeUnlock();
        }
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/9b061b49f18e4ef8ba05782e9a12088f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/151b9b67dcdb45939c5493a200e0d266.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201117155010806.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center