Hadoop DataNode启动之asyncDiskService

  客户端在删除HDFS里的文件时，首先是更新NN里的元数据，然后通过心跳向DN发送块删除指令，DN接到指令执行块删除操作，和异步块报告一样，这个过程也是异步的，原因是出于性能和实时性的考虑，在DN中，每个文件卷都有一个专门的线程池来接收删除指令，该池的默认删除线程有4个，从名字可以看出，该类并不只是用于块删除操作的，在以后的版本中可能会增加其他的服务线程。该类在异步块报告线程启动后紧跟创建，下面分析下这个类。

public FSDataset(DataStorage storage, Configuration conf) throwsIOException {

.....

for (int idx = 0; idx <storage.getNumStorageDirs(); idx++) {

roots[idx] =storage.getStorageDir(idx).getCurrentDir();

}

//异步磁盘服务类的创建

asyncDiskService = newFSDatasetAsyncDiskService(roots);

registerMBean(storage.getStorageID());

}

看下FSDatasetAsyncDiskService的构造函数，该类的代码比较简短

FSDatasetAsyncDiskService(File[] volumes) {

//创建一个线程工厂，用于产生空线程

threadFactory = newThreadFactory() {

public ThreadnewThread(Runnable r) {

return newThread(threadGroup, r);

}

};
// 为每个文件卷创建一个线程池

for (int v = 0 ; v <volumes.length; v++) {

//创建线程池，并以KV形式放入MAP中，核心线程默认一个，最大线程4个

ThreadPoolExecutor executor= new ThreadPoolExecutor(

CORE_THREADS_PER_VOLUME, MAXIMUM_THREADS_PER_VOLUME,

THREADS_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,

newLinkedBlockingQueue<Runnable>(), threadFactory);

// This can reduce thenumber of running threads

executor.allowCoreThreadTimeOut(true);

//以KV形式放入MAP中，到现在有文件卷，有线程池，那么里面的线程到底做些什么操作呢？

executors.put(volumes[v],executor);

}

}

在这里，我们只看块删除的操作，在与NN的心跳中，如果DN需要删除块，则走下面这个流程

offerService() ->processCommand(DatanodeCommandcmd) -> data.invalidate(toDelete)
-> asyncDiskService.deleteAsync(v, f, metaFile, dfsBytes,invalidBlks[i].toString())
以这个删除为例，看下deleteAsync函数

void deleteAsync(FSDataset.FSVolume volume, File blockFile,

File metaFile, long dfsBytes, String blockName) {

DataNode.LOG.info("Scheduling block " + blockName + " file " + blockFile

+ " for deletion");

//这里是个块删除的线程题，每当删除的时候，把他丢入线程池便会执行

ReplicaFileDeleteTask deletionTask =

new ReplicaFileDeleteTask(volume, blockFile,metaFile, dfsBytes,

blockName);

//开始执行块删除的操作

execute(volume.getCurrentDir(),deletionTask);

}

下面看看这个线程体的逻辑吧，只看构造方法和run方法就知道他的删除逻辑了

ReplicaFileDeleteTask(FSDataset.FSVolumevolume, File blockFile,

File metaFile, long dfsBytes, String blockName) {

this.volume = volume;         //文件卷名

this.blockFile = blockFile; //块文件名

this.metaFile = metaFile;    //meta文件名

this.dfsBytes = dfsBytes;   //删除文件总大小

this.blockName = blockName; //块名称

}

块删除线程执行体

public void run() {

//删除相关文件

if ( !blockFile.delete() || ( !metaFile.delete() && metaFile.exists() ) ) {

DataNode.LOG.warn("Unexpected error trying to deleteblock "

+ blockName + " at file " + blockFile + ".Ignored.");

} else {

//更新卷组使用尺寸

volume.decDfsUsed(dfsBytes);

DataNode.LOG.info("Deleted block " + blockName + " at file" + blockFile);

}

}

}

异步磁盘服务的原理比较简单，就是利用线程池执行异步操作提高性能：1、提高了和NN通信的效率，先完成通信，后台异步删除
2、降低了频繁创建线程带来的性能消耗。