/** * Call * {@link #create(String, FsPermission, EnumSet, boolean, short, long, Progressable, int, ChecksumOpt)} * with <code>createParent</code> set to true. */public DFSOutputStream create(String src, FsPermission permission, EnumSet<CreateFlag> flag, short replication, long blockSize, Progressable progress, int buffersize, ChecksumOpt checksumOpt) throws IOException { //似曾相识create return create(src, permission, flag, true, replication, blockSize, progress, buffersize, checksumOpt, null);}
public DFSOutputStream create(String src, FsPermission permission, EnumSet<CreateFlag> flag, boolean createParent, short replication, long blockSize, Progressable progress, int buffersize, ChecksumOpt checksumOpt, InetSocketAddress[] favoredNodes) throws IOException { checkOpen(); if (permission == null) { permission = FsPermission.getFileDefault(); } FsPermission masked = permission.applyUMask(dfsClientConf.uMask); if (LOG.isDebugEnabled()) { LOG.debug(src + ": masked=" + masked); }   //这里主要做了三个动作，newStreamForCreate这个方法 // 1、往文件目录树INodeDirectory添加了INodeFile // 2、添加文件契约lease // 3、启动DataStreamer,负责与NameNodeRPCServer通信 final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this, src, masked, flag, createParent, replication, blockSize, progress, buffersize, dfsClientConf.createChecksum(checksumOpt), getFavoredNodesStr(favoredNodes));
 //开启契约 beginFileLease(result.getFileId(), result); return result;}
/** Get a lease and start automatic renewal *///获取一个文件契约并启动自动续约private void beginFileLease(final long inodeId, final DFSOutputStream out) throws IOException { getLeaseRenewer().put(inodeId, out, this);}
//这里使用了synchronized关键字，保证同一个时间只有一个客户端能写数据，其他的阻塞synchronized void put(final long inodeId, final DFSOutputStream out, final DFSClient dfsc) { if (dfsc.isClientRunning()) { if (!isRunning() || isRenewerExpired()) { //start a new deamon with a new id. final int id = ++currentId; //创建了一个后台线程 daemon = new Daemon(new Runnable() { @Override public void run() { try { if (LOG.isDebugEnabled()) { LOG.debug("Lease renewer daemon for " + clientsString() + " with renew id " + id + " started"); } //LeaseRenewer----负责管理续约 LeaseRenewer.this.run(id); } catch(InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug(LeaseRenewer.this.getClass().getSimpleName() + " is interrupted.", e); } } finally { synchronized(LeaseRenewer.this) { Factory.INSTANCE.remove(LeaseRenewer.this); } if (LOG.isDebugEnabled()) { LOG.debug("Lease renewer daemon for " + clientsString() + " with renew id " + id + " exited"); } } }  @Override public String toString() { return String.valueOf(LeaseRenewer.this); } }); //启动后台守护线程 daemon.start(); } dfsc.putFileBeingWritten(inodeId, out); emptyTime = Long.MAX_VALUE; }}

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;
 // Retry the create if we get a RetryStartFileException up to a maximum // number of times boolean shouldRetry = true; int retryCount = CREATE_RETRY_COUNT;  //不断重试，确保文件目录创建成功 while (shouldRetry) { shouldRetry = false; try { //hdfs创建文件，可以详细看NameNodeRPC端的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!");
 //初始化了DataStreamer，DataStreamer是写数据流程的核心类，实质上DataStreamer是一个线程 final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat, flag, progress, checksum, favoredNodes); //启动了DataStreamer out.start(); return out; } finally { scope.close(); }}

/** Construct a new output stream for creating a file. */private DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat, EnumSet<CreateFlag> flag, Progressable progress, DataChecksum checksum, String[] favoredNodes) throws IOException { this(dfsClient, src, progress, stat, checksum); this.shouldSyncBlock = flag.contains(CreateFlag.SYNC_BLOCK); /** * Directory -> File -> Block(128M) -> packet（64k）-> chunk（516byte） */ //计算数据单元的值，比较简单，不展开，点进去看就知道了 computePacketChunkSize(dfsClient.getConf().writePacketSize, bytesPerChecksum);  //创建了DataStreamer，这才是核心所在 streamer = new DataStreamer(stat, null); if (favoredNodes != null && favoredNodes.length != 0) { streamer.setFavoredNodes(favoredNodes); }}

private DataStreamer(HdfsFileStatus stat, ExtendedBlock block) { isAppend = false; isLazyPersistFile = isLazyPersist(stat); this.block = block; stage = BlockConstructionStage.PIPELINE_SETUP_CREATE;}

/** * Initialize for data streaming */private void initDataStreaming() { this.setName("DataStreamer for file " + src + " block " + block);  response = new ResponseProcessor(nodes); //启动线程，看到这里是不是跟DataStreamer很相似 response.start(); stage = BlockConstructionStage.DATA_STREAMING;}
 ResponseProcessor(DatanodeInfo[] targets) { this.targets = targets; } @Override public void run() { setName("ResponseProcessor for block " + block); //ack响应队列，返回DataNode写数据的结果 PipelineAck ack = new PipelineAck(); TraceScope scope = NullScope.INSTANCE; while (!responderClosed && dfsClient.clientRunning && !isLastPacketInBlock) { // process responses from datanodes. try { // read an ack from the pipeline long begin = Time.monotonicNow(); //读取下游的处理结果 ack.readFields(blockReplyStream); long duration = Time.monotonicNow() - begin; if (duration > dfsclientSlowLogThresholdMs && ack.getSeqno() != DFSPacket.HEART_BEAT_SEQNO) { DFSClient.LOG.warn("Slow ReadProcessor read fields took " + duration + "ms (threshold=" + dfsclientSlowLogThresholdMs + "ms); ack: " + ack + ", targets: " + Arrays.asList(targets)); } else if (DFSClient.LOG.isDebugEnabled()) { DFSClient.LOG.debug("DFSClient " + ack); } long seqno = ack.getSeqno(); // processes response status from datanodes. for (int i = ack.getNumOfReplies() - 1; i >= 0 && dfsClient.clientRunning; i--) { final Status reply = PipelineAck.getStatusFromHeader(ack.getHeaderFlag(i)); // Restart will not be treated differently unless it is // the local node or the only one in the pipeline. if (PipelineAck.isRestartOOBStatus(reply) && shouldWaitForRestart(i)) { restartDeadline = dfsClient.getConf().datanodeRestartTimeout + Time.monotonicNow(); setRestartingNodeIndex(i); String message = "A datanode is restarting: " + targets[i]; DFSClient.LOG.info(message); throw new IOException(message); } // node error if (reply != SUCCESS) { setErrorIndex(i); // first bad datanode throw new IOException("Bad response " + reply + " for block " + block + " from datanode " + targets[i]); } } assert seqno != PipelineAck.UNKOWN_SEQNO : "Ack for unknown seqno should be a failed ack: " + ack; if (seqno == DFSPacket.HEART_BEAT_SEQNO) { // a heartbeat ack continue; } // a success ack for a data packet DFSPacket one; synchronized (dataQueue) { one = ackQueue.getFirst(); } if (one.getSeqno() != seqno) { throw new IOException("ResponseProcessor: Expecting seqno " + " for block " + block + one.getSeqno() + " but received " + seqno); } isLastPacketInBlock = one.isLastPacketInBlock(); // Fail the packet write for testing in order to force a // pipeline recovery. if (DFSClientFaultInjector.get().failPacket() && isLastPacketInBlock) { failPacket = true; throw new IOException("Failing the last packet for testing."); } // update bytesAcked block.setNumBytes(one.getLastByteOffsetBlock()); synchronized (dataQueue) { scope = Trace.continueSpan(one.getTraceSpan()); one.setTraceSpan(null); lastAckedSeqno = seqno; //如果ack发送成功那么就会把ackQueue里面packet移除来 ackQueue.removeFirst(); dataQueue.notifyAll(); one.releaseBuffer(byteArrayManager); } } catch (Exception e) { if (!responderClosed) { if (e instanceof IOException) { setLastException((IOException) e); } hasError = true; // If no explicit error report was received, mark the primary // node as failed. tryMarkPrimaryDatanodeFailed(); synchronized (dataQueue) { dataQueue.notifyAll(); } if (restartingNodeIndex.get() == -1) { DFSClient.LOG.warn( "DFSOutputStream ResponseProcessor exception " + " for block " + block, e); } responderClosed = true; } } finally { scope.close(); } } }  void close() { responderClosed = true; //打断当前的线程，让线程快速退出 this.interrupt(); }}

/** * Wraps the stream in a CryptoOutputStream if the underlying file is encrypted. */ //返回的是HDFSOutputStreampublic HdfsDataOutputStream createWrappedOutputStream(DFSOutputStream dfsos, FileSystem.Statistics statistics) throws IOException { return createWrappedOutputStream(dfsos, statistics, 0);}