Hadoop学习二十一：Hadoop-Hdfs DataNode 源码

link

一. DataNode类图

二. DateNode属性说明

•  DatanodeProtocol namenode：RPC代理类。RPC.getProxy(nameNodeAddress)得到远程NameNode代理类。后续的versionRequest()，register()，sendHeartbeat()，blockReceived()，blockReport()都是namenode的行为。
  
• org.apache.hadoop.ipc.Server ipcServer：RPC Server。conf里配置ipcAddr=0.0.0.0:50020，ipcServer = RPC.getServer(this, ipcAddr.getHostName()...)，此DataNode作为RPC Server提供远程服务，提供接口ClientDatanodeProtocol和InterDatanodeProtocol中定义的服务。
  
• HttpServer infoServer：a Jetty embedded server to answer http requests。
  
• DataStorage storage：参考http://zy19982004.iyunv.com/admin/blogs/1878758。
  
• DatanodeRegistration dnRegistration：DatanodeRegistration class conatins all information the Namenode needs to identify and verify a Datanode when it contacts the Namenode.后续的sendHeartbeat()，blockReceived()，blockReport()方法第一个参数就是DatanodeRegistration 。
  
• FSDatasetInterface data：FSDataset的接口，管理着DataNode的所有block。参考http://zy19982004.iyunv.com/admin/blogs/1880303。
  
• DataBlockScanner blockScanner：定时对数据块文件进行校验。
  
• DataXceiverServer dataXceiverServer：参考http://zy19982004.iyunv.com/admin/blogs/1881117。
  
• LinkedList<Block> receivedBlockList：成功创建的新block。
  
• LinkedList<String> delHints：可以删掉该数据块的节点。DataXceiver.replaceBlock()和writeBlock成功后会调用DataNode.notifyNamenodeReceivedBlock()方法给这两个List add value.
  protected void notifyNamenodeReceivedBlock(Block block, String delHint) {
  synchronized (receivedBlockList) {
  synchronized (delHints) {
  receivedBlockList.add(block);
  delHints.add(delHint);
  receivedBlockList.notifyAll();
  }
  }
  }
   
  

三. 代码顺序阅读

•  main()函数入口。
    public static void main(String args[]) {
  secureMain(args, null);
  }
  
• secureMain()：初始化DataNode；加入线程里。
  public static void secureMain(String [] args, SecureResources resources) {
  DataNode datanode = createDataNode(args, null, resources);
  if (datanode != null)
  datanode.join();
  }
  
• createDataNode()：初始化DataNode；启动该线程。
    public static DataNode createDataNode(String args[],
  Configuration conf, SecureResources resources) throws IOException {
  DataNode dn = instantiateDataNode(args, conf, resources);
  runDatanodeDaemon(dn);
  return dn;
  }
  
• instantiateDataNode()：读取dataDirs；调用makeInstance()。makeInstance()：check dataDirs；new DateNode。new DataNode()：调用startDataNode()。
  public static DataNode instantiateDataNode(String args[],
  Configuration conf,
  SecureResources resources) throws IOException {
  String[] dataDirs = conf.getStrings(DATA_DIR_KEY);
  return makeInstance(dataDirs, conf, resources);
  }
  public static DataNode makeInstance(String[] dataDirs, Configuration conf,
  SecureResources resources) throws IOException {
  DiskChecker.checkDir(localFS, new Path(dir), dataDirPermission);
  return new DataNode(conf, dirs, resources);
  }
  DataNode(final Configuration conf,
  final AbstractList<File> dataDirs, SecureResources resources) throws IOException {
  startDataNode(conf, dataDirs, resources);
  }
  
• startDataNode()：1.根据conf对象初始化各个ip port；2.初始化DataStorage；3.初始化DatanodeRegistration；4.初始化DatanodeProtocol；5.handshake() get version and id info from the name-node，实际调用namenode.versionRequest()；6.DataStorage.recoverTransitionRead(...)检查文件系统的状态并做恢复；7.初始化FSDatasetInterface；8.初始化DataBlockScanner；9.创建serverSocket，据此初始化DataXceiverServer；10.HttpServer，配置参数然后启动；11.初始化ipcServer；
  
• 接3，runDatanodeDaemon()：注册DataNode；启动DataNode线程。
    public static void runDatanodeDaemon(DataNode dn) throws IOException {
  if (dn != null) {
  dn.register();
  dn.dataNodeThread = new Thread(dn, dnThreadName);
  dn.dataNodeThread.setDaemon(true); // needed for JUnit testing
  dn.dataNodeThread.start();
  }
  }
  private void register() throws IOException {
  dnRegistration = namenode.register(dnRegistration);
  } 
  
• 进入run()：启动DataXceiverServer线程；启动ipcServer；startDistributedUpgradeIfNeeded() Start distributed upgrade if it should be initiated by the data-node；offerService()开始工作。
    public void run() {
  LOG.info(dnRegistration + "In DataNode.run, data = " + data);
  dataXceiverServer.start();
  ipcServer.start();
  while (shouldRun) {
  startDistributedUpgradeIfNeeded();
  offerService();
  }
  
• offerService()：1.定期heartBeatInterval调用namenode.sendHeartbeat()发送DataNode信息的心跳，并执行带回的DatanodeCommand；2.一直调用namenode.blockReceived()向Namenode报告最近接收到的数据块、要删除的多余块副本；3.定期调用namenode.blockReport()发送block report 告诉NameNode此DataNode上的block信息，并执行带回的DatanodeCommand。4.启动blockScannerThread线程。
  public void offerService() throws Exception {
  
  while (shouldRun) {
  try {
  long startTime = now();
  if (startTime - lastHeartbeat > heartBeatInterval) {
  lastHeartbeat = startTime;
  1.DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
  data.getCapacity(),
  data.getDfsUsed(),
  data.getRemaining(),
  xmitsInProgress.get(),
  getXceiverCount());
  processCommand(cmds)；
  }
  2.namenode.blockReceived(dnRegistration, blockArray, delHintArray);
  blockReceived成功后删除receivedBlockListhe delHints里的数据
  if (startTime - lastBlockReport > blockReportInterval) {
  3.DatanodeCommand cmd = namenode.blockReport(dnRegistration,
  BlockListAsLongs.convertToArrayLongs(bReport));
  processCommand(cmd);
  }
  blockScannerThread = new Daemon(blockScanner);
  blockScannerThread.start();
  } // while (shouldRun)
  } // offerService
  }