概述
以下是本人研究源代码成果, 此文僅献给我和我的小伙伴们，不足之处，欢迎斧正-------------------------------------------------致谢道格等人！
注：hadoop版本0.20.2,有童鞋表示看代码头晕，所以本文采用纯文字描述，哥还特意为你们把字体调调颜色噢 ^ o ^上一篇文章，我们一起讨论了DFSClient初始化过程，下面我们一起讨论关于数据读写的详细过程，该过程对比前面，稍稍复杂一点点，
它们不但要与名字节点通信，还需要访问数据节点,读取数据过程中，名字节点提供了两种远程方法：
a,getBlockLocations：确定数据的位置
b,reportBadBlocks：向 名字节点 报告客户端发现的坏块
下面我们详细的分析下读取数据的详细过程
DFSClient的open()方法用于打开文件,检查文件系统是否已经打开后构造并返回一个DFSInputStream对象，接下来用户就可以通过这个对象读取HDFS文件数据
=========================================================================================================
--------------------------------------------------------------------------------------------------------------------------------------------
/**
DFSClient类的成员变量：
public static final Log LOG = LogFactory.getLog(DFSClient.class);
public static final Log LOG = LogFactory.getLog(DFSClient.class);
public static final int MAX_BLOCK_ACQUIRE_FAILURES = 3;
private static final int TCP_WINDOW_SIZE = 128 * 1024; // 128 KB
/**namenode在rpcNamenode基础上增加了失败重试的功能**/
public final ClientProtocol namenode;
private final ClientProtocol rpcNamenode;
/**unuix系统用户组信息**/
final UnixUserGroupInformation ugi;
/**标识DFSClient客户端是否正在运行**/
volatile boolean clientRunning = true;
Random r = new Random();
/**客户端的名称**/
final String clientName;
/**租约续约的检查线程**/
final LeaseChecker leasechecker = new LeaseChecker();
private Configuration conf;
/**数据块的默认大小**/
private long defaultBlockSize;
/**数据Block的默认副本数**/
private short defaultReplication;
/**创建socket连接的工厂类**/
private SocketFactory socketFactory;
/**socket连接的过期时间**/
private int socketTimeout;
/***通过socket向dataNode写入数据的超期时间**/
private int datanodeWriteTimeout;
/**数据包最多能达到64K字节**/
final int writePacketSize;//64K
/**收集文件系统统计信息的对象**/
private final FileSystem.Statistics stats;
private int maxBlockAcquireFailures;
---------------------------------------------------------------------------------------------------------------------------------------------
FileSystem的open抽象方法：
---------------------------------------------------------------------------------------------------------------------------------------------
/**
* 打开path对应文件的FSDataInputStream输入流
*/
public abstract FSDataInputStream open(Path f, int bufferSize)
throws IOException;
------------------------------------------------------------------------------------------------------------------------------------------------------------------------
DistributedFileSystem的open方法：
--------------------------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------------------
结合上下文我们可以看出：
文件系统是通过内部包装的DFSInputStream来完成 真正的读取数过程
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
DistributedFileSystem的open方法：
/**
* 打开指定文件，返回输入流
*/
public FSDataInputStream open(Path f, int bufferSize) throws IOException {
return new DFSClient.DFSDataInputStream(
dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics));
}
--------------------------------------------------------------------------------------------------------------------------------------------
open方法：返回一个
DFSInputStream
DFSInputStream open(String src, int buffersize, boolean verifyChecksum,
DFSInputStream open(String src, int buffersize, boolean verifyChecksum,
FileSystem.Statistics stats
) throws IOException {
checkOpen();
// Get block info from namenode
return new DFSInputStream(src, buffersize, verifyChecksum);
}-----------------------------------------------------------------------------------------------------------------------
结合上下文我们可以看出：
文件系统是通过内部包装的DFSInputStream来完成 真正的读取数过程
下面一起分析创建DFSInputStream对象的过程
a,初始化以下对象：
1，是否对读取的数据进行校验的标识，
this.verifyChecksum = verifyChecksum;
2，缓冲区大小
this.buffersize = buffersize;
3，要打开的源文件路径
this.src = src;
this.buffersize = buffersize;
3，要打开的源文件路径
this.src = src;
4，预读取文件大小(默认为10个块大小)
prefetchSize = conf.getLong("dfs.read.prefetch.size", prefetchSize);
b，从NameNode中获得将要被打开文件的元数据信息
1, 调用callGetBlockLocations()方法取得用户请求的file的元数据信息,数据保存在locatedBlocks 中
//非常关键的一步
LocatedBlocks newInfo = callGetBlockLocations(namenode, src, 0, prefetchSize);
prefetchSize = conf.getLong("dfs.read.prefetch.size", prefetchSize);
b，从NameNode中获得将要被打开文件的元数据信息
1, 调用callGetBlockLocations()方法取得用户请求的file的元数据信息,数据保存在locatedBlocks 中
//非常关键的一步
LocatedBlocks newInfo = callGetBlockLocations(namenode, src, 0, prefetchSize);
2,若无法定位某个文件的数据Block，则表示文件不存在，则抛出对应异常
if (newInfo == null) {
if (locatedBlocks != null) {
this.currentNode = null;
=========================================================================================================
DFSInputStream构造完毕后，就可以通过read()方法读取数据了
read()方法：
从当前位置开始读取length字节到buffetr的offset开始的空间中。
两种read方法比较： read(long,byte[],int,int)和 read(byte[],int,int)方法的底层实现是一样的，只不过前者是实现了随机读，后者是顺序读
下面以第2种方法来分析：
0,checkOpen();
1，如果 输出流已经被关闭，则会抛出I/O异常
if (closed) {
if (newInfo == null) {
throw new IOException("Cannot open filename " + src);
}
3,
使用定位到locatedBlocks来更新当前的locatedBlocks
if (locatedBlocks != null) {
Iterator<LocatedBlock> oldIter = locatedBlocks.getLocatedBlocks().iterator();
Iterator<LocatedBlock> newIter = newInfo.getLocatedBlocks().iterator();
while (oldIter.hasNext() && newIter.hasNext()) {
if (! oldIter.next().getBlock().equals(newIter.next().getBlock())) {
throw new IOException("Blocklist for " + src + " has changed!");
}
}
}
this.locatedBlocks = newInfo;
4,置空currentNode
this.locatedBlocks = newInfo;
this.currentNode = null;
=========================================================================================================
DFSInputStream构造完毕后，就可以通过read()方法读取数据了
/**
* 读取一个字节
*/
@Override
public synchronized int read() throws IOException {
int ret = read( oneByteBuf, 0, 1 );
return ( ret <= 0 ) ? -1 : (oneByteBuf[0] & 0xff);
}
read()方法：
从当前位置开始读取length字节到buffetr的offset开始的空间中。
两种read方法比较： read(long,byte[],int,int)和 read(byte[],int,int)方法的底层实现是一样的，只不过前者是实现了随机读，后者是顺序读
下面以第2种方法来分析：
0,checkOpen();
1，如果 输出流已经被关闭，则会抛出I/O异常
if (closed) {
throw new IOException("Stream closed");
}
2，把 读取数据失败的次数重置为0
failures = 0;
3， 判断要读取的位置 pos 是否超过了文件的范围( getFileLength()) ，如果超过了，直接返回-1，否则执行下面的过程，并将重试的次数设置为2，发生I/O异常后 会重试一次，重试次数为1时，会提示警告，重试次数为0时，抛出异常
if (pos < getFileLength()) {
2，把 读取数据失败的次数重置为0
failures = 0;
3， 判断要读取的位置 pos 是否超过了文件的范围( getFileLength()) ，如果超过了，直接返回-1，否则执行下面的过程，并将重试的次数设置为2，发生I/O异常后 会重试一次，重试次数为1时，会提示警告，重试次数为0时，抛出异常
if (pos < getFileLength()) {
int retries = 2;
while (retries > 0) {
try {。。。。
4，
如果
当前文件指针
不在当前数据块，则定位到当前文件指针所在的数据块
if (pos > blockEnd) {
//第一次读取时，blockEnd=-1,所以这步，一定会被执行
if (pos > blockEnd) {
//第一次读取时，blockEnd=-1,所以这步，一定会被执行
currentNode =
blockSeekTo(pos);
}
5，
计算出实际需要读取的数据的长度(如果要读取的长度+当前数据流的位置-1<Block的结束位置，则返回要读取的长度，
否则返回 Block的结束位置+1- 当前数据流的位置 )
int realLen = Math.min(len, (int) (blockEnd - pos + 1));
否则返回 Block的结束位置+1- 当前数据流的位置 )
int realLen = Math.min(len, (int) (blockEnd - pos + 1));
6，通过调用readBuffer()方法来在当前的数据Block中读取数据
int result = readBuffer(buf, off, realLen);
int result = readBuffer(buf, off, realLen);
7，
如果读取到了当前数据Block的末尾，则抛出对应的异常
if (result >= 0) {
//更新pos
if (result >= 0) {
//更新pos
pos += result;
} else {
// got a EOS from reader though we expect more data on it.
throw new IOException("Unexpected EOS from the reader");
}
8， 将读取到的字节数添加到文件系统的统计对象中，并返回实际读取到数据的长度
if (stats != null && result != -1) {
8， 将读取到的字节数添加到文件系统的统计对象中，并返回实际读取到数据的长度
if (stats != null && result != -1) {
stats.incrementBytesRead(result);
}
9, return result;
read方法总结：read方法在一次调用中并不会主动跨越数据块读取数据，DFSInputStream尽可能的len个字节的数据（当前数据块有多少数据就读多少数据，直到满足len的长度，若当前数据块不足len长度，也不会跳到下一个数据块，而是直接返回） 另外， BlockReader主要是用来接收某一个数据节点发送来的数据块的数据，它的实现很简单，有兴趣的话可以阅读它的源代码
=========================================================================================================
blockSeekTo()方法：
读取操作越过某个数据块边界时，用于建立到下一个数据块所在节点的联系，并初始化读取数据需要的BlockReader输入流对象
1,如果要读取的位置超过了文件的范围，则抛出对应的异常
if (target >= getFileLength()) {
read方法总结：read方法在一次调用中并不会主动跨越数据块读取数据，DFSInputStream尽可能的len个字节的数据（当前数据块有多少数据就读多少数据，直到满足len的长度，若当前数据块不足len长度，也不会跳到下一个数据块，而是直接返回） 另外， BlockReader主要是用来接收某一个数据节点发送来的数据块的数据，它的实现很简单，有兴趣的话可以阅读它的源代码
=========================================================================================================
blockSeekTo()方法：
读取操作越过某个数据块边界时，用于建立到下一个数据块所在节点的联系，并初始化读取数据需要的BlockReader输入流对象
1,如果要读取的位置超过了文件的范围，则抛出对应的异常
if (target >= getFileLength()) {
throw new IOException("Attempted to read past end of file");
}
2,如果blockReader不为null,则 blockReader.close();
if ( blockReader != null ) {
2,如果blockReader不为null,则 blockReader.close();
if ( blockReader != null ) {
//如果blockReader不为null,则关闭对当前block进行读取操作的blockReader
blockReader.close();
blockReader = null;
}
3,如果当前与DataNode的连接socket不为空，则 s.close();
if (s != null) {
LocatedBlock targetBlock = getBlockAt(target);
long offsetIntoBlock = target - targetBlock.getStartOffset();
DatanodeInfo chosenNode = null;
3,如果当前与DataNode的连接socket不为空，则 s.close();
if (s != null) {
//如果当前与DataNode的连接socket不为空，则关闭这个Socket
s.close();
s = null;
}
4,通过
getBlockAt(target)方法
获取文件指定位置的LocatedBlock对象（包含了数据块的位置信息）LocatedBlock targetBlock = getBlockAt(target);
assert (target==this.pos) : "Wrong postion " + pos + " expect " + target;
5,
取得要读取的target位置在Block中的偏移量
long offsetIntoBlock = target - targetBlock.getStartOffset();
DatanodeInfo chosenNode = null;
6,当与DataNode的连接关闭后，
循环执行
下面代码
while (s == null) { 。。。
while (s == null) { 。。。
7,通过
chooseDataNode(
targetBlock
)
取得包含目标Block的最优的DataNode
DNAddrPair retval = chooseDataNode(targetBlock);
chosenNode = retval.info;
DNAddrPair retval = chooseDataNode(targetBlock);
chosenNode = retval.info;
InetSocketAddress targetAddr = retval.addr;
8,
创建与远程DataNode的Socket连接
try {
s = socketFactory.createSocket();
try {
s = socketFactory.createSocket();
NetUtils.connect(s, targetAddr, socketTimeout);
s.setSoTimeout(socketTimeout);
Block blk = targetBlock.getBlock();
9, 使用BlockReader的newBlockReader()方法来创建读取Block数据的BlockReader
blockReader = BlockReader.newBlockReader(s, src, blk.getBlockId(),
addToDeadNodes(chosenNode);
9, 使用BlockReader的newBlockReader()方法来创建读取Block数据的BlockReader
blockReader = BlockReader.newBlockReader(s, src, blk.getBlockId(),
blk.getGenerationStamp(),
offsetIntoBlock, blk.getNumBytes() - offsetIntoBlock,
buffersize, verifyChecksum, clientName);
10,如果期间发生了I/O异常，
addToDeadNodes(chosenNode);
11,return chosenNode;
blockSeekTo()方法总结：
blockSeekTo用于返回一个DataNoDeInfo,首先它会关闭已经存在的BlockReader和socket,
然后获取LocaketBlock 对象，然后根据LocatedBlock对象通过ChooseDataNode方法找到对应的
DNAddrPai创建与远程DataNode的Socket连接，然后使用BlockReader的newBlockReader方法来创建读取Block数据的BlockReaderLocaketBlock 对象,然后返回
DataNoDeInfo
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
getBlockAt(target)方法：
private
LocatedBlock getBlockAt(long offset) throws IOException {
assert (locatedBlocks != null) : "locatedBlocks is null";
// 先从缓存里面去查找
int targetBlockIdx = locatedBlocks.findBlock(offset);
if (targetBlockIdx < 0) {
// 缓存里没有这Block
targetBlockIdx = LocatedBlocks.getInsertIndex(targetBlockIdx);
// 调用远程方法重新获取
LocatedBlocks
LocatedBlocks newBlocks;
newBlocks = callGetBlockLocations(namenode, src, offset, prefetchSize);
assert (newBlocks != null) : "Could not find target position " + offset;
locatedBlocks.insertRange(targetBlockIdx, newBlocks.getLocatedBlocks());
}
//根据 targetBlockIdx 获取 LocatedBlock 对象
//根据 targetBlockIdx 获取 LocatedBlock 对象
LocatedBlock blk = locatedBlocks.get(targetBlockIdx);
// 更新当前位置
this.pos = offset;
this.blockEnd = blk.getStartOffset() + blk.getBlockSize() - 1;
this.currentBlock = blk.getBlock();
return blk;
}
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
chooseDataNode( targetBlock )方法：
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
=========================================================================================================
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
chooseDataNode( targetBlock )方法：
/**
* 从包含指定Block的DataNode集合中选取最优的一个DataNode来进行读取操作
* @param block
* @return
* @throws IOException
*/
private DNAddrPair chooseDataNode(LocatedBlock block)
throws IOException {
while (true) {
//获取 nodes 信息，事先已经按照网络顺序排好序
//获取 nodes 信息，事先已经按照网络顺序排好序
DatanodeInfo[] nodes = block.getLocations();
try {
DatanodeInfo chosenNode = bestNode(nodes, deadNodes);
//获取DataNode地址
InetSocketAddress targetAddr =
NetUtils.createSocketAddr(chosenNode.getName());
//返回一个最优DataNode
return new DNAddrPair(chosenNode, targetAddr);
} catch (IOException ie) {
String blockInfo = block.getBlock() + " file=" + src;
if (failures >= maxBlockAcquireFailures) {
throw new IOException("Could not obtain block: " + blockInfo);
}
if (nodes == null || nodes.length == 0) {
LOG.info("No node available for block: " + blockInfo);
}
LOG.info("Could not obtain block " + block.getBlock() + " from any node: " + ie);
try {
Thread.sleep(3000);
} catch (InterruptedException iex) {
}
deadNodes.clear(); //2nd option is to remove only nodes[blockId]
//清空死亡节点后，重新从NameNode获取数据的元信息，并增加失败次数
//清空死亡节点后，重新从NameNode获取数据的元信息，并增加失败次数
openInfo();
failures++;
continue;
}
}
}
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------
bestNode(nodes, deadNodes)方法：
/**
/**
* 输入的DataNode集合nodes已经是按照DataNode的优先级排好序的，
* bestNodes方法会从nodes中选择出第一个不在deadNodes集合中的DataNode作为最优
* @param nodes
* @param deadNodes
* @return
* @throws IOException
*/
private
DatanodeInfo bestNode(DatanodeInfo nodes[],
AbstractMap<DatanodeInfo, DatanodeInfo> deadNodes)
throws IOException {
if (nodes != null) {
for (int i = 0; i < nodes.length; i++) {
if (!deadNodes.containsKey(nodes[i])) {
return nodes[i];
}
}
}
throw new IOException("No live nodes contain current block");
}
=========================================================================================================
readBuffer() 方法：执行读取数据的具体逻辑
//重试DataNode的标识 设置为true
boolean retryCurrentNode = true;
while (true) {
try {
return blockReader.read(buf, off, len);
} catch ( ChecksumException ce ) {
// 如果读取数据期间发生了异常且该异常是由于校验和失败，则需要向NameNode报告已经出错的Block，并将重试DataNode的标 识设置为false，因为校验和失败意味着此数据块不可用
// 如果读取数据期间发生了异常且该异常是由于校验和失败，则需要向NameNode报告已经出错的Block，并将重试DataNode的标 识设置为false，因为校验和失败意味着此数据块不可用
LOG.warn("Found Checksum error for " + currentBlock + " from " +
currentNode.getName() + " at " + ce.getPos());
reportChecksumFailure(src, currentBlock, currentNode);
ioe = ce;
retryCurrentNode = false;
} catch ( IOException e ) {
// 如果该异常是I/O异常，且 重试DataNode的标识为false,则将资源被找到的标识 设置为false，并将当前DataNode加入到已死亡Da taNode集合中，然后尝试从其他存活的的DataNode集合中挑选最优的DataNode并建立Socket连接，并读取数据，如果仍然未找 到资源，则抛出异常
// 如果该异常是I/O异常，且 重试DataNode的标识为false,则将资源被找到的标识 设置为false，并将当前DataNode加入到已死亡Da taNode集合中，然后尝试从其他存活的的DataNode集合中挑选最优的DataNode并建立Socket连接，并读取数据，如果仍然未找 到资源，则抛出异常
如果重试DataNode的标识为true,则重新定位到目标文件所在的最优DataNode，然后建立起Socket连接，并开始读取数据，如果仍 然未找到资源，则抛出异常
if (!retryCurrentNode) {
LOG.warn("Exception while reading from " + currentBlock +
" of " + src + " from " + currentNode + ": " +
StringUtils.stringifyException(e));
}
ioe = e;
}
boolean sourceFound = false;
if (retryCurrentNode) {
/* possibly retry the same node so that transient errors don't
* result in application level failures (e.g. Datanode could have
* closed the connection because the client is idle for too long).
*/
sourceFound = seekToBlockSource(pos);
} else {
addToDeadNodes(currentNode);//加入死亡名单
sourceFound = seekToNewSource(pos);
}
if (!sourceFound) {
throw ioe;
}
retryCurrentNode = false;
}
readBuffer方法小结：
委托blockReader读取数据，处理读取数据期间发生的异常
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
reportChecksumFailure( src, currentBlock, currentNode）方法：向NameNode报告校验和失败的数据块
seekToBlockSource(pos)方法：
private synchronized boolean seekToBlockSource(long targetPos)
throws IOException {
currentNode = blockSeekTo(targetPos);
return true;
}
addToDeadNodes(currentNode) 方法：
addToDeadNodes(currentNode) 方法：
void addToDeadNodes(DatanodeInfo dnInfo) {
deadNodes.put(dnInfo, dnInfo);
}
seekToNewSource(long targetPos)方法：
真正读取数据的方法：
seekToNewSource(long targetPos)方法：
public synchronized boolean seekToNewSource(long targetPos) throws IOException {
boolean markedDead = deadNodes.containsKey(currentNode);
addToDeadNodes(currentNode);
DatanodeInfo oldNode = currentNode;
DatanodeInfo newNode =
blockSeekTo(targetPos);
if (!markedDead) {
/* remove it from deadNodes. blockSeekTo could have cleared
* deadNodes and added currentNode again. Thats ok. */
deadNodes.remove(oldNode);
}
if (!oldNode.getStorageID().equals(newNode.getStorageID())) {
currentNode = newNode;
return true;
} else {
return false;
}
}
----------------------------------------------------------------------------------------------------------------------------------------------------------------------
真正读取数据的方法：
/**
* 从数据节点流接口上读入一份校验块数据
*/
@Override
protected synchronized int readChunk(long pos, byte[] buf, int offset,
int len, byte[] checksumBuf)
throws IOException {
/**
* 如果已经达到数据Block的末尾，则返回-1
*/
if ( gotEOS ) {
if ( startOffset < 0 ) {
//This is mainly for debugging. can be removed.
throw new IOException( "BlockRead: already got EOS or an error" );
}
startOffset = -1;
return -1;
}
//取得下一个要被读取的Chunk的偏移量
long chunkOffset = lastChunkOffset;
if ( lastChunkLen > 0 ) {
chunkOffset += lastChunkLen;
}
//判断要被读取的Chunk的偏移量是否有效，如果无效，则会抛出对应的异常
if ( (pos + firstChunkOffset) != chunkOffset ) {
throw new IOException("Mismatch in pos : " + pos + " + " +
firstChunkOffset + " != " + chunkOffset);
}
//如果前一个Packet中的数据已经被读取完毕了，则读取下一个Packet
if (dataLeft <= 0) {
//读取Packet的头信息
int packetLen = in.readInt();//包长度
long offsetInBlock = in.readLong();//偏移量
long seqno = in.readLong();//版本号
boolean lastPacketInBlock = in.readBoolean();
if (LOG.isDebugEnabled()) {
LOG.debug("DFSClient readChunk got seqno " + seqno +
" offsetInBlock " + offsetInBlock +
" lastPacketInBlock " + lastPacketInBlock +
" packetLen " + packetLen);
}
//从输入流获得要读取的数据的长度
int dataLen = in.readInt();
//检查要被读取数据的长度的有效性，dataLen必须大于0，而且除了Block中的最后一个Packet之外，其它的Packet中的数据长度必须为一个校验和对应字节的整数倍
if ( dataLen < 0 ||
( (dataLen % bytesPerChecksum) != 0 && !lastPacketInBlock ) ||
(seqno != (lastSeqNo + 1)) ) {
throw new IOException("BlockReader: error in packet header" +
"(chunkOffset : " + chunkOffset +
", dataLen : " + dataLen +
", seqno : " + seqno +
" (last: " + lastSeqNo + "))");
}
lastSeqNo = seqno;
isLastPacket = lastPacketInBlock;
dataLeft = dataLen;
//调用adjustChecksumBytes方法来根据要读取的数据的长度调整用于保存校验和的缓冲区的长度
adjustChecksumBytes(dataLen);
//调用IOUtils.readFully方法来从Socket的输入流中读取校验和信息
if (dataLen > 0) {
IOUtils.readFully(in, checksumBytes.array(), 0,
checksumBytes.limit());
}
}
//计算要被读取的chunk的长度
int chunkLen = Math.min(dataLeft, bytesPerChecksum);
//从Socket输入流中读取原生的数据信息
if ( chunkLen > 0 ) {
// len should be >= chunkLen
IOUtils.readFully(in, buf, offset, chunkLen);
checksumBytes.get(checksumBuf, 0, checksumSize);
}
dataLeft -= chunkLen;
lastChunkOffset = chunkOffset;
lastChunkLen = chunkLen;
//判断是否到达Block的末尾，如果已到末尾则标记为true
if ((dataLeft == 0 && isLastPacket) || chunkLen == 0) {
gotEOS = true;
}
//如果被读取的chunk的长度为0，则直接返回-1，否则，返回读取到的chunk的长度
if ( chunkLen == 0 ) {
return -1;
}
return chunkLen;
}
=========================================================================================================
随机读取的read方法：
/**
* 从position位置开始读取length字节到buffetr的offset开始的空间中
*/
@Override
public int read(long position, byte[] buffer, int offset, int length)
throws IOException {
// 检查客户端是否正在运行
checkOpen();
//如果DFSInputStream输入流被关闭了，则抛出对应的异常
if (closed) {
throw new IOException("Stream closed");
}
//如果要进行读取的位置不在文件中，则返回-1
failures = 0;
long filelen = getFileLength();
if ((position < 0) || (position >= filelen)) {
return -1;
}
//如果要读取的数据的长度比position到文件末尾的长度大
//则截取position到文件末尾的长度作为实际读取的长度
int realLen = length;
if ((position + length) > filelen) {
realLen = (int)(filelen - position);
}
//取得要读取的数据所对应的Block范围
List<LocatedBlock> blockRange =
getBlockRange(position, realLen);
int remaining = realLen;
//调用fetchBlockByteRange方法来从某一段数据Block中读取一段数据
for (LocatedBlock blk : blockRange) {
long targetStart = position - blk.getStartOffset();
long bytesToRead = Math.min(remaining, blk.getBlockSize() - targetStart);
fetchBlockByteRange(blk, targetStart,
targetStart + bytesToRead - 1, buffer, offset);
remaining -= bytesToRead;
position += bytesToRead;
offset += bytesToRead;
}
assert remaining == 0 : "Wrong number of bytes read.";
//将读取到的字节数添加到文件系统的统计对象中
if (stats != null) {
stats.incrementBytesRead(realLen);
}
return realLen;
}
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
getBlockRange方法：
private synchronized List<LocatedBlock> getBlockRange(long offset,
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
getBlockRange方法：
private synchronized List<LocatedBlock> getBlockRange(long offset,
long length)
throws IOException {
assert (locatedBlocks != null) : "locatedBlocks is null";
List<LocatedBlock> blockRange = new ArrayList<LocatedBlock>();
// search cached blocks first
int blockIdx = locatedBlocks.findBlock(offset);
if (blockIdx < 0) {
// block is not cached
blockIdx = LocatedBlocks.getInsertIndex(blockIdx);
}
long remaining = length;
long curOff = offset;
while(remaining > 0) {
LocatedBlock blk = null;
if(blockIdx < locatedBlocks.locatedBlockCount())
blk = locatedBlocks.get(blockIdx);
if (blk == null || curOff < blk.getStartOffset()) {
LocatedBlocks newBlocks;
newBlocks = callGetBlockLocations(namenode, src, curOff, remaining);
locatedBlocks.insertRange(blockIdx, newBlocks.getLocatedBlocks());
continue;
}
assert curOff >= blk.getStartOffset() : "Block not found";
blockRange.add(blk);
long bytesRead = blk.getStartOffset() + blk.getBlockSize() - curOff;
remaining -= bytesRead;
curOff += bytesRead;
blockIdx++;
}
return blockRange;
}
---------------------------------------------------------------------------------------------------------------------------------------------------------------
简单的描述
read(byte buf[], int off, int len)方法：
pos > blockEnd
-----------------------------------> blockSeekTo(pos)-------------------> readBuffer(buf, off, realLen);
DatanodeInfo currentNode int result
blockSeekTo(long target)方法：LocatedBlock targetBlock = getBlockAt(target);
s==null
-------------------------------------->getBlockAt(target)-------------------->chooseDataNode(targetBlock)----------------------> BlockReader.newBlockReader
LocatedBlock targetBlock DNAddrPair blockReader
readBuffer(buf, off, realLen)方法：
——----------------- 一切顺利 --------------------------- blockReader.read(buf, off, len)
---------------------------------------------------------------------------------------------------------------------------------------------------------------
fetchBlockByteRange方法：
private void fetchBlockByteRange(LocatedBlock block, long start,
private void fetchBlockByteRange(LocatedBlock block, long start,
long end, byte[] buf, int offset) throws IOException {
//
// Connect to best DataNode for desired Block, with potential offset
//
Socket dn = null;
int numAttempts = block.getLocations().length;
IOException ioe = null;
while (dn == null && numAttempts-- > 0 ) {
DNAddrPair retval = chooseDataNode(block);
DatanodeInfo chosenNode = retval.info;
InetSocketAddress targetAddr = retval.addr;
BlockReader reader = null;
try {
dn = socketFactory.createSocket();
NetUtils.connect(dn, targetAddr, socketTimeout);
dn.setSoTimeout(socketTimeout);
int len = (int) (end - start + 1);
reader = BlockReader.newBlockReader(dn, src,
block.getBlock().getBlockId(),
block.getBlock().getGenerationStamp(),
start, len, buffersize,
verifyChecksum, clientName);
int nread = reader.readAll(buf, offset, len);
if (nread != len) {
throw new IOException("truncated return from reader.read(): " +
"excpected " + len + ", got " + nread);
}
return;
} catch (ChecksumException e) {
ioe = e;
LOG.warn("fetchBlockByteRange(). Got a checksum exception for " +
src + " at " + block.getBlock() + ":" +
e.getPos() + " from " + chosenNode.getName());
reportChecksumFailure(src, block.getBlock(), chosenNode);
} catch (IOException e) {
ioe = e;
LOG.warn("Failed to connect to " + targetAddr +
" for file " + src +
" for block " + block.getBlock().getBlockId() + ":" +
StringUtils.stringifyException(e));
} finally {
IOUtils.closeStream(reader);
IOUtils.closeSocket(dn);
dn = null;
}
// Put chosen node into dead list, continue
addToDeadNodes(chosenNode);
}
throw (ioe == null) ? new IOException("Could not read data") : ioe;
}
========================================================================================================
简单的描述
read(byte buf[], int off, int len)方法：
pos > blockEnd
-----------------------------------> blockSeekTo(pos)-------------------> readBuffer(buf, off, realLen);
DatanodeInfo currentNode int result
blockSeekTo(long target)方法：LocatedBlock targetBlock = getBlockAt(target);
s==null
-------------------------------------->getBlockAt(target)-------------------->chooseDataNode(targetBlock)----------------------> BlockReader.newBlockReader
LocatedBlock targetBlock DNAddrPair blockReader
readBuffer(buf, off, realLen)方法：
——----------------- 一切顺利 --------------------------- blockReader.read(buf, off, len)
|
ChecksumException | Exception& & retry==true
-----------------------------> reportChecksumFailure()--------------------------------------> seekToBlockSource(pos)
| boolean sourceFound
|
|------------ Exception& & retry==true --> addToDeadNodes(currentNode)--------------------> seekToNewSource(pos);
sourceFound
ChecksumException | Exception& & retry==true
-----------------------------> reportChecksumFailure()--------------------------------------> seekToBlockSource(pos)
| boolean sourceFound
|
|------------ Exception& & retry==true --> addToDeadNodes(currentNode)--------------------> seekToNewSource(pos);
sourceFound
最后
以上就是瘦瘦柜子为你收集整理的DFSClient技术内幕（读取数据）的全部内容，希望文章能够帮你解决DFSClient技术内幕（读取数据）所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
发表评论 取消回复