FileSystem hdfs = FileSystem.get(new Configuration());
Path path = new Path("/testfile");// reading
FSDataInputStream dis = hdfs.open(path);
byte[] writeBuf = new byte[1024];
int len = dis.read(writeBuf);
System.out.println(new String(writeBuf, 0, len, "UTF-8"));
dis.close();hdfs.close();

public FSDataInputStream open(Path f, int bufferSize) throws IOException {return new DFSClient.DFSDataInputStream(dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics));
}

synchronized void openInfo() throws IOException {  LocatedBlocks newInfo = callGetBlockLocations(namenode, src, 0, prefetchSize);this.locatedBlocks = newInfo;this.currentNode = null;
}private static LocatedBlocks callGetBlockLocations(ClientProtocol namenode,String src, long start, long length) throws IOException {return namenode.getBlockLocations(src, start, length);
}

public synchronized int read(byte buf[], int off, int len) throws IOException {...if (pos < getFileLength()) {int retries = 2;while (retries > 0) {try {if (pos > blockEnd) {//首次pos=0,blockEnd=-1,必定调用方法blockSeekTo,初始化blockEnd,以后是读完了当前块，需要读下一个块，才会调用blockSeekTocurrentNode = blockSeekTo(pos);//根据pos选择块和数据节点，选择算法是遍历块所在的所有数据节点，选择第一个非死亡节点
        }int realLen = Math.min(len, (int) (blockEnd - pos + 1));int result = readBuffer(buf, off, realLen);if (result >= 0) {pos += result;} else {throw new IOException("Unexpected EOS from the reader");}...return result;} catch (ChecksumException ce) {throw ce;            } catch (IOException e) {...if (currentNode != null) { addToDeadNodes(currentNode); }//遇到无法读的DataNode,添加到死亡节点if (--retries == 0) {//尝试读三次都失败，就抛出异常throw e;}}}}return -1;
}

public int read(long position, byte[] buffer, int offset, int length)throws IOException {
  long filelen = getFileLength();
  int realLen = length;if ((position + length) > filelen) {realLen = (int)(filelen - position);}//首先得到包含从offset到offset + length内容的block列表//比如对于64M一个block的文件系统来说，欲读取从100M开始，长度为128M的数据，则block列表包括第2，3，4块blockList<LocatedBlock> blockRange = getBlockRange(position, realLen);int remaining = realLen;//对每一个block，从中读取内容//对于上面的例子，对于第2块block，读取从36M开始，读取长度28M，对于第3块，读取整一块64M，对于第4块，读取从0开始，长度为36M，共128M数据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;}...return realLen;
}

private void fetchBlockByteRange(LocatedBlock block, long start,long end, byte[] buf, int offset) throws IOException {Socket dn = null;int numAttempts = block.getLocations().length;//此while循环为读取失败后的重试次数while (dn == null && numAttempts-- > 0 ) {//选择一个DataNode来读取数据DNAddrPair retval = chooseDataNode(block);DatanodeInfo chosenNode = retval.info;InetSocketAddress targetAddr = retval.addr;BlockReader reader = null;int len = (int) (end - start + 1);try {if (shouldTryShortCircuitRead(targetAddr)) {//如果要读取的块所属的DataNode与客户端是同一个节点，直接通过本地磁盘访问，减少网络流量reader = getLocalBlockReader(conf, src, block.getBlock(),accessToken, chosenNode, DFSClient.this.socketTimeout, start);} else {//创建Socket连接到DataNodedn = socketFactory.createSocket();dn.connect(targetAddr, socketTimeout);dn.setSoTimeout(socketTimeout);//利用建立的Socket链接，生成一个reader负责从DataNode读取数据reader = BlockReader.newBlockReader(dn, src, block.getBlock().getBlockId(), accessToken,block.getBlock().getGenerationStamp(), start, len, buffersize, verifyChecksum, clientName);}        //读取数据int nread = reader.readAll(buf, offset, len);return;} finally {IOUtils.closeStream(reader);IOUtils.closeSocket(dn);dn = null;}//如果读取失败，则将此DataNode标记为失败节点
    addToDeadNodes(chosenNode);}
}

public static BlockReader newBlockReader( Socket sock, String file,long blockId,long genStamp,long startOffset, long len,int bufferSize, boolean verifyChecksum,String clientName) throws IOException {//使用Socket建立写入流，向DataNode发送读指令DataOutputStream out = new DataOutputStream(new BufferedOutputStream(NetUtils.getOutputStream(sock,HdfsConstants.WRITE_TIMEOUT)));out.writeShort( DataTransferProtocol.DATA_TRANSFER_VERSION );out.write( DataTransferProtocol.OP_READ_BLOCK );out.writeLong( blockId );out.writeLong( genStamp );out.writeLong( startOffset );out.writeLong( len );Text.writeString(out, clientName);out.flush();//使用Socket建立读入流，用于从DataNode读取数据DataInputStream in = new DataInputStream(new BufferedInputStream(NetUtils.getInputStream(sock),bufferSize));short status = in.readShort();//块读取的状态标记，一般是成功DataChecksum checksum = DataChecksum.newDataChecksum( in );long firstChunkOffset = in.readLong();//生成一个reader，主要包含读入流，用于读取数据return new BlockReader( file, blockId, in, checksum, verifyChecksum,startOffset, firstChunkOffset, sock );
}

private void readBlock(DataInputStream in) throws IOException {//读取指令long blockId = in.readLong();         Block block = new Block( blockId, 0 , in.readLong());long startOffset = in.readLong();long length = in.readLong();String clientName = Text.readString(in);//创建一个写入流，用于向客户端写数据OutputStream baseStream = NetUtils.getOutputStream(s,datanode.socketWriteTimeout);DataOutputStream out = new DataOutputStream(new BufferedOutputStream(baseStream, SMALL_BUFFER_SIZE));//生成BlockSender用于读取本地的block的数据，并发送给客户端//BlockSender有一个成员变量InputStream blockIn用于读取本地block的数据BlockSender blockSender = new BlockSender(block, startOffset, length,true, true, false, datanode, clientTraceFmt);out.writeShort(DataTransferProtocol.OP_STATUS_SUCCESS); // 发送操作成功的状态//向客户端写入数据long read = blockSender.sendBlock(out, baseStream, null);……} finally {IOUtils.closeStream(out);IOUtils.closeStream(blockSender);}
}

long sendBlock(DataOutputStream out, OutputStream baseStream, BlockTransferThrottler throttler) throws IOException {...try {try {checksum.writeHeader(out);//写入checksum的加密类型和加密长度if ( chunkOffsetOK ) {out.writeLong( offset );}out.flush();} catch (IOException e) { //socket errorthrow ioeToSocketException(e);}...ByteBuffer pktBuf = ByteBuffer.allocate(pktSize);while (endOffset > offset) {//循环写入数据包long len = sendChunks(pktBuf, maxChunksPerPacket, streamForSendChunks);offset += len;totalRead += len + ((len + bytesPerChecksum - 1)/bytesPerChecksum*checksumSize);seqno++;}try {out.writeInt(0); //标记结束
        out.flush();} catch (IOException e) { //socket errorthrow ioeToSocketException(e);}}...return totalRead;
}

//默认是crc校验，bytesPerChecksum默认是512，checksumSize默认是4，表示数据块每512个字节，做一次checksum校验，checksum的结果是4个字节
private int sendChunks(ByteBuffer pkt, int maxChunks, OutputStream out) throws IOException {int len = Math.min((int) (endOffset - offset),bytesPerChecksum * maxChunks);//len是要发送的数据长度if (len == 0) {return 0;}int numChunks = (len + bytesPerChecksum - 1)/bytesPerChecksum;//这次要发送的chunk数量int packetLen = len + numChunks*checksumSize + 4;//packetLen是整个包的长度，包括包头，校验码，数据
    pkt.clear();// write packet headerpkt.putInt(packetLen);//整个packet的长度pkt.putLong(offset);//块的偏移量pkt.putLong(seqno);//序列号pkt.put((byte)((offset + len >= endOffset) ? 1 : 0));//是否最后一个packetpkt.putInt(len);//发送的数据长度int checksumOff = pkt.position();int checksumLen = numChunks * checksumSize;byte[] buf = pkt.array();if (checksumSize > 0 && checksumIn != null) {try {checksumIn.readFully(buf, checksumOff, checksumLen);//填充chucksum的内容} catch (IOException e) {...}}int dataOff = checksumOff + checksumLen;if (blockInPosition < 0) {IOUtils.readFully(blockIn, buf, dataOff, len);//填充块数据的内容if (verifyChecksum) {//默认是false,不验证//校验处理
      }}try {//通过socket发送数据到客户端
      } catch (IOException e) {throw ioeToSocketException(e);}...return len;
}

public synchronized int read(byte[] buf, int off, int len) throws IOException {//第一次read, 忽略前面的额外数据if (lastChunkLen < 0 && startOffset > firstChunkOffset && len > 0) {int toSkip = (int)(startOffset - firstChunkOffset);if ( skipBuf == null ) {skipBuf = new byte[bytesPerChecksum];}if ( super.read(skipBuf, 0, toSkip) != toSkip ) {//忽略// should never happenthrow new IOException("Could not skip required number of bytes");}}boolean eosBefore = gotEOS;int nRead = super.read(buf, off, len);// if gotEOS was set in the previous read and checksum is enabled :if (dnSock != null && gotEOS && !eosBefore && nRead >= 0&& needChecksum()) {//checksum is verified and there are no errors.
        checksumOk(dnSock);}return nRead;
}

public synchronized int read(byte[] b, int off, int len) throws IOException {//参数检查int n = 0;for (;;) {int nread = read1(b, off + n, len - n);if (nread <= 0) return (n == 0) ? nread : n;n += nread;if (n >= len)return n;}
}
//read1的len被忽略，只返回一个chunk的数据长度(最后一个chunk可能不足一个完整chunk的长度)
private int read1(byte b[], int off, int len)throws IOException {int avail = count-pos;if( avail <= 0 ) {if(len>=buf.length) {//直接读取一个数据chunk到用户buffer，避免多余一次复制        //很巧妙，buf初始化的大小是chunk的大小，默认是512，这里的代码会在块的剩余内容大于一个chunk的大小时调用int nread = readChecksumChunk(b, off, len);return nread;} else {//读取一个数据chunk到本地buffer,也是调用readChecksumChunk方法        //很巧妙，buf初始化大小是chunk的大小，默认是512，这里的代码会在块的剩余内容不足一个chunk的大小时进入调用
        fill();if( count <= 0 ) {return -1;} else {avail = count;}}}//从本地buffer拷贝数据到用户buffer，避免最后一个chunk导致数组越界int cnt = (avail < len) ? avail : len;System.arraycopy(buf, pos, b, off, cnt);pos += cnt;return cnt;
}

//只返回一个chunk的数据长度(默认512，最后一个chunk可能不足一个完整chunk的长度)
private int readChecksumChunk(byte b[], int off, int len)throws IOException {// invalidate buffercount = pos = 0;int read = 0;boolean retry = true;int retriesLeft = numOfRetries; //本案例中，numOfRetries是1，也就是说不会多次尝试do {retriesLeft--;try {read = readChunk(chunkPos, b, off, len, checksum);if( read > 0 ) {if( needChecksum() ) {//这里会做checksum校验
            sum.update(b, off, read);verifySum(chunkPos);}chunkPos += read;} retry = false;} catch (ChecksumException ce) {...if (retriesLeft == 0) {//本案例中，numOfRetries是1，也就是说不会多次尝试,失败了，直接抛出异常throw ce;}//如果读取的chunk校验失败，以当前的chunkpos为起始偏移量，尝试新的副本if (seekToNewSource(chunkPos)) {seek(chunkPos);} else {//找不到新的副本,抛出异常throw ce;}}} while (retry);return read;
}

//只返回一个chunk的数据长度(默认512，最后一个chunk可能不足一个完整chunk的长度)
protected synchronized int readChunk(long pos, byte[] buf, int offset,int len, byte[] checksumBuf) throws IOException {//读取一个 DATA_CHUNK.long chunkOffset = lastChunkOffset;if ( lastChunkLen > 0 ) {chunkOffset += lastChunkLen;}//如果先前的packet已经读取完毕，就读下一个packet。if (dataLeft <= 0) {//读包的头部int packetLen = in.readInt();long offsetInBlock = in.readLong();long seqno = in.readLong();boolean lastPacketInBlock = in.readBoolean();int dataLen = in.readInt();//校验长度
        lastSeqNo = seqno;isLastPacket = lastPacketInBlock;dataLeft = dataLen;adjustChecksumBytes(dataLen);if (dataLen > 0) {IOUtils.readFully(in, checksumBytes.array(), 0,checksumBytes.limit());//读取当前包的所有数据块内容对应的checksum,后面的流程会讲checksum和读取的chunk内容做校验
        }}int chunkLen = Math.min(dataLeft, bytesPerChecksum); //确定此次读取的chunk长度，正常情况下是一个bytesPerChecksum(512字节)，当文件最后不足一个bytesPerChecksum,读取剩余的内容。if ( chunkLen > 0 ) {IOUtils.readFully(in, buf, offset, chunkLen);//读取一个数据块的chunkchecksumBytes.get(checksumBuf, 0, checksumSize);}dataLeft -= chunkLen;lastChunkOffset = chunkOffset;lastChunkLen = chunkLen;...if ( chunkLen == 0 ) {return -1;}return chunkLen;
}