HDFS读写过程解析

古城微笑少年丶 2022-05-26 09:53 210阅读 0赞

**一、文件的打开**

**1.1、客户端**

HDFS打开一个文件，需要在客户端调用DistributedFileSystem.open(Path f, int bufferSize)，其实现为：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>public FSDataInputStream open(Path f, int bufferSize) throws IOException { </p><p>&nbsp; return new DFSClient.DFSDataInputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics));</p><p>}</p></td> 
  </tr> 
 </tbody> 
</table>

其中dfs为DistributedFileSystem的成员变量DFSClient，其open函数被调用，其中创建一个DFSInputStream(src, buffersize, verifyChecksum)并返回。

在DFSInputStream的构造函数中，openInfo函数被调用，其主要从namenode中得到要打开的文件所对应的blocks的信息，实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>synchronized void openInfo() throws IOException { </p><p>&nbsp; LocatedBlocks newInfo = callGetBlockLocations(namenode, src, 0, prefetchSize);</p><p>&nbsp; this.locatedBlocks = newInfo;</p><p>&nbsp; this.currentNode = null;</p><p>}</p></td> 
  </tr> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>private static LocatedBlocks callGetBlockLocations(ClientProtocol namenode,</p><p>&nbsp;&nbsp;&nbsp; String src, long start, long length) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; return namenode.getBlockLocations(src, start, length);</p><p>&nbsp;</p><p>}</p></td> 
  </tr> 
 </tbody> 
</table>

LocatedBlocks主要包含一个链表的List<LocatedBlock> blocks，其中每个LocatedBlock包含如下信息：

· Block b：此block的信息

· long offset：此block在文件中的偏移量

· DatanodeInfo\[\] locs：此block位于哪些DataNode上

上面namenode.getBlockLocations是一个RPC调用，最终调用NameNode类的getBlockLocations函数。

**1.2、NameNode**

NameNode.getBlockLocations实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>public LocatedBlocks&nbsp;&nbsp; getBlockLocations(String src, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long offset, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long length) throws IOException { </p><p>&nbsp; return namesystem.getBlockLocations(getClientMachine(), </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; src, offset, length);</p><p>}</p></td> 
  </tr> 
 </tbody> 
</table>

namesystem是NameNode一个成员变量，其类型为FSNamesystem，保存的是NameNode的name space树，其中一个重要的成员变量为FSDirectory dir。

FSDirectory和Lucene中的FSDirectory没有任何关系，其主要包括FSImage fsImage，用于读写硬盘上的fsimage文件，FSImage类有成员变量FSEditLog editLog，用于读写硬盘上的edit文件，这两个文件的关系在上一篇文章中已经解释过。

FSDirectory还有一个重要的成员变量INodeDirectoryWithQuota rootDir，INodeDirectoryWithQuota的父类为INodeDirectory，实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>public class INodeDirectory extends INode { </p><p>&nbsp; ……</p><p>&nbsp; private List&lt;INode&gt; children;</p><p>&nbsp; ……</p><p>｝</p></td> 
  </tr> 
 </tbody> 
</table>

由此可见INodeDirectory本身是一个INode，其中包含一个链表的INode，此链表中，如果仍为文件夹，则是类型INodeDirectory，如果是文件，则是类型INodeFile，INodeFile中有成员变量BlockInfo blocks\[\]，是此文件包含的block的信息。显然这是一棵树形的结构。

FSNamesystem.getBlockLocations函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>public LocatedBlocks getBlockLocations(String src, long offset, long length,</p><p>&nbsp;&nbsp;&nbsp; boolean doAccessTime) throws IOException { </p><p>&nbsp; final LocatedBlocks ret = getBlockLocationsInternal(src, dir.getFileINode(src),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; offset, length, Integer.MAX_VALUE, doAccessTime);&nbsp; </p><p>&nbsp; return ret;</p><p>}</p></td> 
  </tr> 
 </tbody> 
</table>

dir.getFileINode(src)通过路径名从文件系统树中找到INodeFile，其中保存的是要打开的文件的INode的信息。

getBlockLocationsInternal的实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>private synchronized LocatedBlocks getBlockLocationsInternal(String src,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; INodeFile inode,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long offset, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long length,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int nrBlocksToReturn,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean doAccessTime) </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throws IOException { </p><p>&nbsp; <strong>//<span style="font-family:'宋体';">得到此文件的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">信息</span></strong></p><p>&nbsp; Block[] blocks = inode.getBlocks();</p><p>&nbsp; List&lt;LocatedBlock&gt; results = new ArrayList&lt;LocatedBlock&gt;(blocks.length); </p><p>&nbsp; <strong>//<span style="font-family:'宋体';">计算从</span><span style="font-family:Verdana;">offset</span><span style="font-family:'宋体';">开始，长度为</span><span style="font-family:Verdana;">length</span><span style="font-family:'宋体';">所涉及的</span><span style="font-family:Verdana;">blocks</span></strong></p><p>&nbsp; int curBlk = 0;</p><p>&nbsp; long curPos = 0, blkSize = 0;</p><p>&nbsp; int nrBlocks = (blocks[0].getNumBytes() == 0) ? 0 : blocks.length;</p><p>&nbsp; for (curBlk = 0; curBlk &lt; nrBlocks; curBlk++) { </p><p>&nbsp;&nbsp;&nbsp; blkSize = blocks[curBlk].getNumBytes();</p><p>&nbsp;&nbsp;&nbsp; if (curPos + blkSize &gt; offset) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">当</span><span style="font-family:Verdana;">offset</span><span style="font-family:'宋体';">在</span><span style="font-family:Verdana;">curPos</span><span style="font-family:'宋体';">和</span><span style="font-family:Verdana;">curPos + blkSize</span><span style="font-family:'宋体';">之间的时候，</span><span style="font-family:Verdana;">curBlk</span><span style="font-family:'宋体';">指向</span><span style="font-family:Verdana;">offset</span><span style="font-family:'宋体';">所在的</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; break;</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; curPos += blkSize;</p><p>&nbsp; }</p><p>&nbsp; long endOff = offset + length;</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">循环，依次遍历从</span><span style="font-family:Verdana;">curBlk</span><span style="font-family:'宋体';">开始的每个</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，直到当前位置</span><span style="font-family:Verdana;">curPos</span><span style="font-family:'宋体';">越过</span><span style="font-family:Verdana;">endOff</span></strong></p><p>&nbsp; do { </p><p>&nbsp;&nbsp;&nbsp; int numNodes = blocksMap.numNodes(blocks[curBlk]);</p><p>&nbsp;&nbsp;&nbsp; int numCorruptNodes = countNodes(blocks[curBlk]).corruptReplicas();</p><p>&nbsp;&nbsp;&nbsp; int numCorruptReplicas = corruptReplicas.numCorruptReplicas(blocks[curBlk]); </p><p>&nbsp;&nbsp;&nbsp; boolean blockCorrupt = (numCorruptNodes == numNodes);</p><p>&nbsp;&nbsp;&nbsp; int numMachineSet = blockCorrupt ? numNodes : </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (numNodes - numCorruptNodes);</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">依次找到此</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">所对应的</span><span style="font-family:Verdana;">datanode</span><span style="font-family:'宋体';">，将其中没有损坏的放入</span><span style="font-family:Verdana;">machineSet</span><span style="font-family:'宋体';">中</span></strong></p><p>&nbsp;&nbsp;&nbsp; DatanodeDescriptor[] machineSet = new DatanodeDescriptor[numMachineSet];</p><p>&nbsp;&nbsp;&nbsp; if (numMachineSet &gt; 0) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; numNodes = 0;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for(Iterator&lt;DatanodeDescriptor&gt; it = </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blocksMap.nodeIterator(blocks[curBlk]); it.hasNext();) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DatanodeDescriptor dn = it.next();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean replicaCorrupt = corruptReplicas.isReplicaCorrupt(blocks[curBlk], dn);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (blockCorrupt || (!blockCorrupt &amp;&amp; !replicaCorrupt))</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; machineSet[numNodes++] = dn;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">使用此</span><span style="font-family:Verdana;">machineSet</span><span style="font-family:'宋体';">和当前的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">构造一个</span><span style="font-family:Verdana;">LocatedBlock</span></strong></p><p>&nbsp;&nbsp;&nbsp; results.add(new LocatedBlock(blocks[curBlk], machineSet, curPos,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockCorrupt));</p><p>&nbsp;&nbsp;&nbsp; curPos += blocks[curBlk].getNumBytes();</p><p>&nbsp;&nbsp;&nbsp; curBlk++;</p><p>&nbsp; } while (curPos &lt; endOff </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &amp;&amp; curBlk &lt; blocks.length </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &amp;&amp; results.size() &lt; nrBlocksToReturn);</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">使用此</span><span style="font-family:Verdana;">LocatedBlock</span><span style="font-family:'宋体';">链表构造一个</span><span style="font-family:Verdana;">LocatedBlocks</span><span style="font-family:'宋体';">对象返回</span></strong></p><p>&nbsp; return inode.createLocatedBlocks(results);</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

**1.3、客户端**

通过RPC调用，在NameNode得到的LocatedBlocks对象，作为成员变量构造DFSInputStream对象，最后包装为FSDataInputStream返回给用户。

**二、文件的读取**

**2.1、客户端**

文件读取的时候，客户端利用文件打开的时候得到的FSDataInputStream.read(long position, byte\[\] buffer, int offset, int length)函数进行文件读操作。

FSDataInputStream会调用其封装的DFSInputStream的read(long position, byte\[\] buffer, int offset, int length)函数，实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>public int read(long position, byte[] buffer, int offset, int length)</p><p>&nbsp; throws IOException { </p><p>&nbsp; long filelen = getFileLength();</p><p>&nbsp; int realLen = length;</p><p>&nbsp; if ((position + length) &gt; filelen) { </p><p>&nbsp;&nbsp;&nbsp; realLen = (int)(filelen - position);</p><p>&nbsp; }</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">首先得到包含从</span><span style="font-family:Verdana;">offset</span><span style="font-family:'宋体';">到</span><span style="font-family:Verdana;">offset + length</span><span style="font-family:'宋体';">内容的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">列表</span></strong></p><p>&nbsp; <strong>//<span style="font-family:'宋体';">比如对于</span><span style="font-family:Verdana;">64M</span><span style="font-family:'宋体';">一个</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">的文件系统来说，欲读取从</span><span style="font-family:Verdana;">100M</span><span style="font-family:'宋体';">开始，长度为</span><span style="font-family:Verdana;">128M</span><span style="font-family:'宋体';">的数据，则</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">列表包括第</span><span style="font-family:Verdana;">2</span><span style="font-family:'宋体';">，</span><span style="font-family:Verdana;">3</span><span style="font-family:'宋体';">，</span><span style="font-family:Verdana;">4</span><span style="font-family:'宋体';">块</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp; List&lt;LocatedBlock&gt; blockRange = getBlockRange(position, realLen);</p><p>&nbsp; int remaining = realLen;</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">对每一个</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，从中读取内容</span></strong></p><p>&nbsp; <strong>//<span style="font-family:'宋体';">对于上面的例子，对于第</span><span style="font-family:Verdana;">2</span><span style="font-family:'宋体';">块</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，读取从</span><span style="font-family:Verdana;">36M</span><span style="font-family:'宋体';">开始，读取长度</span><span style="font-family:Verdana;">28M</span><span style="font-family:'宋体';">，对于第</span><span style="font-family:Verdana;">3</span><span style="font-family:'宋体';">块，读取整一块</span><span style="font-family:Verdana;">64M</span><span style="font-family:'宋体';">，对于第</span><span style="font-family:Verdana;">4</span><span style="font-family:'宋体';">块，读取从</span><span style="font-family:Verdana;">0</span><span style="font-family:'宋体';">开始，长度为</span><span style="font-family:Verdana;">36M</span><span style="font-family:'宋体';">，共</span><span style="font-family:Verdana;">128M</span><span style="font-family:'宋体';">数据</span></strong></p><p>&nbsp; for (LocatedBlock blk : blockRange) { </p><p>&nbsp;&nbsp;&nbsp; long targetStart = position - blk.getStartOffset();</p><p>&nbsp;&nbsp;&nbsp; long bytesToRead = Math.min(remaining, blk.getBlockSize() - targetStart);</p><p>&nbsp;&nbsp;&nbsp; fetchBlockByteRange(blk, targetStart, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; targetStart + bytesToRead - 1, buffer, offset);</p><p>&nbsp;&nbsp;&nbsp; remaining -= bytesToRead;</p><p>&nbsp;&nbsp;&nbsp; position += bytesToRead;</p><p>&nbsp;&nbsp;&nbsp; offset += bytesToRead;</p><p>&nbsp; }</p><p>&nbsp; assert remaining == 0 : "Wrong number of bytes read.";</p><p>&nbsp; if (stats != null) { </p><p>&nbsp;&nbsp;&nbsp; stats.incrementBytesRead(realLen);</p><p>&nbsp; }</p><p>&nbsp; return realLen;</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

其中getBlockRange函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>private synchronized List&lt;LocatedBlock&gt; getBlockRange(long offset, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long length) </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throws IOException { </p><p>&nbsp; List&lt;LocatedBlock&gt; blockRange = new ArrayList&lt;LocatedBlock&gt;();</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">首先从缓存的</span><span style="font-family:Verdana;">locatedBlocks</span><span style="font-family:'宋体';">中查找</span><span style="font-family:Verdana;">offset</span><span style="font-family:'宋体';">所在的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">在缓存链表中的位置</span></strong></p><p>&nbsp; int blockIdx = locatedBlocks.findBlock(offset);</p><p>&nbsp; if (blockIdx &lt; 0) { // block is not cached</p><p>&nbsp;&nbsp;&nbsp; blockIdx = LocatedBlocks.getInsertIndex(blockIdx);</p><p>&nbsp; }</p><p>&nbsp; long remaining = length;</p><p>&nbsp; long curOff = offset;</p><p>&nbsp; while(remaining &gt; 0) { </p><p>&nbsp;&nbsp;&nbsp; LocatedBlock blk = null;</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">按照</span><span style="font-family:Verdana;">blockIdx</span><span style="font-family:'宋体';">的位置找到</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp; if(blockIdx &lt; locatedBlocks.locatedBlockCount())</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blk = locatedBlocks.get(blockIdx);</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">为空，则缓存中没有此</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，则直接从</span><span style="font-family:Verdana;">NameNode</span><span style="font-family:'宋体';">中查找这些</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，并加入缓存</span></strong></p><p>&nbsp;&nbsp;&nbsp; if (blk == null || curOff &lt; blk.getStartOffset()) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; LocatedBlocks newBlocks;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; newBlocks = callGetBlockLocations(namenode, src, curOff, remaining);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; locatedBlocks.insertRange(blockIdx, newBlocks.getLocatedBlocks());</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; continue;</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">找到，则放入结果集</span></strong></p><p>&nbsp;&nbsp;&nbsp; blockRange.add(blk);</p><p>&nbsp;&nbsp;&nbsp; long bytesRead = blk.getStartOffset() + blk.getBlockSize() - curOff;</p><p>&nbsp;&nbsp;&nbsp; remaining -= bytesRead;</p><p>&nbsp;&nbsp;&nbsp; curOff += bytesRead;</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">取下一个</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp; blockIdx++;</p><p>&nbsp; }</p><p>&nbsp; return blockRange;</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

其中fetchBlockByteRange实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>private void fetchBlockByteRange(LocatedBlock block, long start,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long end, byte[] buf, int offset) throws IOException { </p><p>&nbsp; Socket dn = null;</p><p>&nbsp; int numAttempts = block.getLocations().length;</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">此</span><span style="font-family:Verdana;">while</span><span style="font-family:'宋体';">循环为读取失败后的重试次数</span></strong></p><p>&nbsp; while (dn == null &amp;&amp; numAttempts-- &gt; 0 ) { </p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">选择一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">来读取数据</span></strong></p><p>&nbsp;&nbsp;&nbsp; DNAddrPair retval = chooseDataNode(block);</p><p>&nbsp;&nbsp;&nbsp; DatanodeInfo chosenNode = retval.info;</p><p>&nbsp;&nbsp;&nbsp; InetSocketAddress targetAddr = retval.addr;</p><p>&nbsp;&nbsp;&nbsp; BlockReader reader = null;</p><p>&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建</span><span style="font-family:Verdana;">Socket</span><span style="font-family:'宋体';">连接到</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dn = socketFactory.createSocket();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dn.connect(targetAddr, socketTimeout);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dn.setSoTimeout(socketTimeout);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int len = (int) (end - start + 1);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">利用建立的</span><span style="font-family:Verdana;">Socket</span><span style="font-family:'宋体';">链接，生成一个</span><span style="font-family:Verdana;">reader</span><span style="font-family:'宋体';">负责从</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">读取数据</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; reader = BlockReader.newBlockReader(dn, src, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; block.getBlock().getBlockId(),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; block.getBlock().getGenerationStamp(),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; start, len, buffersize, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; verifyChecksum, clientName);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">读取数据</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int nread = reader.readAll(buf, offset, len);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return;</p><p>&nbsp;&nbsp;&nbsp; } finally { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeStream(reader);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeSocket(dn);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dn = null;</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果读取失败，则将此</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">标记为失败节点</span></strong></p><p>&nbsp;&nbsp;&nbsp; addToDeadNodes(chosenNode);</p><p>&nbsp; }</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

BlockReader.newBlockReader函数实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>public static BlockReader newBlockReader( Socket sock, String file,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long blockId, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long genStamp,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long startOffset, long len,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int bufferSize, boolean verifyChecksum,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String clientName)</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throws IOException { </p><p>&nbsp; <strong>//<span style="font-family:'宋体';">使用</span><span style="font-family:Verdana;">Socket</span><span style="font-family:'宋体';">建立写入流，向</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">发送读指令</span></strong></p><p>&nbsp; DataOutputStream out = new DataOutputStream(</p><p>&nbsp;&nbsp;&nbsp; new BufferedOutputStream(NetUtils.getOutputStream(sock,HdfsConstants.WRITE_TIMEOUT))); </p><p>&nbsp; out.writeShort( DataTransferProtocol.DATA_TRANSFER_VERSION );</p><p>&nbsp; out.write( DataTransferProtocol.OP_READ_BLOCK );</p><p>&nbsp; out.writeLong( blockId );</p><p>&nbsp; out.writeLong( genStamp );</p><p>&nbsp; out.writeLong( startOffset );</p><p>&nbsp; out.writeLong( len );</p><p>&nbsp; Text.writeString(out, clientName);</p><p>&nbsp; out.flush();</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">使用</span><span style="font-family:Verdana;">Socket</span><span style="font-family:'宋体';">建立读入流，用于从</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">读取数据 </span></strong></p><p>&nbsp; DataInputStream in = new DataInputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new BufferedInputStream(NetUtils.getInputStream(sock), </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bufferSize));</p><p>&nbsp; DataChecksum checksum = DataChecksum.newDataChecksum( in );</p><p>&nbsp; long firstChunkOffset = in.readLong();</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">生成一个</span><span style="font-family:Verdana;">reader</span><span style="font-family:'宋体';">，主要包含读入流，用于读取数据 </span></strong></p><p>&nbsp; return new BlockReader( file, blockId, in, checksum, verifyChecksum,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; startOffset, firstChunkOffset, sock );</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

BlockReader的readAll函数就是用上面生成的DataInputStream读取数据。

**2.2、DataNode**

在DataNode启动的时候，会调用函数startDataNode，其中与数据读取有关的逻辑如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>void startDataNode(Configuration conf, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; AbstractList&lt;File&gt; dataDirs</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) throws IOException { </p><p>&nbsp; ……</p><p>&nbsp; <strong>// <span style="font-family:'宋体';">建立一个</span><span style="font-family:Verdana;">ServerSocket</span><span style="font-family:'宋体';">，并生成一个</span><span style="font-family:Verdana;">DataXceiverServer</span><span style="font-family:'宋体';">来监控客户端的链接</span></strong></p><p>&nbsp; ServerSocket ss = (socketWriteTimeout &gt; 0) ? </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ServerSocketChannel.open().socket() : new ServerSocket();</p><p>&nbsp; Server.bind(ss, socAddr, 0);</p><p>&nbsp; ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE); </p><p>&nbsp; // adjust machine name with the actual port</p><p>&nbsp; tmpPort = ss.getLocalPort();</p><p>&nbsp; selfAddr = new InetSocketAddress(ss.getInetAddress().getHostAddress(),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tmpPort);</p><p>&nbsp; this.dnRegistration.setName(machineName + ":" + tmpPort);</p><p>&nbsp; this.threadGroup = new ThreadGroup("dataXceiverServer");</p><p>&nbsp; this.dataXceiverServer = new Daemon(threadGroup, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new DataXceiverServer(ss, conf, this));</p><p>&nbsp; this.threadGroup.setDaemon(true); // auto destroy when empty </p><p>&nbsp; ……</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

DataXceiverServer.run()函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>public void run() { </p><p>&nbsp; while (datanode.shouldRun) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">接受客户端的链接</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Socket s = ss.accept();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.setTcpNoDelay(true);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">生成一个线程</span><span style="font-family:Verdana;">DataXceiver</span><span style="font-family:'宋体';">来对建立的链接提供服务</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new Daemon(datanode.threadGroup, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new DataXceiver(s, datanode, this)).start();</p><p>&nbsp; }</p><p>&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp; ss.close();</p><p>&nbsp; } catch (IOException ie) { </p><p>&nbsp;&nbsp;&nbsp; LOG.warn(datanode.dnRegistration + ":DataXceiveServer: " </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; + StringUtils.stringifyException(ie));</p><p>&nbsp; }</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

DataXceiver.run()函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>public void run() { </p><p>&nbsp; DataInputStream in=null; </p><p>&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">建立一个输入流，读取客户端发送的指令</span></strong></p><p>&nbsp;&nbsp;&nbsp; in = new DataInputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new BufferedInputStream(NetUtils.getInputStream(s), </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; SMALL_BUFFER_SIZE));</p><p>&nbsp;&nbsp;&nbsp; short version = in.readShort();</p><p>&nbsp;&nbsp;&nbsp; boolean local = s.getInetAddress().equals(s.getLocalAddress());</p><p>&nbsp;&nbsp;&nbsp; byte op = in.readByte();</p><p>&nbsp;&nbsp;&nbsp; // Make sure the xciver count is not exceeded</p><p>&nbsp;&nbsp;&nbsp; int curXceiverCount = datanode.getXceiverCount();</p><p>&nbsp;&nbsp;&nbsp; long startTime = DataNode.now();</p><p>&nbsp;&nbsp;&nbsp; switch ( op ) { </p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">读取</span></strong></p><p>&nbsp;&nbsp;&nbsp; case DataTransferProtocol.OP_READ_BLOCK:</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">真正的读取数据</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; readBlock( in );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.readBlockOp.inc(DataNode.now() - startTime);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (local)</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.readsFromLocalClient.inc();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; else</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.readsFromRemoteClient.inc();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; break;</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">写入</span></strong></p><p>&nbsp;&nbsp;&nbsp; case DataTransferProtocol.OP_WRITE_BLOCK:</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">真正的写入数据</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; writeBlock( in );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.writeBlockOp.inc(DataNode.now() - startTime);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (local)</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.writesFromLocalClient.inc();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; else</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.myMetrics.writesFromRemoteClient.inc();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; break;</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">其他的指令</span></strong></p><p>&nbsp;&nbsp;&nbsp; ……</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp; } catch (Throwable t) { </p><p>&nbsp;&nbsp;&nbsp; LOG.error(datanode.dnRegistration + ":DataXceiver",t);</p><p>&nbsp; } finally { </p><p>&nbsp;&nbsp;&nbsp; IOUtils.closeStream(in);</p><p>&nbsp;&nbsp;&nbsp; IOUtils.closeSocket(s);</p><p>&nbsp;&nbsp;&nbsp; dataXceiverServer.childSockets.remove(s);</p><p>&nbsp; }</p><p>} </p></td> 
  </tr> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>private void readBlock(DataInputStream in) throws IOException { </p><p>&nbsp; <strong>//<span style="font-family:'宋体';">读取指令</span></strong></p><p>&nbsp; long blockId = in.readLong();&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </p><p>&nbsp; Block block = new Block( blockId, 0 , in.readLong()); </p><p>&nbsp; long startOffset = in.readLong();</p><p>&nbsp; long length = in.readLong();</p><p>&nbsp; String clientName = Text.readString(in);</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">创建一个写入流，用于向客户端写数据</span></strong></p><p>&nbsp; OutputStream baseStream = NetUtils.getOutputStream(s, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanode.socketWriteTimeout);</p><p>&nbsp; DataOutputStream out = new DataOutputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new BufferedOutputStream(baseStream, SMALL_BUFFER_SIZE));</p><p>&nbsp; <strong>//<span style="font-family:'宋体';">生成</span><span style="font-family:Verdana;">BlockSender</span><span style="font-family:'宋体';">用于读取本地的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">的数据，并发送给客户端</span></strong></p><p>&nbsp; <strong>//BlockSender<span style="font-family:'宋体';">有一个成员变量</span><span style="font-family:Verdana;">InputStream blockIn</span><span style="font-family:'宋体';">用于读取本地</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">的数据</span></strong></p><p>&nbsp; BlockSender blockSender = new BlockSender(block, startOffset, length,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; true, true, false, datanode, clientTraceFmt);</p><p>&nbsp;&nbsp; out.writeShort(DataTransferProtocol.OP_STATUS_SUCCESS); // send op status</p><p>&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">向客户端写入数据</span></strong>&nbsp;</p><p>&nbsp;&nbsp; long read = blockSender.sendBlock(out, baseStream, null); </p><p>&nbsp;&nbsp; ……</p><p>&nbsp; } finally { </p><p>&nbsp;&nbsp;&nbsp; IOUtils.closeStream(out);</p><p>&nbsp;&nbsp;&nbsp; IOUtils.closeStream(blockSender);</p><p>&nbsp; }</p><p>} </p></td> 
  </tr> 
 </tbody> 
</table>

**三、文件的写入**

下面解析向hdfs上传一个文件的过程。

**3.1、客户端**

上传一个文件到hdfs，一般会调用DistributedFileSystem.create，其实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp;</p><p>&nbsp; public FSDataOutputStream create(Path f, FsPermission permission,</p><p>&nbsp;&nbsp;&nbsp; boolean overwrite,</p><p>&nbsp;&nbsp;&nbsp; int bufferSize, short replication, long blockSize,</p><p>&nbsp;&nbsp;&nbsp; Progressable progress) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; return new FSDataOutputStream</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (dfs.create(getPathName(f), permission,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; overwrite, replication, blockSize, progress, bufferSize),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; statistics);</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

其最终生成一个FSDataOutputStream用于向新生成的文件中写入数据。其成员变量dfs的类型为DFSClient，DFSClient的create函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; public OutputStream create(String src, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; FsPermission permission,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean overwrite, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; short replication,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long blockSize,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Progressable progress,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int buffersize</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; checkOpen();</p><p>&nbsp;&nbsp;&nbsp; if (permission == null) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; permission = FsPermission.getDefault();</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; FsPermission masked = permission.applyUMask(FsPermission.getUMask(conf));</p><p>&nbsp;&nbsp;&nbsp; OutputStream result = new DFSOutputStream(src, masked,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; overwrite, replication, blockSize, progress, buffersize,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; conf.getInt("io.bytes.per.checksum", 512));</p><p>&nbsp;&nbsp;&nbsp; leasechecker.put(src, result);</p><p>&nbsp;&nbsp;&nbsp; return result;</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

其中构造了一个DFSOutputStream，在其构造函数中，同过RPC调用NameNode的create来创建一个文件。   
当然，构造函数中还做了一件重要的事情，就是streamer.start()，也即启动了一个pipeline，用于写数据，在写入数据的过程中，我们会仔细分析。

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; DFSOutputStream(String src, FsPermission masked, boolean overwrite, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; short replication, long blockSize, Progressable progress,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int buffersize, int bytesPerChecksum) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; this(src, blockSize, progress, bytesPerChecksum); </p><p>&nbsp;&nbsp;&nbsp; computePacketChunkSize(writePacketSize, bytesPerChecksum); </p><p>&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; namenode.create(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; src, masked, clientName, overwrite, replication, blockSize);</p><p>&nbsp;&nbsp;&nbsp; } catch(RemoteException re) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw re.unwrapRemoteException(AccessControlException.class,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; QuotaExceededException.class);</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; streamer.start();</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

**3.2、NameNode**

NameNode的create函数调用namesystem.startFile函数，其又调用startFileInternal函数，实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; private synchronized void startFileInternal(String src,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; PermissionStatus permissions,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String holder, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String clientMachine, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean overwrite,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean append,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; short replication,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long blockSize</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建一个新的文件，状态为</span><span style="font-family:Verdana;">under construction</span><span style="font-family:'宋体';">，没有任何</span><span style="font-family:Verdana;">data block</span><span style="font-family:'宋体';">与之对应</span></strong></p><p>&nbsp;&nbsp; long genstamp = nextGenerationStamp();</p><p>&nbsp;&nbsp; INodeFileUnderConstruction newNode = dir.addFile(src, permissions,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; replication, blockSize, holder, clientMachine, clientNode, genstamp);</p><p>&nbsp;&nbsp; ......</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

**3.3、客户端**

下面轮到客户端向新创建的文件中写入数据了，一般会使用FSDataOutputStream的write函数，最终会调用DFSOutputStream的writeChunk函数：

按照hdfs的设计，对block的数据写入使用的是pipeline的方式，也即将数据分成一个个的package，如果需要复制三分，分别写入DataNode 1, 2, 3，则会进行如下的过程：

· 首先将package 1写入DataNode 1

· 然后由DataNode 1负责将package 1写入DataNode 2，同时客户端可以将pacage 2写入DataNode 1

· 然后DataNode 2负责将package 1写入DataNode 3, 同时客户端可以讲package 3写入DataNode 1，DataNode 1将package 2写入DataNode 2

· 就这样将一个个package排着队的传递下去，直到所有的数据全部写入并复制完毕

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; protected synchronized void writeChunk(byte[] b, int offset, int len, byte[] checksum) </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throws IOException { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建一个</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">，并写入数据</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; currentPacket = new Packet(packetSize, chunksPerPacket, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bytesCurBlock); </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; currentPacket.writeChecksum(checksum, 0, cklen);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; currentPacket.writeData(b, offset, len);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; currentPacket.numChunks++;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bytesCurBlock += len; </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果此</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">已满，则放入队列中准备发送</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (currentPacket.numChunks == currentPacket.maxChunks ||</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; bytesCurBlock == blockSize) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataQueue.addLast(currentPacket);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">唤醒等待</span><span style="font-family:Verdana;">dataqueue</span><span style="font-family:'宋体';">的传输线程，也即</span><span style="font-family:Verdana;">DataStreamer</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataQueue.notifyAll();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; currentPacket = null;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

DataStreamer的run函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; public void run() { </p><p>&nbsp;&nbsp;&nbsp; while (!closed &amp;&amp; clientRunning) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Packet one = null;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; synchronized (dataQueue) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果队列中没有</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">，则等待</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while ((!closed &amp;&amp; !hasError &amp;&amp; clientRunning </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &amp;&amp; dataQueue.size() == 0) || doSleep) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataQueue.wait(1000);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } catch (InterruptedException&nbsp; e) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; doSleep = false;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">得到队列中的第一个</span><span style="font-family:Verdana;">package</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; one = dataQueue.getFirst();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long offsetInBlock = one.offsetInBlock;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">由</span><span style="font-family:Verdana;">NameNode</span><span style="font-family:'宋体';">分配</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">，并生成一个写入流指向此</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (blockStream == null) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; nodes = nextBlockOutputStream(src); </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; response = new ResponseProcessor(nodes);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; response.start();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ByteBuffer buf = one.getBuffer();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">将</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">从</span><span style="font-family:Verdana;">dataQueue</span><span style="font-family:'宋体';">移至</span><span style="font-family:Verdana;">ackQueue,</span><span style="font-family:'宋体';">等待确认</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataQueue.removeFirst();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataQueue.notifyAll();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; synchronized (ackQueue) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ackQueue.addLast(one);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ackQueue.notifyAll();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">利用生成的写入流将数据写入</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">中的</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockStream.write(buf.array(), buf.position(), buf.remaining());</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (one.lastPacketInBlock) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockStream.writeInt(0); <strong>//<span style="font-family:'宋体';">表示此</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">写入完毕</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockStream.flush();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } catch (Throwable e) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

其中重要的一个函数是nextBlockOutputStream，实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; private DatanodeInfo[] nextBlockOutputStream(String client) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; LocatedBlock lb = null;</p><p>&nbsp;&nbsp;&nbsp; boolean retry = false;</p><p>&nbsp;&nbsp;&nbsp; DatanodeInfo[] nodes;</p><p>&nbsp;&nbsp;&nbsp; int count = conf.getInt("dfs.client.block.write.retries", 3);</p><p>&nbsp;&nbsp;&nbsp; boolean success;</p><p>&nbsp;&nbsp;&nbsp; do { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">由</span><span style="font-family:Verdana;">NameNode</span><span style="font-family:'宋体';">为文件分配</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">和</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; lb = locateFollowingBlock(startTime);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; block = lb.getBlock();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; nodes = lb.getLocations();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建向</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">的写入流</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; success = createBlockOutputStream(nodes, clientName, false);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; } while (retry &amp;&amp; --count &gt;= 0);</p><p>&nbsp;&nbsp;&nbsp; return nodes;</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

locateFollowingBlock中通过RPC调用namenode.addBlock(src, clientName)函数

**3.4、NameNode**

NameNode的addBlock函数实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; public LocatedBlock addBlock(String src, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String clientName) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; LocatedBlock locatedBlock = namesystem.getAdditionalBlock(src, clientName);</p><p>&nbsp;&nbsp;&nbsp; return locatedBlock;</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

FSNamesystem的getAdditionalBlock实现如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; public LocatedBlock getAdditionalBlock(String src, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String clientName</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; long fileLength, blockSize;</p><p>&nbsp;&nbsp;&nbsp; int replication;</p><p>&nbsp;&nbsp;&nbsp; DatanodeDescriptor clientNode = null;</p><p>&nbsp;&nbsp;&nbsp; Block newBlock = null;</p><p>&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">为新的</span><span style="font-family:Verdana;">block</span><span style="font-family:'宋体';">选择</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp; DatanodeDescriptor targets[] = replicator.chooseTarget(replication,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; clientNode,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; null,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockSize);</p><p>&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">得到文件路径中所有</span><span style="font-family:Verdana;">path</span><span style="font-family:'宋体';">的</span><span style="font-family:Verdana;">INode</span><span style="font-family:'宋体';">，其中最后一个是新添加的文件对的</span><span style="font-family:Verdana;">INode</span><span style="font-family:'宋体';">，状态为</span><span style="font-family:Verdana;">under construction</span></strong>&nbsp;</p><p>&nbsp;&nbsp;&nbsp; INode[] pathINodes = dir.getExistingPathINodes(src);</p><p>&nbsp;&nbsp;&nbsp; int inodesLen = pathINodes.length;</p><p>&nbsp;&nbsp;&nbsp; INodeFileUnderConstruction pendingFile&nbsp; = (INodeFileUnderConstruction) </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; pathINodes[inodesLen - 1];</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">为文件分配</span><span style="font-family:Verdana;">block, </span><span style="font-family:'宋体';">并设置在那写</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">上</span></strong></p><p>&nbsp;&nbsp;&nbsp; newBlock = allocateBlock(src, pathINodes);</p><p>&nbsp;&nbsp;&nbsp; pendingFile.setTargets(targets);</p><p>&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; return new LocatedBlock(newBlock, targets, fileLength);</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

**3.5、客户端**

在分配了DataNode和block以后，createBlockOutputStream开始写入数据。

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; private boolean createBlockOutputStream(DatanodeInfo[] nodes, String client,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; boolean recoveryFlag) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建一个</span><span style="font-family:Verdana;">socket</span><span style="font-family:'宋体';">，链接</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; InetSocketAddress target = NetUtils.createSocketAddr(nodes[0].getName());</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s = socketFactory.createSocket();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int timeoutValue = 3000 * nodes.length + socketTimeout;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.connect(target, timeoutValue);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.setSoTimeout(timeoutValue);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.setSendBufferSize(DEFAULT_DATA_SOCKET_SIZE);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; long writeTimeout = HdfsConstants.WRITE_TIMEOUT_EXTENSION * nodes.length +</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; datanodeWriteTimeout;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DataOutputStream out = new DataOutputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new BufferedOutputStream(NetUtils.getOutputStream(s, writeTimeout), </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DataNode.SMALL_BUFFER_SIZE));</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockReplyStream = new DataInputStream(NetUtils.getInputStream(s));</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">写入指令</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeShort( DataTransferProtocol.DATA_TRANSFER_VERSION );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.write( DataTransferProtocol.OP_WRITE_BLOCK );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeLong( block.getBlockId() );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeLong( block.getGenerationStamp() );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeInt( nodes.length );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeBoolean( recoveryFlag );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Text.writeString( out, client );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeBoolean(false);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.writeInt( nodes.length - 1 );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">注意，次循环从</span><span style="font-family:Verdana;">1</span><span style="font-family:'宋体';">开始，而非从</span><span style="font-family:Verdana;">0</span><span style="font-family:'宋体';">开始。将除了第一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">以外的另外两个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">的信息发送给第一个</span><span style="font-family:Verdana;">DataNode, </span><span style="font-family:'宋体';">第一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">可以根据此信息将数据写给另两个</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for (int i = 1; i &lt; nodes.length; i++) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; nodes[i].write(out);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; checksum.writeHeader( out );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; out.flush();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; firstBadLink = Text.readString(blockReplyStream);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (firstBadLink.length() != 0) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new IOException("Bad connect ack with firstBadLink " + firstBadLink);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockStream = out;</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

客户端在DataStreamer的run函数中创建了写入流后，调用blockStream.write将数据写入DataNode

**3.6、DataNode**

DataNode的DataXceiver中，收到指令DataTransferProtocol.OP\_WRITE\_BLOCK则调用writeBlock函数：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; private void writeBlock(DataInputStream in) throws IOException { </p><p>&nbsp;&nbsp;&nbsp; DatanodeInfo srcDataNode = null;</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">读入头信息</span></strong></p><p>&nbsp;&nbsp;&nbsp; Block block = new Block(in.readLong(), </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; dataXceiverServer.estimateBlockSize, in.readLong());</p><p>&nbsp;&nbsp;&nbsp; int pipelineSize = in.readInt(); // num of datanodes in entire pipeline</p><p>&nbsp;&nbsp;&nbsp; boolean isRecovery = in.readBoolean(); // is this part of recovery?</p><p>&nbsp;&nbsp;&nbsp; String client = Text.readString(in); // working on behalf of this client</p><p>&nbsp;&nbsp;&nbsp; boolean hasSrcDataNode = in.readBoolean(); // is src node info present</p><p>&nbsp;&nbsp;&nbsp; if (hasSrcDataNode) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; srcDataNode = new DatanodeInfo();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; srcDataNode.readFields(in);</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; int numTargets = in.readInt();</p><p>&nbsp;&nbsp;&nbsp; if (numTargets &lt; 0) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new IOException("Mislabelled incoming datastream.");</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">读入剩下的</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">列表，如果当前是第一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">，则此列表中收到的是第二个，第三个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">的信息，如果当前是第二个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">，则受到的是第三个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">的信息</span></strong></p><p>&nbsp;&nbsp;&nbsp; DatanodeInfo targets[] = new DatanodeInfo[numTargets];</p><p>&nbsp;&nbsp;&nbsp; for (int i = 0; i &lt; targets.length; i++) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DatanodeInfo tmp = new DatanodeInfo();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; tmp.readFields(in);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; targets[i] = tmp;</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; DataOutputStream mirrorOut = null;&nbsp; // stream to next target</p><p>&nbsp;&nbsp;&nbsp; DataInputStream mirrorIn = null;&nbsp;&nbsp;&nbsp; // reply from next target</p><p>&nbsp;&nbsp;&nbsp; DataOutputStream replyOut = null;&nbsp;&nbsp; // stream to prev target</p><p>&nbsp;&nbsp;&nbsp; Socket mirrorSock = null;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // socket to next target</p><p>&nbsp;&nbsp;&nbsp; BlockReceiver blockReceiver = null; // responsible for data handling</p><p>&nbsp;&nbsp;&nbsp; String mirrorNode = null;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // the name:port of next target</p><p>&nbsp;&nbsp;&nbsp; String firstBadLink = "";&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // first datanode that failed in connection setup</p><p>&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">生成一个</span><span style="font-family:Verdana;">BlockReceiver, </span><span style="font-family:'宋体';">其有成员变量</span><span style="font-family:Verdana;">DataInputStream in</span><span style="font-family:'宋体';">为从客户端或者上一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">读取数据，还有成员变量</span><span style="font-family:Verdana;">DataOutputStream mirrorOut</span><span style="font-family:'宋体';">，用于向下一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">写入数据，还有成员变量</span><span style="font-family:Verdana;">OutputStream out</span><span style="font-family:'宋体';">用于将数据写入本地。</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockReceiver = new BlockReceiver(block, in, </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.getRemoteSocketAddress().toString(),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s.getLocalSocketAddress().toString(),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; isRecovery, client, srcDataNode, datanode);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // get a connection back to the previous target</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; replyOut = new DataOutputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; NetUtils.getOutputStream(s, datanode.socketWriteTimeout));</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">如果当前不是最后一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">，则同下一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">建立</span><span style="font-family:Verdana;">socket</span><span style="font-family:'宋体';">连接</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (targets.length &gt; 0) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; InetSocketAddress mirrorTarget = null;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // Connect to backup machine</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorNode = targets[0].getName();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorTarget = NetUtils.createSocketAddr(mirrorNode);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorSock = datanode.newSocket();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int timeoutValue = numTargets * datanode.socketTimeout;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int writeTimeout = datanode.socketWriteTimeout + </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; (HdfsConstants.WRITE_TIMEOUT_EXTENSION * numTargets);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorSock.connect(mirrorTarget, timeoutValue);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorSock.setSoTimeout(timeoutValue);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorSock.setSendBufferSize(DEFAULT_DATA_SOCKET_SIZE);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">创建向下一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">写入数据的流</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut = new DataOutputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new BufferedOutputStream(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; NetUtils.getOutputStream(mirrorSock, writeTimeout),</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; SMALL_BUFFER_SIZE));</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorIn = new DataInputStream(NetUtils.getInputStream(mirrorSock));</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeShort( DataTransferProtocol.DATA_TRANSFER_VERSION );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.write( DataTransferProtocol.OP_WRITE_BLOCK );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeLong( block.getBlockId() );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeLong( block.getGenerationStamp() );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeInt( pipelineSize );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeBoolean( isRecovery );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Text.writeString( mirrorOut, client );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeBoolean(hasSrcDataNode);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (hasSrcDataNode) { // pass src node information</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; srcDataNode.write(mirrorOut);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeInt( targets.length - 1 );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">此出也是从</span><span style="font-family:Verdana;">1</span><span style="font-family:'宋体';">开始，将除了下一个</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">的其他</span><span style="font-family:Verdana;">DataNode</span><span style="font-family:'宋体';">信息发送给下一个</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; for ( int i = 1; i &lt; targets.length; i++ ) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; targets[i].write( mirrorOut );</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockReceiver.writeChecksumHeader(mirrorOut);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.flush();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">使用</span><span style="font-family:Verdana;">BlockReceiver</span><span style="font-family:'宋体';">接受</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String mirrorAddr = (mirrorSock == null) ? null : mirrorNode;</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; blockReceiver.receiveBlock(mirrorOut, mirrorIn, replyOut,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorAddr, null, targets.length);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; } finally { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // close all opened streams</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeStream(mirrorOut);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeStream(mirrorIn);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeStream(replyOut);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeSocket(mirrorSock);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; IOUtils.closeStream(blockReceiver);</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

BlockReceiver的receiveBlock函数中，一段重要的逻辑如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; void receiveBlock(</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DataOutputStream mirrOut, // output to next datanode</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DataInputStream mirrIn,&nbsp;&nbsp; // input from next datanode</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; DataOutputStream replyOut,&nbsp; // output to previous datanode</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; String mirrAddr, BlockTransferThrottler throttlerArg,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; int numTargets) throws IOException { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">不断的接受</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">，直到结束</span></strong></p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while (receivePacket() &gt; 0) {}</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if (mirrorOut != null) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.writeInt(0); // mark the end of the block</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.flush();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } catch (IOException e) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; handleMirrorOutError(e);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

BlockReceiver的receivePacket函数如下：

<table> 
 <tbody> 
  <tr> 
   <td style="background:rgb(219,229,241);"><p>&nbsp; private int receivePacket() throws IOException { </p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">从客户端或者上一个节点接收一个</span><span style="font-family:Verdana;">package</span></strong></p><p>&nbsp;&nbsp;&nbsp; int payloadLen = readNextPacket();</p><p>&nbsp;&nbsp;&nbsp; buf.mark();</p><p>&nbsp;&nbsp;&nbsp; //read the header</p><p>&nbsp;&nbsp;&nbsp; buf.getInt(); // packet length</p><p>&nbsp;&nbsp;&nbsp; offsetInBlock = buf.getLong(); // get offset of packet in block</p><p>&nbsp;&nbsp;&nbsp; long seqno = buf.getLong();&nbsp;&nbsp;&nbsp; // get seqno</p><p>&nbsp;&nbsp;&nbsp; boolean lastPacketInBlock = (buf.get() != 0);</p><p>&nbsp;&nbsp;&nbsp; int endOfHeader = buf.position();</p><p>&nbsp;&nbsp;&nbsp; buf.reset();</p><p>&nbsp;&nbsp;&nbsp; setBlockPosition(offsetInBlock);</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">将</span><span style="font-family:Verdana;">package</span><span style="font-family:'宋体';">写入下一个</span><span style="font-family:Verdana;">DataNode</span></strong></p><p>&nbsp;&nbsp;&nbsp; if (mirrorOut != null) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.write(buf.array(), buf.position(), buf.remaining());</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; mirrorOut.flush();</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } catch (IOException e) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; handleMirrorOutError(e);</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; buf.position(endOfHeader);&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </p><p>&nbsp;&nbsp;&nbsp; int len = buf.getInt();</p><p>&nbsp;&nbsp;&nbsp; offsetInBlock += len;</p><p>&nbsp;&nbsp;&nbsp; int checksumLen = ((len + bytesPerChecksum - 1)/bytesPerChecksum)*</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; checksumSize;</p><p>&nbsp;&nbsp;&nbsp; int checksumOff = buf.position();</p><p>&nbsp;&nbsp;&nbsp; int dataOff = checksumOff + checksumLen;</p><p>&nbsp;&nbsp;&nbsp; byte pktBuf[] = buf.array();</p><p>&nbsp;&nbsp;&nbsp; buf.position(buf.limit()); // move to the end of the data.</p><p>&nbsp;&nbsp;&nbsp; ......</p><p>&nbsp;&nbsp;&nbsp; <strong>//<span style="font-family:'宋体';">将数据写入本地的</span><span style="font-family:Verdana;">block</span></strong></p><p>&nbsp;&nbsp;&nbsp; out.write(pktBuf, dataOff, len);</p><p>&nbsp;&nbsp;&nbsp; /// flush entire packet before sending ack</p><p>&nbsp;&nbsp;&nbsp; flush();</p><p>&nbsp;&nbsp;&nbsp; // put in queue for pending acks</p><p>&nbsp;&nbsp;&nbsp; if (responder != null) { </p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ((PacketResponder)responder.getRunnable()).enqueue(seqno,</p><p>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; lastPacketInBlock); </p><p>&nbsp;&nbsp;&nbsp; }</p><p>&nbsp;&nbsp;&nbsp; return payloadLen;</p><p>&nbsp; } </p></td> 
  </tr> 
 </tbody> 
</table>

发表评论取消回复

表情：

评论列表（有 0 条评论，210人围观）

还没有评论，来说两句吧...

相关阅读

相关 HDFS读写流程

数据流读流程的详解 > 读操作： > > hdfs dfs -get /file02 ./file02 > > hdfs dfs

素颜马尾好姑娘i/ 2023年10月13日 15:48/ 0 赞/ 86 阅读

相关 HDFS读写流程

文章目录 1 读 1.1 Client 访问NN 查询元数据信息 1.2 返回输入流对象(FSData InputStrea

小鱼儿/ 2022年11月27日 12:26/ 0 赞/ 254 阅读

相关 HDFS数据的读写过程

1.一般的文件读取操作包括：open 、read、close等 ![Center][] 客户端读取数据过程，其中1、3、6步由客户端发起：客户端首先获取FileSyst

痛定思痛。/ 2022年07月24日 10:26/ 0 赞/ 193 阅读

相关 HDFS读写过程解析

一、文件的打开 1.1、客户端 HDFS打开一个文件，需要在客户端调用DistributedFileSystem.open(Path f, int bufferSize)，

古城微笑少年丶/ 2022年05月26日 09:53/ 0 赞/ 211 阅读

相关 HDFS写过程

1.初始化FileSystem，客户端调用create()来创建文件 2.FileSystem用RPC调用元数据节点，在文件系统的命名空间中创建一个新的文件，元数据节点首先确

叁歲伎倆/ 2022年05月24日 11:19/ 0 赞/ 171 阅读

相关 HDFS读过程

1.初始化FileSystem，然后客户端(client)用FileSystem的open()函数打开文件 2.FileSystem用RPC调用元数据节点，得到文件的数据块信

分手后的思念是犯贱/ 2022年05月24日 11:19/ 0 赞/ 181 阅读

相关 HDFS的读写流程

HDFS写流程 ![70][] NameNode负责管理存储在HDFS.上所有文件的元数据，它会确认客户端的请求，并记录下文件的名字和存储这个文件的DataNode

╰+攻爆jí腚メ/ 2022年05月20日 08:43/ 0 赞/ 491 阅读

相关 Linux读写权限解析

本文转自：[https://blog.csdn.net/luoshuitianyi/article/details/78112498][https_blog.csdn.net_

喜欢ヅ旅行/ 2022年05月13日 03:20/ 0 赞/ 217 阅读

相关 ReentrantReadWriteLock读写锁解析

获取读锁readLock ReadLock readLock = readWriteLock.readLock(); 调用lock开始锁定 public

朴灿烈づ我的快乐病毒、/ 2022年03月17日 17:56/ 0 赞/ 343 阅读

相关 HDFS 读写流程

HDFS简介 HDFS全称是Hadoop Distributed System。HDFS是为以流的方式存取大文件而设计的。适用于几百MB，GB以及TB，并写一次读多次的场合。

r囧r小猫/ 2022年03月12日 03:22/ 0 赞/ 292 阅读