public static void read(Path path) throws IOException{    FileSystem hdfs = HdfsUtils.getFilesystem();  //步骤 1    FSDataInputStream fsDataInputStream =  hdfs.open(path); //步骤 2    IOUtils.copyBytes(fsDataInputStream, System.out, 4096,false);  //步骤 3}

package com.yq.common; import java.net.URI; import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem; public class HdfsUtils {    public static FileSystem getFilesystem(){        FileSystem hdfs=null;        Configuration conf=new Configuration();         try{            URI uri = new URI("hdfs://localhost:9000");            hdfs = FileSystem.get(uri,conf);        }        catch(Exception ex){        //        }    return hdfs;    }}

FSDataInputStream fsDataInputStream =  hdfs.open(path);

  public FSDataInputStream open(Path f) throws IOException {    return open(f, getConf().getInt("io.file.buffer.size", 4096));  }

  public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException;      //这个方法在DistributedFileSystem类有实现，如下  @Override  public FSDataInputStream open(Path f, final int bufferSize) throws IOException {    statistics.incrementReadOps(1);    Path absF = fixRelativePart(f);    return new FileSystemLinkResolver<FSDataInputStream>() {      @Override      public FSDataInputStream doCall(final Path p)          throws IOException, UnresolvedLinkException {        return new HdfsDataInputStream(            dfs.open(getPathName(p), bufferSize, verifyChecksum));      }      @Override      public FSDataInputStream next(final FileSystem fs, final Path p)          throws IOException {        return fs.open(p, bufferSize);      }    }.resolve(this, absF);  }

  public DFSInputStream open(String src, int buffersize, boolean verifyChecksum)      throws IOException, UnresolvedLinkException {    checkOpen();    //    Get block info from namenode    return new DFSInputStream(this, src, buffersize, verifyChecksum);  }

  DFSInputStream(DFSClient dfsClient, String src, int buffersize, boolean verifyChecksum                 ) throws IOException, UnresolvedLinkException {    this.dfsClient = dfsClient;    this.verifyChecksum = verifyChecksum;    this.buffersize = buffersize;    this.src = src;    this.cachingStrategy =        dfsClient.getDefaultReadCachingStrategy();    openInfo();  }

 /**   * Grab the open-file info from namenode   */  synchronized void openInfo() throws IOException, UnresolvedLinkException {    lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();    int retriesForLastBlockLength = dfsClient.getConf().retryTimesForGetLastBlockLength;    while (retriesForLastBlockLength > 0) {      // Getting last block length as -1 is a special case. When cluster      // restarts, DNs may not report immediately. At this time partial block      // locations will not be available with NN for getting the length. Lets      // retry for 3 times to get the length.      if (lastBlockBeingWrittenLength == -1) {        DFSClient.LOG.warn("Last block locations not available. "            + "Datanodes might not have reported blocks completely."            + " Will retry for " + retriesForLastBlockLength + " times");        waitFor(dfsClient.getConf().retryIntervalForGetLastBlockLength);        lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();      } else {        break;      }      retriesForLastBlockLength--;    }    if (retriesForLastBlockLength == 0) {      throw new IOException("Could not obtain the last block locations.");    }  }

  private long fetchLocatedBlocksAndGetLastBlockLength() throws IOException {    final LocatedBlocks newInfo = dfsClient.getLocatedBlocks(src, 0);    if (DFSClient.LOG.isDebugEnabled()) {      DFSClient.LOG.debug("newInfo = " + newInfo);    }    if (newInfo == null) {      throw new IOException("Cannot open filename " + src);    }     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!");        }      }    }    locatedBlocks = newInfo;    long lastBlockBeingWrittenLength = 0;    if (!locatedBlocks.isLastBlockComplete()) {      final LocatedBlock last = locatedBlocks.getLastLocatedBlock();      if (last != null) {        if (last.getLocations().length == 0) {          if (last.getBlockSize() == 0) {            // if the length is zero, then no data has been written to            // datanode. So no need to wait for the locations.            return 0;          }          return -1;        }        final long len = readBlockLength(last);        last.getBlock().setNumBytes(len);        lastBlockBeingWrittenLength = len;       }    }     currentNode = null;    return lastBlockBeingWrittenLength;  }

 private final List<LocatedBlock> blocks; // array of blocks with prioritized locations

 public static void copyBytes(InputStream in, OutputStream out, int buffSize, boolean close)     throws IOException {    try {      copyBytes(in, out, buffSize);      if(close) {        out.close();        out = null;        in.close();        in = null;      }    } finally {      if(close) {        closeStream(out);        closeStream(in);      }    }  }

public static void copyBytes(InputStream in, OutputStream out, int buffSize)     throws IOException {    PrintStream ps = out instanceof PrintStream ? (PrintStream)out : null;    byte buf[] = new byte[buffSize];    int bytesRead = in.read(buf);    while (bytesRead >= 0) {      out.write(buf, 0, bytesRead);      if ((ps != null) && ps.checkError()) {        throw new IOException("Unable to write to output stream.");      }      bytesRead = in.read(buf);    }  }