Hadoop之HDFS原理及文件上传下载源码分析（下）

qwe3223678qwe

//DFSOutputStream.java　　
  static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
　　
       FsPermission masked, EnumSet<CreateFlag> flag, boolean createParent,
　　
       short replication, long blockSize, Progressable progress, int buffersize,
　　
       DataChecksum checksum, String[] favoredNodes) throws IOException {
　　
     TraceScope scope =
　　
         dfsClient.getPathTraceScope("newStreamForCreate", src);
　　
     try {
　　
       HdfsFileStatus stat = null;
　　
       boolean shouldRetry = true;
　　
       int retryCount = CREATE_RETRY_COUNT;
　　
       while (shouldRetry) {
　　
         shouldRetry = false;
　　
         try {
　　
           //这里通过dfsClient的NameNode代理对象调用NameNodeRpcServer中实现的create()方法
　　
           stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
　　
               new EnumSetWritable<CreateFlag>(flag), createParent, replication,
　　
               blockSize, SUPPORTED_CRYPTO_VERSIONS);
　　
           break;
　　
         } catch (RemoteException re) {
　　
           IOException e = re.unwrapRemoteException(
　　
               AccessControlException.class,
　　
               DSQuotaExceededException.class,
　　
               FileAlreadyExistsException.class,
　　
               FileNotFoundException.class,
　　
               ParentNotDirectoryException.class,
　　
               NSQuotaExceededException.class,
　　
               RetryStartFileException.class,
　　
               SafeModeException.class,
　　
               UnresolvedPathException.class,
　　
               SnapshotAccessControlException.class,
　　
               UnknownCryptoProtocolVersionException.class);
　　
           if (e instanceof RetryStartFileException) {
　　
             if (retryCount > 0) {
　　
               shouldRetry = true;
　　
               retryCount--;
　　
             } else {
　　
               throw new IOException("Too many retries because of encryption" +
　　
                   " zone operations", e);
　　
             }
　　
           } else {
　　
             throw e;
　　
           }
　　
         }
　　
       }
　　
       Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");
　　
      //new输出流对象
　　
       final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat,
　　
           flag, progress, checksum, favoredNodes);
　　
       out.start();//调用内部类DataStreamer的start()方法，DataStreamer继承Thread，所以说这是一个线程，从NameNode中申请新的block信息；
　　同时前面我们介绍hdfs原理的时候提到的流水线作业（Pipeline）也是在这里实现，有兴趣的同学可以去研究下，这里就不带大家看了
　　
       return out;
　　
     } finally {
　　
       scope.close();
　　
     }
　　
   }