在分布式文件系统源码里写数据流程是我一直建议先啃的一块骨头。它不像读流程那样有清晰的“请求-应答”边界而是把客户端、名称节点、数据节点三个角色用一条管道绑在一起中间还夹着租约、校验、重试、流水线复制。HDFS里最核心的协议细节几乎都藏在这一条链路中。系列前面几篇已经把目录树、租约、块报告、读流程拆得差不多了这篇轮到写路径本身。如果你是从第一篇一路追过来的对NameNode的元数据模型应该已经有底如果你只是中途进来也没关系我会把写路径上会用到的类名和流程重新点一遍当作独立文章读也行。这篇主要解决三个问题一次write调用从客户端进入后数据到底在哪个阶段被转成packet数据节点之间是怎么接力复制的写失败了以后系统靠什么机制恢复。整个过程涉及DFSClient、DFSOutputStream、DataStreamer、BlockReceiver、PipelineAck这几个核心角色理解了它们后面再去看append、truncate那些高段位流程会轻松很多。1. 写数据流程全景一次write调用背后的五个阶段1.1 从write到落盘五个阶段各干了什么写数据不是“客户端发一个包、数据节点存一个包”这么简单。我习惯把它拆成五个阶段理解每个阶段的主角完全不同这也对应了源码里几个核心类的边界。创建文件阶段客户端调用create()通过RPC通知名称节点“我要建一个文件”。名称节点做权限检查、目录检查、配额检查在目录树里插入一个INode同时给客户端建立租约。此阶段不分配实际数据块只是“占了个名字”。建立管道阶段客户端攒了第一批数据后DataStreamer线程去名称节点申请块拿到块ID和一组数据节点位置然后与第一个数据节点建立TCP连接。数据节点之间再接力建立第二跳、第三跳连接整条流水线才算通。数据发送阶段客户端把内存中的packet按顺序写到第一个数据节点数据节点收到后再转发给第二个第二个转发给第三个。每个packet里有用户数据块和对应的校验块。ack确认阶段每个数据节点写完一个packet后要向上游返回确认。最后一个数据节点先回给第二个第二个回给第一个第一个把状态汇总后回给客户端。客户端收到ack才能把该packet从待确认队列里移除。关闭提交阶段最后一个块写完客户端通知名称节点“写完了”名称节点把最后一个块标记为完成更新文件大小、释放租约整个写操作对外可见。这五个阶段不是严格串行的。管道建立时前一个块的数据可能还在传输块与块之间也是“申请新块 - 建立新管道 - 继续传”的轮换。理解这一点再看DataStreamer的循环逻辑就不会觉得绕。1.2 关键类与线程谁负责哪块活写数据路径上的类不多但每个都身兼数职。我把它们按“客户端侧”和“数据节点侧”分了个角色表。角色所在进程核心职责关键成员DFSClient客户端对外API入口维护配置、租约续期RPC代理、create/write接口DFSOutputStream客户端写缓冲、chunk切分、packet构造currentPacket、dataQueue、ackQueueDataStreamer客户端后台发送线程负责申请块、建管道、发数据blockStream、statePacketResponder客户端数据节点处理ack匹配序号更新确认状态seqno、ackQueueDataNode数据节点接收请求调度数据读写DataXceiver线程池BlockReceiver数据节点接收packet、写本地、转发下游mirrorOutStreamFSNamesystem名称节点块分配、租约维护、文件元数据变更pendingCreates我第一次看这套代码时最容易搞混的是DataStreamer和PacketResponder。其实分工很清楚DataStreamer管“发”PacketResponder管“收确认”。它们俩通过ackQueue这个共享队列协作一个从队尾加包一个从队首确认。类似快递员送货和前端签收确认的关系一个负责跑路一个负责对账。2. 客户端侧核心DFSOutputStream的缓冲、packet构造与入队2.1 writeChunk与packet的边界为什么是512字节和64KB客户端侧最底层的写入口是DFSOutputStream.write()但它实际继承自FSOutputSummer。这个父类做的事情非常朴素把用户的数据先攒到一块内存buf里每攒够一个chunk就计算一次CRC32校验和然后调用writeChunk()把数据和校验交给子类。HDFS默认dfs.bytes-per-checksum512也就是每512字节用户数据对应一个4字节CRC32校验值。为什么是512这是磁盘扇区的整数倍同时CRC计算代价适中。你把它改成4096也不是不行但校验粒度变粗块恢复时要重传的单元会变大。DFSOutputStream.writeChunk()干的活可以用下面这段简化逻辑描述protected synchronized void writeChunk(byte[] b, int offset, int len, byte[] checksum, int ckoff, int cklen) { // 如果当前没有正在构造的packet就新建一个 if (currentPacket null) { currentPacket createPacket(); } // 把chunk数据追加到packet的数据区 currentPacket.writeData(b, offset, len); // 把CRC校验值追加到packet的校验区 currentPacket.writeChecksum(checksum, ckoff, cklen); // 当前packet的chunk计数加一 currentPacket.setNumChunks(currentPacket.getNumChunks() 1); // 达到packet容量上限就固化并放入发送队列 if (currentPacket.getNumChunks() maxChunks) { enqueueCurrentPacket(); } }这里有个关键数字一个packet的默认容量约64KB但实际可承载的用户数据大约只有63KB出头。因为一个chunk带数据512字节加校验4字节总共516字节再加上header占位大概127个chunk就顶到64KB的门槛了。所以maxChunks并不是写死的而是在创建输出流时根据dfs.client-write-packet-size、chunk大小和header长度动态算出来的。我建议你阅读源码时特意关注enqueueCurrentPacket()它先把currentPacket固化设置包序号、偏移、最后包标记然后塞进dataQueue同时把currentPacket置空等待下一次writeChunk重新创建。这里就是“用户写线程”和“DataStreamer发送线程”的交汇点也是背压发生的地方。2.2 两个队列的配合dataQueue与ackQueue为什么必须分开DFSOutputStream内部维护了两个LinkedBlockingQueueDFSPacketdataQueue装着已经构造好、但还没发出去的packet。ackQueue装着已经发出去、还没收到确认的packet。为什么要拆两个队列你可以把这两个队列想象成流水线上的“待加工区”和“待质检区”。DataStreamer只从待加工区取货PacketResponder只负责把质检合格的产品移到已发货区中间任何一个packet没确认都不能丢掉。更重要的原因是背压控制。HDFS客户端不能无限制地往网络里灌数据否则数据节点内存会爆。DataStreamer在发送每个packet前会检查ackQueue的长度如果待确认的包太多它会调用waitForSpaceAndFlushPacket()把自己挂起。用户写线程继续构造packet往dataQueue里塞但DataStreamer不消费dataQueue就会越积越多最终导致内存压力。这时候HDFS就通过带缓冲的阻塞队列让数据节点反馈压力传导回用户线程。这个设计虽然简单但从工程角度看恰到好处它不需要复杂的流控协议只靠“发送前等ack队列收缩”就完成了客户端到数据节点之间的速率匹配。我曾经在测试环境把dfs.client.write.max-packets-in-flight调成很小再去跑高并发写入任务立刻看到吞吐暴跌因为每个包都要等确认才能继续发管道利用率被拉满了。这就是背压最直观的效果。2.3 DataStreamer主循环从建管道到发数据的运转节奏DataStreamer虽然名字叫“流”本质是一个常驻的Daemon线程。它不单单发数据还负责申请新块、建立管道、处理错误恢复。核心运行逻辑看简化伪代码就很清楚public void run() { while (!shouldStop()) { // 1. 如果有待确认的包先处理ack if (!ackQueue.isEmpty() !response.isPending()) { waitForAck(); } // 2. 判断当前块是否需要收尾 if (shouldEndBlock()) { endBlock(); } // 3. 队列为空就歇一会 if (dataQueue.isEmpty()) { continue; } // 4. 管道还没建立先去申请块并建管道 if (blockStream null) { establishBlock(); } // 5. 从队列取出一个包发送 DFSPacket packet dataQueue.peek(); packet.writeTo(blockStream); blockStream.flush(); // 6. 包已发出挪到待确认队列 dataQueue.removeFirst(); ackQueue.addLast(packet); } }这里最容易误解的地方是第4步管道是懒加载的。客户端调用create()时并没有和任何数据节点建立连接只有DataStreamer第一次准备发数据时才发现blockStream null然后触发establishBlock()。这个过程的细节放到下一节。第5步的peek()是有讲究的。它取出但不移除只有成功写入网络流后才从dataQueue移除。如果writeTo抛异常这个包还留在队首错误恢复后重发就不会丢数据。这是写数据路径上最容易踩坑的地方调试时发现复现了重复包其实是正常现象HDFS靠seqno去重客户端并不会重发一条命令让数据节点二次存储。3. 管道建立与块分配NameNode的选址细节3.1 create阶段到底做了什么很多人初读源码以为create()就该返回块的位置信息。实际上新文件在create()阶段只拿到文件元数据块信息是空的。名称节点在这个阶段做三件事检查路径是否合法、父目录是否存在、是否超过配额。在目录树中创建INodeFile初始大小是0。为该客户端建立租约记录到租约管理器的pendingCreates表里。租约是写数据流程的“通行证”。名称节点靠它防止多个客户端同时写同一个文件也靠它判断“这个客户端是不是已经挂了”。如果客户端进程崩溃租约得不到续期名称节点最终会回收文件并执行恢复流程。不要小看配额检查这一步。生产环境里因为共享目录quota写满导致QuotaExceededException的比例非常高而且这个错误往往发生在写入刚开始的时候因为名称节点创建文件时就会检查目录配额。如果你在客户端日志里频繁看到这个异常先查目录配额不要傻傻地查磁盘空间。3.2 块放置策略与机架感知真正给文件分配数据块发生在DataStreamer触发addBlock()之后。名称节点返回的LocatedBlock里会带一组DatanodeInfo这组节点的选择不是随机的而是走块放置策略。默认策略是每个副本考虑机架第一个副本选择客户端本地或负载较低的数据节点第二个副本放在不同机架第三个副本放在与第二个同机架的不同节点。目标很简单既要保证机架级容错又要尽量缩小跨机架流量。更关键的是客户端收到这组位置后的处理。DataStreamer在建立管道前会按网络距离排序优先选离自己最近的数据节点作为管道头。这里的“距离”指的是机架感知里的拓扑距离同节点 同机架 同数据中心 随意。排序结果基本决定了数据从哪里进入管道也就决定了客户端到第一个数据节点的延迟。我不知道你是否遇到过“写入慢但数据节点负载都不高”的情况。如果客户端机架配置没有正确上报名称节点会把它当作“位置未知”节点块放置策略可能把第一个副本分配到离得很远的数据节点跨机架带宽就成了瓶颈。检查方法很简单看客户端日志里addBlock返回的位置列表再对比物理机架一眼就能发现问题。3.3 locatedBlock与管道建立时机有一个很容易混淆的点我把客户端拿到的LocatedBlock和最终使用的管道节点分成两件事locatedBlock.getLocations()是名称节点给的可选节点集合。客户端真正往第一个节点发数据时才和那个节点建立TCP连接然后由第一个节点连接第二个、第二个连接第三个形成真正的写管道。也就是说管道是在数据发送前由DataStreamer建立的不是名称节点直接下发一个管道列表。名称节点只负责给出候选数据节点管道的组装工作在客户端。这个设计带来的好处是管道建立失败时客户端可以自己剔除坏节点重新组装不需要名称节点介入。坏节点名单通过excludedNodes参数传给下一次addBlock()这样申请下一个块时会自动绕开。4. 数据包格式、流水线复制与ack机制4.1 DFSPacket的内存布局与CRC32进入网络传输阶段数据的单位变成DFSPacket。HDFS协议里一个packet大致是这个布局---------------------------------------------------------- | Header约24字节 | 数据区 | 校验区 | | 包长度/偏移/序号/标志 | chunk[0..n-1] | crc[0..n-1] | ----------------------------------------------------------Header里的offsetInBlock表示这个包在块内部的偏移seqno是包的序号从0开始递增。数据节点恢复管道时这些字段用于对齐“我已经写了多少数据”。每个chunk的CRC32是客户端算好的随着数据一起发送。数据节点每收到一个包会对每个chunk重新计算校验值并比对不一致就抛ChecksumException。这条校验链是整个HDFS数据可靠性最底层的防线。有一个细节值得单独拿出来说客户端发送packet时是把数据和校验合在同一个writeTo调用里写出去的而不是先发数据再发校验。这样数据节点可以一次性读取完整包减少系统调用次数。如果你自己实现过传输协议这个细节可能深有体会分两次写会让接收端多等待一次对高吞吐场景是致命的。4.2 BlockReceiver边存储边转发的接力赛数据节点收到opWriteBlock请求后会创建BlockReceiver专门处理这个块的写入流量。它的主循环可以简化成以下几步从网络流读取packet header。读取数据区和校验区到内存缓冲。校验每个chunk的CRC。把数据写入磁盘或文件缓存。如果管道下游还有数据节点把同一份packet写到mirrorOutStream。等下游ack返回后向上游发送本节点的ack。注意第5步的“同一份packet”。中间数据节点转发的是原始数据包不是自己重新组装的包。这样保证包内容整条管道完全一致而每个节点只需要关心自己本地的磁盘写入是否成功。我最初看这段代码时有个疑问为什么不先转发给下游再写本地磁盘这样不是能减少延迟吗后来想明白了如果先转发但本地写失败下游已经存了一份数据自己这一份却可能损坏节点间块内容对不上后续副本校验会一直报警。HDFS宁可让写本地和转发串行也要保证每个节点存的是同一份完整数据。4.3 PacketResponderack如何逐级合并ack机制是写数据流程的“信用凭证”。每个数据节点上都有一个PacketResponder线程它盯着BlockReceiver的接收进度。当本节点写完一个packet并且下游也返回了对应的ack它才向上游回复。客户端侧的PacketResponder则负责从管道头读取ack然后按照seqno把对应packet从ackQueue移除。如果某个packet迟迟等不到ack客户端会认为管道出问题了进入错误恢复流程。这里有一个我在调试时抓包才真正明白的地方客户端并不是收到三个节点的ack它只收到管道头返回的一个ack。这个ack里携带了所有副本节点的状态吗也携带但客户端处理时通常只关心它是否成功。管道内部各级节点自己维护“我这一跳是否收到下游确认”的状态确保ack能一级级合并回传。因此你在客户端能看到的现象是发送一个包等待一个ack然后继续发下一个。管道内部的二级、三级确认对客户端完全透明。这种设计把复杂的多副本确认逻辑收拢到了管道内部客户端代码不需要区分副本数量。5. 写失败与管道恢复数据不丢的关键防线5.1 错误恢复的两段式处理写数据过程中数据节点宕机、网络闪断、磁盘满都可能导致管道中断。客户端检测到写异常后会由DataStreamer执行错误恢复。恢复流程分两个阶段第一个阶段是“清理现场”。客户端先等ackQueue里已发出的包全部确认完毕或者根据已确认的偏移量确定当前块的有效长度。这些信息会被缓存下来作为恢复后的写入起点。这个阶段要特别注意不能直接把已发送但未确认的包丢掉因为包可能已经写入了一部分数据节点。第二个阶段是“重建管道”。客户端剔除故障数据节点把剩余的可用节点列表提交给名称节点调用updatePipeline让名称节点更新这个块的副本位置。然后客户端重新和剩余的第一个数据节点建立连接继续从断点发送数据。整个过程想得简单一点就是确认哪些节点活着、哪些包已经安全落地、剩下的从哪个偏移继续写。三个问题回答清楚恢复流程自然就清楚了。我有次在一个三副本集群里手动杀掉了中间数据节点的进程观察客户端日志发现恢复几乎是瞬间完成只丢了一个包的重传量。这说明两段式恢复在这个场景下开销很低。5.2 租约、追加写与文件恢复的关系上面说的都是运行中失败。如果客户端整个进程崩溃了比如写数据写一半突然断电谁来恢复这个文件答案是租约机制加追加写流程。名称节点持有每个正在写文件的租约租约到期前客户端必须续约。如果到期还没续名称节点就知道这个客户端已经不可靠了它会把租约标记为可恢复。后续其他客户端尝试写同一个文件时会触发“租约恢复”名称节点要确保该文件的最后一个块所有副本达成一致长度再允许继续追加。这个机制解释了为什么HDFS写数据时要求客户端不能随便“不打招呼就走”。正常情况下写完文件必须调用complete()或close()告诉名称节点“我写完了”此时租约被释放。如果客户端只创建文件不关闭租约就一直在那里挂着直到超时。我之前排查过一个问题某离线任务每天启动时创建文件但异常退出时没调close第二天重跑就报LeaseExpiredException。原因是租约还没过期名称节点不允许第二个客户端覆盖同一个文件。这种问题一般不是代码bug而是业务异常处理没做好。写HDFS的代码close一定要放在finally块里。5.3 常见异常速查表异常典型原因排查方向LeaseExpiredException写进程长时间未续约或异常退出检查客户端是否还活着、租约是否被其他客户端接管QuotaExceededException文件目录配额不足检查目录配额、清理历史文件ChecksumException数据传输损坏或校验不一致检查网卡、磁盘、数据节点日志SocketTimeoutException数据节点ack超时检查网络抖动、数据节点负载、socket超时参数PipelineTimeoutException管道中某节点无响应定位故障节点查看其日志和磁盘状态BlockMissingException数据节点宕机导致副本数不足检查副本管理、机架感知配置排查这类异常有一个心法不要一上来就看代码先看日志时间线。客户端日志、数据节点日志、名称节点审计日志按时间对齐基本能定位出是哪一跳断了。6. 源码阅读与调试心得6.1 用日志把写数据链路拉通源码读得再多不如自己实际复现一次写入过程。我最推荐的调试方法是用log4j把关键类打成DEBUG级别再写一个几千行的写入程序。log4j.logger.org.apache.hadoop.hdfs.DFSClientDEBUG log4j.logger.org.apache.hadoop.hdfs.DFSOutputStreamDEBUG log4j.logger.org.apache.hadoop.hdfs.DataStreamerDEBUG log4j.logger.org.apache.hadoop.hdfs.DFSPacketTRACE打开这些日志后你会在客户端日志里看到这样的信息Creating block [BP-xxx:blk_xxx] Sending packet: seqno0, offsetInBlock0 Ack received: seqno0, statusSUCCESS Sending packet: seqno1, offsetInBlock65536 ...这个序列本身就是一次完美的写数据流程教程。每次Sending packet对应一次packet.writeTo()每次Ack received对应PacketResponder匹配seqno成功。如果在某个seqno只看到发送、看不到ack那里就是故障发生的地方。数据节点侧打开org.apache.hadoop.hdfs.server.datanode.BlockReceiver的DEBUG后你会看到它接收每个包的偏移和写磁盘的状态。把两边的日志放到同一时间轴上整条管道的数据流动就立体了。6.2 影响写入吞吐的关键参数写数据流程的性能受下面几个参数影响最大我把它们列成一张表方便你排查吞吐问题时逐一确认。参数默认值影响dfs.blocksize128MB块越大名称节点元数据越少DataStreamer跨块切换越少dfs.replication3副本数决定管道长度直接影响写延迟dfs.bytes-per-checksum512校验粒度影响计算开销和恢复粒度dfs.client-write-packet-size64KB每包大小影响网络传输效率和队列深度dfs.client.write.max-packets-in-flight80未确认包上限决定背压强度dfs.client.socket-timeout60000ms读写socket超时误伤高延迟管道时会导致重启如果写入吞吐上不去优先看dfs.client.write.max-packets-in-flight和dfs.client-write-packet-size。很多人喜欢无限调大packets-in-flight但调太大反而会因为某个包阻塞导致整体延迟变高。合适的做法是结合网络RTT测试RTT高时适当调大RTT低时保持默认或调小。顺带提醒一句dfs.blocksize调大不能动现有文件只对新写文件生效。生产环境想优化小文件场景不如从调整块大小和归档策略入手而不是盲目改复制因子。6.3 我踩过的几个坑先说校验和。有次我在一个跨交换机集群上跑写入测试频繁出现ChecksumException一开始怀疑是网卡问题后来发现是交换机MTU不一致导致部分大包被拆碎后校验失败。排查这种问题抓包比看代码快得多。再说机架感知。名称节点默认把未知机架的数据节点统统归到一个默认机架这会让块放置策略失效三副本可能全落在同一个物理机架上。表面看写入没问题一旦那个机架断电数据全部不可用。所以集群里每个数据节点的网络位置必须配置准确这个检查是上线前的必经步骤。最后是一个比较隐蔽的坑客户端在调用write后没有及时flush导致最后一个块迟迟不触发DataStreamer发送看起来像是写入“卡住”了。其实数据还在客户端缓冲区里躺着。HDFS里的hflush和hsync是给这类场景准备的理解写数据流程后你会明白为什么close之前必须保证队列里所有包都被ack掉。读到这里的最后一点建议坦白说读写数据流程这段源码我在不同版本里翻来覆去看了好几遍最大的体会是HDFS的设计核心非常朴素——用队列解耦生产与消费用ack保证不丢用管道复制换取吞吐用失败剔除换取可用性。真正难的不是某个算法而是这些约束在同一个流程里同时起作用的交织感。如果你想验证自己是不是真看懂了我建议做一个实验在500MB文件写入过程中手动把管道中的第二个数据节点进程停掉观察客户端日志里管道重建的过程。能完整看到“ack超时 - 剔除节点 - 更新管道 - 从断点继续发”这条链路说明你已经把写数据流程串起来了。后面如果大家有兴趣我再把append写流程和truncate恢复流程也按这个粒度拆一遍。
在分布式文件系统源码里写数据流程是我一直建议先啃的一块骨头。它不像读流程那样有清晰的“请求-应答”边界而是把客户端、名称节点、数据节点三个角色用一条管道绑在一起中间还夹着租约、校验、重试、流水线复制。HDFS里最核心的协议细节几乎都藏在这一条链路中。系列前面几篇已经把目录树、租约、块报告、读流程拆得差不多了这篇轮到写路径本身。如果你是从第一篇一路追过来的对NameNode的元数据模型应该已经有底如果你只是中途进来也没关系我会把写路径上会用到的类名和流程重新点一遍当作独立文章读也行。这篇主要解决三个问题一次write调用从客户端进入后数据到底在哪个阶段被转成packet数据节点之间是怎么接力复制的写失败了以后系统靠什么机制恢复。整个过程涉及DFSClient、DFSOutputStream、DataStreamer、BlockReceiver、PipelineAck这几个核心角色理解了它们后面再去看append、truncate那些高段位流程会轻松很多。1. 写数据流程全景一次write调用背后的五个阶段1.1 从write到落盘五个阶段各干了什么写数据不是“客户端发一个包、数据节点存一个包”这么简单。我习惯把它拆成五个阶段理解每个阶段的主角完全不同这也对应了源码里几个核心类的边界。创建文件阶段客户端调用create()通过RPC通知名称节点“我要建一个文件”。名称节点做权限检查、目录检查、配额检查在目录树里插入一个INode同时给客户端建立租约。此阶段不分配实际数据块只是“占了个名字”。建立管道阶段客户端攒了第一批数据后DataStreamer线程去名称节点申请块拿到块ID和一组数据节点位置然后与第一个数据节点建立TCP连接。数据节点之间再接力建立第二跳、第三跳连接整条流水线才算通。数据发送阶段客户端把内存中的packet按顺序写到第一个数据节点数据节点收到后再转发给第二个第二个转发给第三个。每个packet里有用户数据块和对应的校验块。ack确认阶段每个数据节点写完一个packet后要向上游返回确认。最后一个数据节点先回给第二个第二个回给第一个第一个把状态汇总后回给客户端。客户端收到ack才能把该packet从待确认队列里移除。关闭提交阶段最后一个块写完客户端通知名称节点“写完了”名称节点把最后一个块标记为完成更新文件大小、释放租约整个写操作对外可见。这五个阶段不是严格串行的。管道建立时前一个块的数据可能还在传输块与块之间也是“申请新块 - 建立新管道 - 继续传”的轮换。理解这一点再看DataStreamer的循环逻辑就不会觉得绕。1.2 关键类与线程谁负责哪块活写数据路径上的类不多但每个都身兼数职。我把它们按“客户端侧”和“数据节点侧”分了个角色表。角色所在进程核心职责关键成员DFSClient客户端对外API入口维护配置、租约续期RPC代理、create/write接口DFSOutputStream客户端写缓冲、chunk切分、packet构造currentPacket、dataQueue、ackQueueDataStreamer客户端后台发送线程负责申请块、建管道、发数据blockStream、statePacketResponder客户端数据节点处理ack匹配序号更新确认状态seqno、ackQueueDataNode数据节点接收请求调度数据读写DataXceiver线程池BlockReceiver数据节点接收packet、写本地、转发下游mirrorOutStreamFSNamesystem名称节点块分配、租约维护、文件元数据变更pendingCreates我第一次看这套代码时最容易搞混的是DataStreamer和PacketResponder。其实分工很清楚DataStreamer管“发”PacketResponder管“收确认”。它们俩通过ackQueue这个共享队列协作一个从队尾加包一个从队首确认。类似快递员送货和前端签收确认的关系一个负责跑路一个负责对账。2. 客户端侧核心DFSOutputStream的缓冲、packet构造与入队2.1 writeChunk与packet的边界为什么是512字节和64KB客户端侧最底层的写入口是DFSOutputStream.write()但它实际继承自FSOutputSummer。这个父类做的事情非常朴素把用户的数据先攒到一块内存buf里每攒够一个chunk就计算一次CRC32校验和然后调用writeChunk()把数据和校验交给子类。HDFS默认dfs.bytes-per-checksum512也就是每512字节用户数据对应一个4字节CRC32校验值。为什么是512这是磁盘扇区的整数倍同时CRC计算代价适中。你把它改成4096也不是不行但校验粒度变粗块恢复时要重传的单元会变大。DFSOutputStream.writeChunk()干的活可以用下面这段简化逻辑描述protected synchronized void writeChunk(byte[] b, int offset, int len, byte[] checksum, int ckoff, int cklen) { // 如果当前没有正在构造的packet就新建一个 if (currentPacket null) { currentPacket createPacket(); } // 把chunk数据追加到packet的数据区 currentPacket.writeData(b, offset, len); // 把CRC校验值追加到packet的校验区 currentPacket.writeChecksum(checksum, ckoff, cklen); // 当前packet的chunk计数加一 currentPacket.setNumChunks(currentPacket.getNumChunks() 1); // 达到packet容量上限就固化并放入发送队列 if (currentPacket.getNumChunks() maxChunks) { enqueueCurrentPacket(); } }这里有个关键数字一个packet的默认容量约64KB但实际可承载的用户数据大约只有63KB出头。因为一个chunk带数据512字节加校验4字节总共516字节再加上header占位大概127个chunk就顶到64KB的门槛了。所以maxChunks并不是写死的而是在创建输出流时根据dfs.client-write-packet-size、chunk大小和header长度动态算出来的。我建议你阅读源码时特意关注enqueueCurrentPacket()它先把currentPacket固化设置包序号、偏移、最后包标记然后塞进dataQueue同时把currentPacket置空等待下一次writeChunk重新创建。这里就是“用户写线程”和“DataStreamer发送线程”的交汇点也是背压发生的地方。2.2 两个队列的配合dataQueue与ackQueue为什么必须分开DFSOutputStream内部维护了两个LinkedBlockingQueueDFSPacketdataQueue装着已经构造好、但还没发出去的packet。ackQueue装着已经发出去、还没收到确认的packet。为什么要拆两个队列你可以把这两个队列想象成流水线上的“待加工区”和“待质检区”。DataStreamer只从待加工区取货PacketResponder只负责把质检合格的产品移到已发货区中间任何一个packet没确认都不能丢掉。更重要的原因是背压控制。HDFS客户端不能无限制地往网络里灌数据否则数据节点内存会爆。DataStreamer在发送每个packet前会检查ackQueue的长度如果待确认的包太多它会调用waitForSpaceAndFlushPacket()把自己挂起。用户写线程继续构造packet往dataQueue里塞但DataStreamer不消费dataQueue就会越积越多最终导致内存压力。这时候HDFS就通过带缓冲的阻塞队列让数据节点反馈压力传导回用户线程。这个设计虽然简单但从工程角度看恰到好处它不需要复杂的流控协议只靠“发送前等ack队列收缩”就完成了客户端到数据节点之间的速率匹配。我曾经在测试环境把dfs.client.write.max-packets-in-flight调成很小再去跑高并发写入任务立刻看到吞吐暴跌因为每个包都要等确认才能继续发管道利用率被拉满了。这就是背压最直观的效果。2.3 DataStreamer主循环从建管道到发数据的运转节奏DataStreamer虽然名字叫“流”本质是一个常驻的Daemon线程。它不单单发数据还负责申请新块、建立管道、处理错误恢复。核心运行逻辑看简化伪代码就很清楚public void run() { while (!shouldStop()) { // 1. 如果有待确认的包先处理ack if (!ackQueue.isEmpty() !response.isPending()) { waitForAck(); } // 2. 判断当前块是否需要收尾 if (shouldEndBlock()) { endBlock(); } // 3. 队列为空就歇一会 if (dataQueue.isEmpty()) { continue; } // 4. 管道还没建立先去申请块并建管道 if (blockStream null) { establishBlock(); } // 5. 从队列取出一个包发送 DFSPacket packet dataQueue.peek(); packet.writeTo(blockStream); blockStream.flush(); // 6. 包已发出挪到待确认队列 dataQueue.removeFirst(); ackQueue.addLast(packet); } }这里最容易误解的地方是第4步管道是懒加载的。客户端调用create()时并没有和任何数据节点建立连接只有DataStreamer第一次准备发数据时才发现blockStream null然后触发establishBlock()。这个过程的细节放到下一节。第5步的peek()是有讲究的。它取出但不移除只有成功写入网络流后才从dataQueue移除。如果writeTo抛异常这个包还留在队首错误恢复后重发就不会丢数据。这是写数据路径上最容易踩坑的地方调试时发现复现了重复包其实是正常现象HDFS靠seqno去重客户端并不会重发一条命令让数据节点二次存储。3. 管道建立与块分配NameNode的选址细节3.1 create阶段到底做了什么很多人初读源码以为create()就该返回块的位置信息。实际上新文件在create()阶段只拿到文件元数据块信息是空的。名称节点在这个阶段做三件事检查路径是否合法、父目录是否存在、是否超过配额。在目录树中创建INodeFile初始大小是0。为该客户端建立租约记录到租约管理器的pendingCreates表里。租约是写数据流程的“通行证”。名称节点靠它防止多个客户端同时写同一个文件也靠它判断“这个客户端是不是已经挂了”。如果客户端进程崩溃租约得不到续期名称节点最终会回收文件并执行恢复流程。不要小看配额检查这一步。生产环境里因为共享目录quota写满导致QuotaExceededException的比例非常高而且这个错误往往发生在写入刚开始的时候因为名称节点创建文件时就会检查目录配额。如果你在客户端日志里频繁看到这个异常先查目录配额不要傻傻地查磁盘空间。3.2 块放置策略与机架感知真正给文件分配数据块发生在DataStreamer触发addBlock()之后。名称节点返回的LocatedBlock里会带一组DatanodeInfo这组节点的选择不是随机的而是走块放置策略。默认策略是每个副本考虑机架第一个副本选择客户端本地或负载较低的数据节点第二个副本放在不同机架第三个副本放在与第二个同机架的不同节点。目标很简单既要保证机架级容错又要尽量缩小跨机架流量。更关键的是客户端收到这组位置后的处理。DataStreamer在建立管道前会按网络距离排序优先选离自己最近的数据节点作为管道头。这里的“距离”指的是机架感知里的拓扑距离同节点 同机架 同数据中心 随意。排序结果基本决定了数据从哪里进入管道也就决定了客户端到第一个数据节点的延迟。我不知道你是否遇到过“写入慢但数据节点负载都不高”的情况。如果客户端机架配置没有正确上报名称节点会把它当作“位置未知”节点块放置策略可能把第一个副本分配到离得很远的数据节点跨机架带宽就成了瓶颈。检查方法很简单看客户端日志里addBlock返回的位置列表再对比物理机架一眼就能发现问题。3.3 locatedBlock与管道建立时机有一个很容易混淆的点我把客户端拿到的LocatedBlock和最终使用的管道节点分成两件事locatedBlock.getLocations()是名称节点给的可选节点集合。客户端真正往第一个节点发数据时才和那个节点建立TCP连接然后由第一个节点连接第二个、第二个连接第三个形成真正的写管道。也就是说管道是在数据发送前由DataStreamer建立的不是名称节点直接下发一个管道列表。名称节点只负责给出候选数据节点管道的组装工作在客户端。这个设计带来的好处是管道建立失败时客户端可以自己剔除坏节点重新组装不需要名称节点介入。坏节点名单通过excludedNodes参数传给下一次addBlock()这样申请下一个块时会自动绕开。4. 数据包格式、流水线复制与ack机制4.1 DFSPacket的内存布局与CRC32进入网络传输阶段数据的单位变成DFSPacket。HDFS协议里一个packet大致是这个布局---------------------------------------------------------- | Header约24字节 | 数据区 | 校验区 | | 包长度/偏移/序号/标志 | chunk[0..n-1] | crc[0..n-1] | ----------------------------------------------------------Header里的offsetInBlock表示这个包在块内部的偏移seqno是包的序号从0开始递增。数据节点恢复管道时这些字段用于对齐“我已经写了多少数据”。每个chunk的CRC32是客户端算好的随着数据一起发送。数据节点每收到一个包会对每个chunk重新计算校验值并比对不一致就抛ChecksumException。这条校验链是整个HDFS数据可靠性最底层的防线。有一个细节值得单独拿出来说客户端发送packet时是把数据和校验合在同一个writeTo调用里写出去的而不是先发数据再发校验。这样数据节点可以一次性读取完整包减少系统调用次数。如果你自己实现过传输协议这个细节可能深有体会分两次写会让接收端多等待一次对高吞吐场景是致命的。4.2 BlockReceiver边存储边转发的接力赛数据节点收到opWriteBlock请求后会创建BlockReceiver专门处理这个块的写入流量。它的主循环可以简化成以下几步从网络流读取packet header。读取数据区和校验区到内存缓冲。校验每个chunk的CRC。把数据写入磁盘或文件缓存。如果管道下游还有数据节点把同一份packet写到mirrorOutStream。等下游ack返回后向上游发送本节点的ack。注意第5步的“同一份packet”。中间数据节点转发的是原始数据包不是自己重新组装的包。这样保证包内容整条管道完全一致而每个节点只需要关心自己本地的磁盘写入是否成功。我最初看这段代码时有个疑问为什么不先转发给下游再写本地磁盘这样不是能减少延迟吗后来想明白了如果先转发但本地写失败下游已经存了一份数据自己这一份却可能损坏节点间块内容对不上后续副本校验会一直报警。HDFS宁可让写本地和转发串行也要保证每个节点存的是同一份完整数据。4.3 PacketResponderack如何逐级合并ack机制是写数据流程的“信用凭证”。每个数据节点上都有一个PacketResponder线程它盯着BlockReceiver的接收进度。当本节点写完一个packet并且下游也返回了对应的ack它才向上游回复。客户端侧的PacketResponder则负责从管道头读取ack然后按照seqno把对应packet从ackQueue移除。如果某个packet迟迟等不到ack客户端会认为管道出问题了进入错误恢复流程。这里有一个我在调试时抓包才真正明白的地方客户端并不是收到三个节点的ack它只收到管道头返回的一个ack。这个ack里携带了所有副本节点的状态吗也携带但客户端处理时通常只关心它是否成功。管道内部各级节点自己维护“我这一跳是否收到下游确认”的状态确保ack能一级级合并回传。因此你在客户端能看到的现象是发送一个包等待一个ack然后继续发下一个。管道内部的二级、三级确认对客户端完全透明。这种设计把复杂的多副本确认逻辑收拢到了管道内部客户端代码不需要区分副本数量。5. 写失败与管道恢复数据不丢的关键防线5.1 错误恢复的两段式处理写数据过程中数据节点宕机、网络闪断、磁盘满都可能导致管道中断。客户端检测到写异常后会由DataStreamer执行错误恢复。恢复流程分两个阶段第一个阶段是“清理现场”。客户端先等ackQueue里已发出的包全部确认完毕或者根据已确认的偏移量确定当前块的有效长度。这些信息会被缓存下来作为恢复后的写入起点。这个阶段要特别注意不能直接把已发送但未确认的包丢掉因为包可能已经写入了一部分数据节点。第二个阶段是“重建管道”。客户端剔除故障数据节点把剩余的可用节点列表提交给名称节点调用updatePipeline让名称节点更新这个块的副本位置。然后客户端重新和剩余的第一个数据节点建立连接继续从断点发送数据。整个过程想得简单一点就是确认哪些节点活着、哪些包已经安全落地、剩下的从哪个偏移继续写。三个问题回答清楚恢复流程自然就清楚了。我有次在一个三副本集群里手动杀掉了中间数据节点的进程观察客户端日志发现恢复几乎是瞬间完成只丢了一个包的重传量。这说明两段式恢复在这个场景下开销很低。5.2 租约、追加写与文件恢复的关系上面说的都是运行中失败。如果客户端整个进程崩溃了比如写数据写一半突然断电谁来恢复这个文件答案是租约机制加追加写流程。名称节点持有每个正在写文件的租约租约到期前客户端必须续约。如果到期还没续名称节点就知道这个客户端已经不可靠了它会把租约标记为可恢复。后续其他客户端尝试写同一个文件时会触发“租约恢复”名称节点要确保该文件的最后一个块所有副本达成一致长度再允许继续追加。这个机制解释了为什么HDFS写数据时要求客户端不能随便“不打招呼就走”。正常情况下写完文件必须调用complete()或close()告诉名称节点“我写完了”此时租约被释放。如果客户端只创建文件不关闭租约就一直在那里挂着直到超时。我之前排查过一个问题某离线任务每天启动时创建文件但异常退出时没调close第二天重跑就报LeaseExpiredException。原因是租约还没过期名称节点不允许第二个客户端覆盖同一个文件。这种问题一般不是代码bug而是业务异常处理没做好。写HDFS的代码close一定要放在finally块里。5.3 常见异常速查表异常典型原因排查方向LeaseExpiredException写进程长时间未续约或异常退出检查客户端是否还活着、租约是否被其他客户端接管QuotaExceededException文件目录配额不足检查目录配额、清理历史文件ChecksumException数据传输损坏或校验不一致检查网卡、磁盘、数据节点日志SocketTimeoutException数据节点ack超时检查网络抖动、数据节点负载、socket超时参数PipelineTimeoutException管道中某节点无响应定位故障节点查看其日志和磁盘状态BlockMissingException数据节点宕机导致副本数不足检查副本管理、机架感知配置排查这类异常有一个心法不要一上来就看代码先看日志时间线。客户端日志、数据节点日志、名称节点审计日志按时间对齐基本能定位出是哪一跳断了。6. 源码阅读与调试心得6.1 用日志把写数据链路拉通源码读得再多不如自己实际复现一次写入过程。我最推荐的调试方法是用log4j把关键类打成DEBUG级别再写一个几千行的写入程序。log4j.logger.org.apache.hadoop.hdfs.DFSClientDEBUG log4j.logger.org.apache.hadoop.hdfs.DFSOutputStreamDEBUG log4j.logger.org.apache.hadoop.hdfs.DataStreamerDEBUG log4j.logger.org.apache.hadoop.hdfs.DFSPacketTRACE打开这些日志后你会在客户端日志里看到这样的信息Creating block [BP-xxx:blk_xxx] Sending packet: seqno0, offsetInBlock0 Ack received: seqno0, statusSUCCESS Sending packet: seqno1, offsetInBlock65536 ...这个序列本身就是一次完美的写数据流程教程。每次Sending packet对应一次packet.writeTo()每次Ack received对应PacketResponder匹配seqno成功。如果在某个seqno只看到发送、看不到ack那里就是故障发生的地方。数据节点侧打开org.apache.hadoop.hdfs.server.datanode.BlockReceiver的DEBUG后你会看到它接收每个包的偏移和写磁盘的状态。把两边的日志放到同一时间轴上整条管道的数据流动就立体了。6.2 影响写入吞吐的关键参数写数据流程的性能受下面几个参数影响最大我把它们列成一张表方便你排查吞吐问题时逐一确认。参数默认值影响dfs.blocksize128MB块越大名称节点元数据越少DataStreamer跨块切换越少dfs.replication3副本数决定管道长度直接影响写延迟dfs.bytes-per-checksum512校验粒度影响计算开销和恢复粒度dfs.client-write-packet-size64KB每包大小影响网络传输效率和队列深度dfs.client.write.max-packets-in-flight80未确认包上限决定背压强度dfs.client.socket-timeout60000ms读写socket超时误伤高延迟管道时会导致重启如果写入吞吐上不去优先看dfs.client.write.max-packets-in-flight和dfs.client-write-packet-size。很多人喜欢无限调大packets-in-flight但调太大反而会因为某个包阻塞导致整体延迟变高。合适的做法是结合网络RTT测试RTT高时适当调大RTT低时保持默认或调小。顺带提醒一句dfs.blocksize调大不能动现有文件只对新写文件生效。生产环境想优化小文件场景不如从调整块大小和归档策略入手而不是盲目改复制因子。6.3 我踩过的几个坑先说校验和。有次我在一个跨交换机集群上跑写入测试频繁出现ChecksumException一开始怀疑是网卡问题后来发现是交换机MTU不一致导致部分大包被拆碎后校验失败。排查这种问题抓包比看代码快得多。再说机架感知。名称节点默认把未知机架的数据节点统统归到一个默认机架这会让块放置策略失效三副本可能全落在同一个物理机架上。表面看写入没问题一旦那个机架断电数据全部不可用。所以集群里每个数据节点的网络位置必须配置准确这个检查是上线前的必经步骤。最后是一个比较隐蔽的坑客户端在调用write后没有及时flush导致最后一个块迟迟不触发DataStreamer发送看起来像是写入“卡住”了。其实数据还在客户端缓冲区里躺着。HDFS里的hflush和hsync是给这类场景准备的理解写数据流程后你会明白为什么close之前必须保证队列里所有包都被ack掉。读到这里的最后一点建议坦白说读写数据流程这段源码我在不同版本里翻来覆去看了好几遍最大的体会是HDFS的设计核心非常朴素——用队列解耦生产与消费用ack保证不丢用管道复制换取吞吐用失败剔除换取可用性。真正难的不是某个算法而是这些约束在同一个流程里同时起作用的交织感。如果你想验证自己是不是真看懂了我建议做一个实验在500MB文件写入过程中手动把管道中的第二个数据节点进程停掉观察客户端日志里管道重建的过程。能完整看到“ack超时 - 剔除节点 - 更新管道 - 从断点继续发”这条链路说明你已经把写数据流程串起来了。后面如果大家有兴趣我再把append写流程和truncate恢复流程也按这个粒度拆一遍。
