Hadoop中DataNode与NameNode之间的心跳机制

短命女 2022-03-20 01:58 473阅读 0赞

DataNode: 用于存储HDFS的数据，

public class DataNode extends Configured   
    implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable \{。。。\}

1,实现了InterDatanodeProtocol, ClientDatanodeProtocol来供与客户端以及其他DataNode的通信，接受请求。

2,实现Runnable 接口

首先是来看下run()方法

public void run() \{  
    LOG.info(dnRegistration + "In DataNode.run, data = " + data);  
  
  
    // start dataXceiveServer  
    dataXceiverServer.start();//启动    dataXceiverServer服务器  
          
    while (shouldRun) \{  
      try \{  
        startDistributedUpgradeIfNeeded();//Hadoop文件系统升级监测  
        offerService();//循环，主要是定时向NameNode发送心跳，响应并执行NameNode返回的命令  
      \} catch (Exception ex) \{  
        LOG.error("Exception: " + StringUtils.stringifyException(ex));  
        if (shouldRun) \{  
          try \{  
            Thread.sleep(5000);  
          \} catch (InterruptedException ie) \{  
          \}  
        \}  
      \}  
    \}  
          
    LOG.info(dnRegistration + ":Finishing DataNode in: "+data);  
    shutdown();  
  \}

①，    dataXceiverServer.start();开启数据监听守护进程，有数据到来时，开启一个DataXceiver线程

DataXceiverServer.java

public void run() \{  
    while (datanode.shouldRun) \{  
      try \{  
        Socket s = ss.accept();  
        s.setTcpNoDelay(true);  
        new Daemon(datanode.threadGroup,   
            new DataXceiver(s, datanode, this)).start();//启动DataXceiver  
      \} catch (SocketTimeoutException ignored) \{  
        // wake up to see if should continue to run  
      \} catch (IOException ie) \{  
        LOG.warn(datanode.dnRegistration + ":DataXceiveServer: "   
                                + StringUtils.stringifyException(ie));  
      \} catch (Throwable te) \{  
        LOG.error(datanode.dnRegistration + ":DataXceiveServer: Exiting due to:"   
                                 + StringUtils.stringifyException(te));  
        datanode.shouldRun = false;  
      \}  
    \}  
    try \{  
      ss.close();  
    \} catch (IOException ie) \{  
      LOG.warn(datanode.dnRegistration + ":DataXceiveServer: "   
                              + StringUtils.stringifyException(ie));  
    \}  
  \}

在DataXceiver中主要是从DataXceiverServer读写数据

\-------DataXceiver.java----------

/\*\*  
   \* Read/write data from/to the DataXceiveServer.  
   \*/  
  public void run() \{  
    DataInputStream in=null;   
    try \{  
      in = new DataInputStream(  
          new BufferedInputStream(NetUtils.getInputStream(s),   
                                  SMALL\_BUFFER\_SIZE));//获取NameNode的流信息  
      short version = in.readShort();//获取版本信息  
      if ( version != DataTransferProtocol.DATA\_TRANSFER\_VERSION ) \{  
        throw new IOException( "Version Mismatch" );  
      \}  
      boolean local = s.getInetAddress().equals(s.getLocalAddress());  
      byte op = in.readByte();  
      // Make sure the xciver count is not exceeded  
      int curXceiverCount = datanode.getXceiverCount();  
      if (curXceiverCount > dataXceiverServer.maxXceiverCount) \{  
        throw new IOException("xceiverCount " + curXceiverCount  
                              + " exceeds the limit of concurrent xcievers "  
                              + dataXceiverServer.maxXceiverCount);  
      \}  
      long startTime = DataNode.now();

//执行相应的操作  
      switch ( op ) \{  
      case DataTransferProtocol.OP\_READ\_BLOCK://读取block信息  
        readBlock( in );  
        datanode.myMetrics.readBlockOp.inc(DataNode.now() - startTime);  
        if (local)  
          datanode.myMetrics.readsFromLocalClient.inc();  
        else  
          datanode.myMetrics.readsFromRemoteClient.inc();  
        break;  
      case DataTransferProtocol.OP\_WRITE\_BLOCK://写block信息  
        writeBlock( in );  
        datanode.myMetrics.writeBlockOp.inc(DataNode.now() - startTime);  
        if (local)  
          datanode.myMetrics.writesFromLocalClient.inc();  
        else  
          datanode.myMetrics.writesFromRemoteClient.inc();  
        break;  
      case DataTransferProtocol.OP\_READ\_METADATA://读取元数据信息  
        readMetadata( in );  
        datanode.myMetrics.readMetadataOp.inc(DataNode.now() - startTime);  
        break;  
      case DataTransferProtocol.OP\_REPLACE\_BLOCK: // for balancing purpose; send to a destination  
        replaceBlock(in);  
        datanode.myMetrics.replaceBlockOp.inc(DataNode.now() - startTime);  
        break;  
      case DataTransferProtocol.OP\_COPY\_BLOCK://复制block信息  
            // for balancing purpose; send to a proxy source  
        copyBlock(in);  
        datanode.myMetrics.copyBlockOp.inc(DataNode.now() - startTime);  
        break;  
      case DataTransferProtocol.OP\_BLOCK\_CHECKSUM: //get the checksum of a block---校验块信息  
        getBlockChecksum(in);  
        datanode.myMetrics.blockChecksumOp.inc(DataNode.now() - startTime);  
        break;  
      default:  
        throw new IOException("Unknown opcode " + op + " in data stream");  
      \}  
    \} catch (Throwable t) \{  
      LOG.error(datanode.dnRegistration + ":DataXceiver",t);  
    \} finally \{  
      LOG.debug(datanode.dnRegistration + ":Number of active connections is: "  
                               + datanode.getXceiverCount());  
      IOUtils.closeStream(in);  
      IOUtils.closeSocket(s);  
      dataXceiverServer.childSockets.remove(s);  
    \}  
  \}

\-------------------------------------------------------------------------------分割线-------------------------------------------------------------

startDistributedUpgradeIfNeeded();  
        offerService();

这里还是如题说说心跳机制吧，心跳由DataNode发起，

心跳首先是由DataNode的一个线程开启的，一直循环调用  offerService();方法，DataNode在这个方法中发送心跳

while (shouldRun) \{  
      try \{  
        long startTime = now();  
  
  
        //  
        // Every so often, send heartbeat or block-report  
        //  
          
        if (startTime - lastHeartbeat > heartBeatInterval) \{  
          //  
          // All heartbeat messages include following info:  
          // -- Datanode name  
          // -- data transfer port  
          // -- Total capacity  
          // -- Bytes remaining  
          //  
          lastHeartbeat = startTime;

//定期发送心跳  
          DatanodeCommand\[\] cmds = namenode.sendHeartbeat(dnRegistration,  
                                                       data.getCapacity(),  
                                                       data.getDfsUsed(),  
                                                       data.getRemaining(),  
                                                       xmitsInProgress.get(),  
                                                       getXceiverCount());  
          myMetrics.heartbeats.inc(now() - startTime);  
          //LOG.info("Just sent heartbeat, with name " + localName);

//响应心跳执行返回的命令  
          if (!processCommand(cmds))  
            continue;  
        \}

在这个方法中通过调用NameNode的RPC调用namenode.sendHeartbeat（）发送心跳,之后就是处理返回的请求

\-------------------------------------再来看看NameNode端------------------------------------------------------------------------------

public DatanodeCommand\[\] sendHeartbeat(DatanodeRegistration nodeReg,  
                                       long capacity,  
                                       long dfsUsed,  
                                       long remaining,  
                                       int xmitsInProgress,  
                                       int xceiverCount) throws IOException \{  
    verifyRequest(nodeReg);  
    return namesystem.handleHeartbeat(nodeReg, capacity, dfsUsed, remaining,  
        xceiverCount, xmitsInProgress);  
  \}

最终是由namesystem来处理的，二namesystem就是FSNamesystem的一个实例（在NameNode.java中public FSNamesystem namesystem; ）

FSNamesystem处理完后返回一个an array of datanode commands 交由DataNode处理，

另外：

/\*\*  
   \* Periodically calls heartbeatCheck().  
   \*/  
  class HeartbeatMonitor implements Runnable \{  
    /\*\*  
     \*/  
    public void run() \{  
      while (fsRunning) \{  
        try \{  
          heartbeatCheck();  
        \} catch (Exception e) \{  
          FSNamesystem.LOG.error(StringUtils.stringifyException(e));  
        \}  
        try \{  
          Thread.sleep(heartbeatRecheckInterval);  
        \} catch (InterruptedException ie) \{  
        \}  
      \}  
    \}  
  \}

其中的HeartbeatMonitor 处理失效的DataNode,将其移除removeDatanode(nodeInfo);

发表评论取消回复

表情：

评论列表（有 0 条评论，473人围观）

还没有评论，来说两句吧...

相关阅读

相关 Hadoop如何解决namenode和datanode缺失问题

如果是namenode缺失，则重新格式化即可`bin/hdfs namenode -format`，但是慎重格式化。如果是datanode缺失，进入到/opt/Hadoop

客官°小女子只卖身不卖艺/ 2022年12月05日 10:13/ 0 赞/ 298 阅读

相关 NameNode和SecondaryNameNode、DataNode工作机制

本文目录题外话 1.NameNode和SecondaryNameNode工作机制 1.NameNode中的元数据是存储在哪里的？

秒速五厘米/ 2022年10月30日 00:33/ 0 赞/ 211 阅读

相关 Hadoop的NameNode与SecondaryNameNode,DataNode

NameNode：主要任务是接收客户端的读写服务，主要功能： 1,文件ownship和permission 2,文件包含哪些块 3,Block保存在哪个

秒速五厘米/ 2022年06月07日 08:35/ 0 赞/ 211 阅读

相关 hadoop心跳机制

主节点和从节点之间的通信是通过心跳机制（心跳实际上是一个RPC函数）实现的所谓“心跳”是一种形象化描述，指的是持续的按照一定频率在运行，类似于心脏在永无休止的跳动。心

绝地灬酷狼/ 2022年06月04日 04:44/ 0 赞/ 336 阅读

相关 Secondary Namenode的Check point机制以及Namenode、Datanode工作机制说明

目录前言： 1、NameNode的工作机制 2、DataNode的工作机制 3、Secondary N

深碍√TFBOYSˉ_/ 2022年06月02日 01:20/ 0 赞/ 185 阅读

相关 Hadoop 多次namenode -format造成datanode进程异常

之前在启动hadoop集群时，使用jps命令没有查看到datanode，但第一次启动集群时有datanode，查看datanode的相关日志，报Java IO异常“java.i

素颜马尾好姑娘i/ 2022年05月27日 12:30/ 0 赞/ 240 阅读

相关《hadoop学习》关于hdfs中的namenode和datanode详解

转载:https://www.cnblogs.com/forget-me-not/p/5749256.html namenode: Namenode是中心服务器，

待我称王封你为后i/ 2022年05月20日 04:13/ 0 赞/ 258 阅读

相关 NameNode与DataNode的多目录配置

NameNode与DataNode的多目录配置 1.为什么要配置NameNode的多目录？ 2.配置 3.为什么配置DataNode？ 1.为什么要

逃离我推掉我的手/ 2022年04月10日 09:30/ 0 赞/ 212 阅读

相关 Hadoop中DataNode与NameNode之间的心跳机制

DataNode: 用于存储HDFS的数据， public class DataNode extends Configured implements Inter

短命女/ 2022年03月20日 01:58/ 0 赞/ 474 阅读

相关 Hadoop名词详解：HDFS、NameNode和DataNode

概览 -------------------- 首先我们来认识一下HDFS， HDFS（Hadoop Distributed File System ）Hadoop分布

深藏阁楼爱情的钟/ 2022年02月04日 03:15/ 0 赞/ 344 阅读