[hadoop源码阅读][8]-datanode-datanode

datanode的定义

public class DataNode extends Configured implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable, DataNodeMXBean

DataNode实现了两个通信接口，其中ClientDatanodeProtocol是用于和Client交互的，InterDatanodeProtocol，是DataNode间的通信接口。ipcServer是DataNode的一个成员变量，它启动了一个IPC服务，这样，DataNode就能提供ClientDatanodeProtocol和InterDatanodeProtocol的能力了。

datanode的启动

从main函数开始,一路往下,实际干活的是在函数startDataNode里.

datanode的运行

datanode的线程循环函数run,不过干活的主要是offerService

offerService这个方法是如何执行的：


接下来看processCommand方法是如何处理命令的，这些操作在DatanodeProtocol中定义：

//  Main loop for the DataNode. Runs until shutdown, forever calling remote NameNode functions.
public void offerService() throws Exception
{
while (shouldRun)
{
try
{
long startTime = now();
if (startTime - lastHeartbeat > heartBeatInterval)//定期发送心跳
{
DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration, data.getCapacity(), data.getDfsUsed(), data.getRemaining(), xmitsInProgress.get(), getXceiverCount());
if (!processCommand(cmds))
continue;
}

blockArray = receivedBlockList.toArray(new Block[numBlocks]);// receivedBlockList表明在这个DataNode成功创建的新的数据块，而delHints，是可以删除该数据块的节点
delHintArray = delHints.toArray(new String[numBlocks]);//在datanode.notifyNamenodeReceivedBlock函数中发生变化
namenode.blockReceived(dnRegistration, blockArray, delHintArray);//Block状态变化报告通过NameNode.blockReceived来报告。

if (startTime - lastBlockReport > blockReportInterval)// 向namenode报告系统中Block状态的变化
{
Block[] bReport = data.getBlockReport();
DatanodeCommand cmd = namenode.blockReport(dnRegistration, BlockListAsLongs.convertToArrayLongs(bReport));
processCommand(cmd);
}

if (blockScanner != null && blockScannerThread == null && upgradeManager.isUpgradeCompleted())// start block scanner
{
blockScannerThread = new Daemon(blockScanner);//启动blockScanner线程 进行block扫描
blockScannerThread.start();
}

// There is no work to do; sleep until hearbeat timer elapses,or work arrives, and then iterate again.
long waitTime = heartBeatInterval - (System.currentTimeMillis() - lastHeartbeat);
synchronized (receivedBlockList)
{
if (waitTime > 0 && receivedBlockList.size() == 0)
{
try
{
receivedBlockList.wait(waitTime);
}
catch (InterruptedException ie)
{
}
delayBeforeBlockReceived();
}
} // synchronized
}
catch (RemoteException re)
{
}
catch (IOException e)
{
}
} // while (shouldRun)
} // offerService

类图





其他的

datanode一路粗略的看完,还有一些问题,我想关注的,留待后面继续,



参考url

http://blog.jeoygin.org/2012/03/hdfs-source-analysis-datanode-startup-service.html

http://caibinbupt.iteye.com/blog/287870