第七章：小朱笔记hadoop之源码分析-hdfs分析 第五节：Datanode 分析

第七章：小朱笔记hadoop之源码分析-hdfs分析

第五节：Datanode 分析

5.1 Datanode 启动过程分析

5.1 Datanode 启动过程分析

（1）shell脚本启动DataNode

start-dfs.sh  
"$bin"/hadoop-daemons.sh --config $HADOOP_CONF_DIR start datanode $dataStartOpt 

（2）main()函数启动分析

        主线程阻塞，让DataNode的任务循环执行，调用createDataNode方法创建datanode,等datanode线程结束 。

    //主线程阻塞，让DataNode的任务循环执行  
     public static void secureMain(String [] args, SecureResources resources) {  
       try {  
         LOG.info("start up datanode...");  
         if(null!=args){   
           for(String arg:args) {  
             LOG.info("arg:"+arg);  
           }  
         }  
         StringUtils.startupShutdownMessage(DataNode.class, args, LOG);  
         DataNode datanode = createDataNode(args, null, resources);  
         if (datanode != null)  
           datanode.join();  
       } catch (Throwable e) {  
         LOG.error(StringUtils.stringifyException(e));  
         System.exit(-1);  
       } finally {  
         // We need to add System.exit here because either shutdown was called or  
         // some disk related conditions like volumes tolerated or volumes required  
         // condition was not met. Also, In secure mode, control will go to Jsvc and  
         // the process hangs without System.exit.  
         LOG.info("Exiting Datanode");  
         System.exit(0);  
       }  
     }  

（3）创建Datanode实例，并启动Datanode线程

  调用instantiateDataNode方法初始化datanode,调用runDatanodeDaemon方法运行datanode线程.

    /** Instantiate & Start a single datanode daemon and wait for it to finish. 
     *  If this thread is specifically interrupted, it will stop waiting. 
     *  LimitedPrivate for creating secure datanodes 
     */  
    public static DataNode createDataNode(String args[],  
              Configuration conf, SecureResources resources) throws IOException {  
       
       //初始化DataNode  
      DataNode dn = instantiateDataNode(args, conf, resources);  
      
      runDatanodeDaemon(dn);  
      //进行DataNode注册，创建线程，设置守护线程，启动线程  
      return dn;  
    }  

（4）实例化DataNode结点

  解析启动参数：
  如果设置了机架配置${dfs.network.script}，退出程序
  通过配置${dfs.data.dir}得到datanode的存储目录
  调用makeInstance方法创建实例
  makeInstance检查数据存储目录的合法性 并初始化DataNode对象
  DataNode构造函数中调用startDataNode根据具体配置文件的信息进行具体的初始化过程
  调用startDataNode方法启动datanode
  如果启动出错，调用shutdown方法关闭datanode

    /** 
      * Start a Datanode with specified server sockets for secure environments 
      * where they are run with privileged ports and injected from a higher 
      * level of capability 
      */  
     DataNode(final Configuration conf,  
              final AbstractList<File> dataDirs, SecureResources resources) throws IOException {  
       super(conf);  
       SecurityUtil.login(conf, DFSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY,   
           DFSConfigKeys.DFS_DATANODE_USER_NAME_KEY);  
      
       datanodeObject = this;  
       supportAppends = conf.getBoolean("dfs.support.append", false);  
       this.userWithLocalPathAccess = conf  
           .get(DFSConfigKeys.DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY);  
       try {  
         startDataNode(conf, dataDirs, resources);  
       } catch (IOException ie) {  
         shutdown();  
         throw ie;  
       }     
     }  

（5）创建Datanode实例

    /** 
      * This method starts the data node with the specified conf. 
      *  
      * @param conf - the configuration 
      *  if conf's CONFIG_PROPERTY_SIMULATED property is set 
      *  then a simulated storage based data node is created. 
      *  
      * @param dataDirs - only for a non-simulated storage data node 
      * @throws IOException 
      * @throws MalformedObjectNameException  
      * @throws MBeanRegistrationException  
      * @throws InstanceAlreadyExistsException  
      */  
     void startDataNode(Configuration conf,   
                        AbstractList<File> dataDirs, SecureResources resources  
                        ) throws IOException {  
       if(UserGroupInformation.isSecurityEnabled() && resources == null)  
         throw new RuntimeException("Cannot start secure cluster without " +  
                "privileged resources.");  
         
       this.secureResources = resources;  
       // use configured nameserver & interface to get local hostname  
       //设置machineName  
       if (conf.get("slave.host.name") != null) {  
         machineName = conf.get("slave.host.name");     
       }  
       if (machineName == null) {  
         machineName = DNS.getDefaultHost(conf.get("dfs.datanode.dns.interface","default"),conf.get("dfs.datanode.dns.nameserver","default"));  
       }  
         
       //获取nameNode的地址信息    
       InetSocketAddress nameNodeAddr = NameNode.getServiceAddress(conf, true);  
         
       //setSocketout时间    
       this.socketTimeout =  conf.getInt("dfs.socket.timeout",HdfsConstants.READ_TIMEOUT);  
       this.socketWriteTimeout = conf.getInt("dfs.datanode.socket.write.timeout",HdfsConstants.WRITE_TIMEOUT);  
         
       /* Based on results on different platforms, we might need set the default  
        * to false on some of them. */  
       this.transferToAllowed = conf.getBoolean("dfs.datanode.transferTo.allowed", true);  
         
       //写包的大小，默认64K    
       this.writePacketSize = conf.getInt("dfs.write.packet.size", 64*1024);  
      
       //创建本地socketaddress地址    
       InetSocketAddress socAddr = DataNode.getStreamingAddr(conf);  
       int tmpPort = socAddr.getPort();  
         
       //DataStorage保存了存储相关的信息  
       storage = new DataStorage();// Data storage information file. 数据存储信息文件    
         
       //构造一个注册器  
       // construct registration  
       this.dnRegistration = new DatanodeRegistration(machineName + ":" + tmpPort);  
      
       //通过动态代理生成namenode实例  
       // connect to name node  
       this.namenode = (DatanodeProtocol) RPC.waitForProxy(DatanodeProtocol.class,DatanodeProtocol.versionID,nameNodeAddr, conf);  
         
       // get version and id info from the name-node   
       // 从名称节点获取版本和id信息  主要包含buildVersin和distributeUpgradeVersion,用于版本检验  
       NamespaceInfo nsInfo = handshake();  
         
         
       StartupOption startOpt = getStartupOption(conf);  
       assert startOpt != null : "Startup option must be set.";  
         
       boolean simulatedFSDataset = conf.getBoolean("dfs.datanode.simulateddatastorage", false);  
         
       //判断一下是否是伪分布式，否则走正常判断，此处分析正常逻辑  
       if (simulatedFSDataset) {  
           setNewStorageID(dnRegistration);  
           dnRegistration.storageInfo.layoutVersion = FSConstants.LAYOUT_VERSION;  
           dnRegistration.storageInfo.namespaceID = nsInfo.namespaceID;  
           // it would have been better to pass storage as a parameter to  
           // constructor below - need to augment ReflectionUtils used below.  
           conf.set("StorageId", dnRegistration.getStorageID());  
           try {  
             //Equivalent of following (can't do because Simulated is in test dir)  
             //  this.data = new SimulatedFSDataset(conf);  
             this.data = (FSDatasetInterface) ReflectionUtils.newInstance(  
                 Class.forName("org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset"), conf);  
           } catch (ClassNotFoundException e) {  
             throw new IOException(StringUtils.stringifyException(e));  
           }  
       } else { // real storage  
         // read storage info, lock data dirs and transition fs state if necessary  
         storage.recoverTransitionRead(nsInfo, dataDirs, startOpt);  
         // adjust  
         //将storage进行信息注册  
         this.dnRegistration.setStorageInfo(storage);  
           
         // initialize data node internal structure  
         //根据storage和conf信息，生成FSDataset，用于数据块操作    
         this.data = new FSDataset(storage, conf);  
       }  
           
       // register datanode MXBean  
       this.registerMXBean(conf); // register the MXBean for DataNode  
         
       // Allow configuration to delay block reports to find bugs  
       artificialBlockReceivedDelay = conf.getInt(  
           "dfs.datanode.artificialBlockReceivedDelay", 0);  
      
       // find free port or use privileged port provide  
       //初始化Socket服务器端，区分NIO和IO    
       ServerSocket ss;  
       if(secureResources == null) {  
         ss = (socketWriteTimeout > 0) ?   
           ServerSocketChannel.open().socket() : new ServerSocket();  
         Server.bind(ss, socAddr, 0);  
       } else {  
         ss = resources.getStreamingSocket();  
       }  
       //设置接收的buffer缓存大小，默认64K  
       ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE);   
         
       // adjust machine name with the actual port  
       tmpPort = ss.getLocalPort();  
       selfAddr = new InetSocketAddress(ss.getInetAddress().getHostAddress(),  
                                        tmpPort);  
       this.dnRegistration.setName(machineName + ":" + tmpPort);  
       LOG.info("Opened info server at " + tmpPort);  
         
       //服务器 用于接收/发送 一个数据块 。 这是创建监听来自客户或其他 DataNodes 的 请求 。 这种小型服务器不使用 thHadoop IPC机制  
       //初始化处理类dataXceiverServer    
       this.threadGroup = new ThreadGroup("dataXceiverServer");  
       this.dataXceiverServer = new Daemon(threadGroup, new DataXceiverServer(ss, conf, this));  
       this.threadGroup.setDaemon(true); // auto destroy when empty  
      
       //分别设置块状态信息间隔时间和心跳间隔时间  
       this.blockReportInterval = conf.getLong("dfs.blockreport.intervalMsec", BLOCKREPORT_INTERVAL);  
       this.initialBlockReportDelay = conf.getLong("dfs.blockreport.initialDelay",BLOCKREPORT_INITIAL_DELAY)* 1000L;   
       if (this.initialBlockReportDelay >= blockReportInterval) {  
         this.initialBlockReportDelay = 0;  
         LOG.info("dfs.blockreport.initialDelay is greater than " +  
           "dfs.blockreport.intervalMsec." + " Setting initial delay to 0 msec:");  
       }  
       this.heartBeatInterval = conf.getLong("dfs.heartbeat.interval", HEARTBEAT_INTERVAL) * 1000L;  
         
         
       DataNode.nameNodeAddr = nameNodeAddr;  
      
       //initialize periodic block scanner  
       String reason = null;  
       if (conf.getInt("dfs.datanode.scan.period.hours", 0) < 0) {  
         reason = "verification is turned off by configuration";  
       } else if ( !(data instanceof FSDataset) ) {  
         reason = "verifcation is supported only with FSDataset";  
       }   
       if ( reason == null ) {  
         //初始化一个定期检查scanner blockScanner用于定时对文件块进行扫描    
         blockScanner = new DataBlockScanner(this, (FSDataset)data, conf);  
       } else {  
         LOG.info("Periodic Block Verification is disabled because " +  
                  reason + ".");  
       }  
      
       //create a servlet to serve full-file content  
       //创建HttpServer，内部用jetty实现，用于页面监控    
       InetSocketAddress infoSocAddr = DataNode.getInfoAddr(conf);  
       String infoHost = infoSocAddr.getHostName();  
       int tmpInfoPort = infoSocAddr.getPort();  
       this.infoServer = (secureResources == null)   
          ? new HttpServer("datanode", infoHost, tmpInfoPort, tmpInfoPort == 0,   
              conf, SecurityUtil.getAdminAcls(conf, DFSConfigKeys.DFS_ADMIN))  
          : new HttpServer("datanode", infoHost, tmpInfoPort, tmpInfoPort == 0,  
              conf, SecurityUtil.getAdminAcls(conf, DFSConfigKeys.DFS_ADMIN),  
              secureResources.getListener());  
       if (conf.getBoolean("dfs.https.enable", false)) {  
         boolean needClientAuth = conf.getBoolean("dfs.https.need.client.auth", false);  
         InetSocketAddress secInfoSocAddr = NetUtils.createSocketAddr(conf.get(  
             "dfs.datanode.https.address", infoHost + ":" + 0));  
         Configuration sslConf = new Configuration(false);  
         sslConf.addResource(conf.get("dfs.https.server.keystore.resource",  
             "ssl-server.xml"));  
         this.infoServer.addSslListener(secInfoSocAddr, sslConf, needClientAuth);  
       }  
       this.infoServer.addInternalServlet(null, "/streamFile/*", StreamFile.class);  
       this.infoServer.addInternalServlet(null, "/getFileChecksum/*",  
           FileChecksumServlets.GetServlet.class);  
      
       this.infoServer.setAttribute("datanode", this);  
       this.infoServer.setAttribute("datanode.blockScanner", blockScanner);  
       this.infoServer.setAttribute(JspHelper.CURRENT_CONF, conf);  
       this.infoServer.addServlet(null, "/blockScannerReport",   
                                  DataBlockScanner.Servlet.class);  
      
       if (WebHdfsFileSystem.isEnabled(conf, LOG)) {  
         infoServer.addJerseyResourcePackage(DatanodeWebHdfsMethods.class  
             .getPackage().getName() + ";" + Param.class.getPackage().getName(),  
             WebHdfsFileSystem.PATH_PREFIX + "/*");  
       }  
       this.infoServer.start();  
         
         
       // adjust info port  
       this.dnRegistration.setInfoPort(this.infoServer.getPort());  
       myMetrics = DataNodeInstrumentation.create(conf,  
                                                  dnRegistration.getStorageID());  
         
       // set service-level authorization security policy  
       if (conf.getBoolean(  
             ServiceAuthorizationManager.SERVICE_AUTHORIZATION_CONFIG, false)) {  
         ServiceAuthorizationManager.refresh(conf, new HDFSPolicyProvider());  
       }  
      
       // BlockTokenSecretManager is created here, but it shouldn't be  
       // used until it is initialized in register().  
       this.blockTokenSecretManager = new BlockTokenSecretManager(false,0, 0);  
         
       //init ipc server  
       //开启本地ipc服务，监听来自client和其它    
       InetSocketAddress ipcAddr = NetUtils.createSocketAddr(conf.get("dfs.datanode.ipc.address"));  
       ipcServer = RPC.getServer(this, ipcAddr.getHostName(), ipcAddr.getPort(),   
           conf.getInt("dfs.datanode.handler.count", 3), false, conf,  
           blockTokenSecretManager);  
         
       dnRegistration.setIpcPort(ipcServer.getListenerAddress().getPort());  
      
       LOG.info("dnRegistration = " + dnRegistration);  
     }  

（6）运行DataNode结点

进行DataNode注册，创建线程，设置守护线程，启动线程。

    /** Start a single datanode daemon and wait for it to finish. 
     *  If this thread is specifically interrupted, it will stop waiting. 
     */  
    public static void runDatanodeDaemon(DataNode dn) throws IOException {  
      if (dn != null) {  
        //register datanode  
        dn.register();  
        dn.dataNodeThread = new Thread(dn, dnThreadName);  
        dn.dataNodeThread.setDaemon(true); // needed for JUnit testing  
        dn.dataNodeThread.start();  
      }  
    }  

 启动DataXceiverServer，然后进入datanode的正常运行。检查是否需要升级，调用offerService方法提供服务.

    public void run() {  
       LOG.info(dnRegistration + "In DataNode.run, data = " + data);  
       ///启动数据块的流读写服务器，  
       // start dataXceiveServer  
       dataXceiverServer.start();  
       //内部hadoop ipc服务器  
       ipcServer.start();  
             
       while (shouldRun) {  
         try {  
           //检测是否需要升级hadoop文件系统  
           startDistributedUpgradeIfNeeded();  
           //DataNode提供服务，定时发送心跳给NameNode,响应NameNode返回的命令并执行  
           offerService();  
         } 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();  
     }