NameNode中数据节点的保存(1)——Host2NodesMap

对于一台主机，我们可以在它上面部署多个DataNode进程，这也就是说在一台机器上有多个DataNode节点，而且这些DataNode节点属于同一个HDFS集群，那么这里就有一个问题了，NameNode节点是如何考虑整个集群的负载均衡的？如果NameNode节点以DataNode节点为单位来考虑负载均衡的话，就会出现包含有多个DataNode节点的主机负载过重，所以就不得不以主机为单位来计算集群的负载情况了。在NameNode中用Host2NodesMap类来存储主机与DataNode节点之间的映射。

当一个DataNode节点向NameNode注册成功的时候，NameNode就会把这个DataNode节点存储到它的host2DataNodeMap属性中，也就是Host2NodesMap类的一个实例，这个类主要包含三个属性：



map：存储集群中所有主机上的所有DataNode节点；

r：用来随机选择给定主机上的某一个DataNode节点；

hostmapLock：控制对map的同步操作；

Host2NodesMap类主要负责对集群中的DataNode节点按在它们所在的主机进行分类管理，它可用来添加、删除、查询一个DataNode节点，它也可以按照主机或者DataNode的名字来查询。这些操作对应的方法是：

//判断一个DataNode节点在不在集群中

boolean contains(DatanodeDescriptor node) {

if (node==null) {

return false;

}

String host = node.getHost();

hostmapLock.readLock().lock();

try {

DatanodeDescriptor[] nodes = map.get(host);

if (nodes != null) {

for(DatanodeDescriptor containedNode:nodes) {

if (node==containedNode) {

return true;

}

}

}

} finally {

hostmapLock.readLock().unlock();

}

return false;

}

//添加一个数据节点

boolean add(DatanodeDescriptor node) {

hostmapLock.writeLock().lock();

try {

if (node==null || contains(node)) {

return false;

}

//找到DataNode节点所在的host

String host = node.getHost();

DatanodeDescriptor[] nodes = map.get(host);

DatanodeDescriptor[] newNodes;

if (nodes==null) {

newNodes = new DatanodeDescriptor[1];

newNodes[0]=node;

} else { // rare case: more than one datanode on the host

newNodes = new DatanodeDescriptor[nodes.length+1];

System.arraycopy(nodes, 0, newNodes, 0, nodes.length);

newNodes[nodes.length] = node;

}

map.put(host, newNodes);

return true;

} finally {

hostmapLock.writeLock().unlock();

}

}

//删除一个节点

boolean remove(DatanodeDescriptor node) {

if (node==null) {

return false;

}

String host = node.getHost();

hostmapLock.writeLock().lock();

try {

DatanodeDescriptor[] nodes = map.get(host);

if (nodes==null) {

return false;

}

if (nodes.length==1) {

if (nodes[0]==node) {

map.remove(host);

return true;

} else {

return false;

}

}

//rare case

int i=0;

for(; i<nodes.length; i++) {

if (nodes[i]==node) {

break;

}

}

if (i==nodes.length) {

return false;

} else {

DatanodeDescriptor[] newNodes;

newNodes = new DatanodeDescriptor[nodes.length-1];

System.arraycopy(nodes, 0, newNodes, 0, i);

System.arraycopy(nodes, i+1, newNodes, i, nodes.length-i-1);

map.put(host, newNodes);

return true;

}

} finally {

hostmapLock.writeLock().unlock();

}

}

//根据主机名获取这个主机上的一个DataNode节点，如果这个主机上有多个DataNode节点，则随机选一个

DatanodeDescriptor getDatanodeByHost(String host) {

if (host==null) {

return null;

}

hostmapLock.readLock().lock();

try {

DatanodeDescriptor[] nodes = map.get(host);

// no entry

if (nodes== null) {

return null;

}

// one node

if (nodes.length == 1) {

return nodes[0];

}

// more than one node

return nodes[r.nextInt(nodes.length)];

} finally {

hostmapLock.readLock().unlock();

}

}

//根据DataNode节点的名字来找到这个节点

public DatanodeDescriptor getDatanodeByName(String name) {

if (name==null) {

return null;

}

int colon = name.indexOf(":");

String host;

if (colon < 0) {

host = name;

} else {

host = name.substring(0, colon);

}

hostmapLock.readLock().lock();

try {

DatanodeDescriptor[] nodes = map.get(host);

// no entry

if (nodes== null) {

return null;

}

for(DatanodeDescriptor containedNode:nodes) {

if (name.equals(containedNode.getName())) {

return containedNode;

}

}

return null;

} finally {

hostmapLock.readLock().unlock();

}

}

另外还有一个问题，刚才说了，NameNode节点如果只以主机为单位来评估集群的负载情况，其实也是不合理的。例如，在实际的应用中，如果在某个集群里面存在主机配置参差不齐的情况，则对于那些配置很高的主机来说，它们的很多资源会处于空闲状态，其工作量相对于那些配置较低的主机来说已经饱和了。正常的情况下，集群的管理人员会在配置差的主机上部署少量的DataNode节点，在配置高的主机上部署较多的DataNode节点，因次，NameNode节点就不能仅仅只通过主机工作量来评估集群的负载情况了。至于NameNode是如何进行集群的负载均衡的，我会在以后的文章中详细介绍，但是它的实现也不一定很好，大家期望也不要太高。

遗憾的是，Hadoop-0.2.0版本并没有这样考虑集群的负载情况，它用Host2NodesMap结构主要是能够根据某个客户端来获取其上的DataNode节点。但是，我上面猜测的并不是没有任何意义，或许将来Hadoop会把主机负载纳入到集群负载的考虑范围。