本文共 6030 字，大约阅读时间需要 20 分钟。

DataXceviverServer:

监听块传输连接请求，同时控制进行的块传输请求数(同一时刻的传输数不能超过maxXceiverCount)和带宽耗费情况(块传输时带宽耗费带宽不能超过预定值BlockTransferThrottler.bytesPerPeriod)。当在run里监听到一个块传输请求时，开启一个DataXceiver线程处理块传输。系统关闭时，会关闭用于监听的连接的ServerSocket同时将DataXceiver所产生的线程关闭，使得DataXceiver因为出现错误而退出。

BlockTransferThrottler:

带宽节流器，用于协调块运输所耗费的带宽。一个块传输过程中，如果已经使用了超过预期的带宽就令其等待wait一段时间，使得不会因为某个块传输而带宽耗尽。

DataXceiver继承自runnable，用于实现块的发送和接收。在run里先进行版本的校验，再判断是否当前同时传输块的连接数超过了dataXceiverServer.maxXceiverCount值，如果是操作失败。然后从客户端读取要进行的操作op：

OP_WRITE_BLOCK (80)：写数据块 

OP_READ_BLOCK (81)：读数据块 

OP_READ_METADATA (82)：读数据块元文件 

OP_REPLACE_BLOCK (83)：替换一个数据块 

OP_COPY_BLOCK (84)：拷贝一个数据块 

OP_BLOCK_CHECKSUM (85)：读数据块检验码 

根据这个操作调用相应的方法进行处理。处理后关闭流，关闭socket。

=============================================================

读操作DataXceiver.readBlock：

依次接收blockId，genStamp块时间戳，startOffset块偏移位置，length要读取的块大小，clientName客户端请求者的名字(非空串表示这是客户端的写块操作，否则为replaceBlock中的主动读取块操作)。

根据以上的信息建立一个BlockReader对象，向请求者发送OP_STATUS_SUCCESS表示操作状态，然后调用BlockReader.sendBlock将块发送给请求者。更新datanode.myMetrics.bytesRead、datanode.myMetrics.blocksRead。最后关闭输出流和blockReader。

BlockReader的构造函数中由于设计到校验和，所以比较复杂。大致流程是获取元数据(校验和)文件的输入流，进行元数据版本的检验，获取校验和checksum，计算块文件和校验和文件开始读取的位置，打开块数据流。

BlockReader.sendBlock先发送checksum.header给请求者，然后将块文件划分成最多maxChunksPerPacket个数据包进行发送，每个数据包大小为pktSize，然后多次调研BlockReader.sendChunks(pktBuf,maxChunksPerPacket,streamForSendChunks)将数据包发送出去。发送数据包结束后发送一个0表示块的发送完成。然后BlockReader.close()关闭各种资源。在发送包的过程中会统计已发送的流量。

BlockReader.sendChunks(ByteBuffer pkt, int maxChunks, OutputStream out)中先计算一个数据包能发送的最多个chunks，然后将packetLen数据包长度、offset数据包开始位置在块中的偏移、seqno数据包的编号、是不是最后一个数据包写到pkt，然后将块元数据文件中的校验数据读出到pkt中。计算出块数据在pkt中的位置(存放在校验数据之后)。然后根据BlockReader.blockInPosition变量值判断是否可以用FileChannel将数据发给块请求者。如果不可以，就将块数据读取到buf中，然后利用checksum对读出来的数据生成校验和，并与先前读出的块元数据文件进行校验，校验成功后将buf发送给请求者。如果可以，就先将buf中的校验和部分发送出去，然后再利用fileChannel将块文件中的数据发送出去。最后调用BlockTransferThrottler.throttle进行节流控制。

【注：BlockReader.blockInPosition是用于判断是否可以打开一个fileChannel，如果可以就根据这个fileChannel.transferTo方法将数据包发送到客户端。】

=============================================================

写操作DataXceiver.writeBlock：

写数据块要涉及到对数据块进行备份，即要将块写到多个Datanodes。会将这些Datanodes组织成一个pipeline。

client发送块给pipeline上的第一个datanode，此datanode将块数据写到本地并传给下一个。然后将后续块返回来的响应信息加上自身的操作结果信息一起返回到前面。在pipeline上，如果某个DataNode有后续节点，那么，它必须等到后续节点的成功应答，才可以发送应答到它前面的节点。

DataXceiver.writeBlock会先创建三个数据流mirrorOut、mirrorIn、replyOut，到下一个DN的socket，以及用于接收块和写块的BlockReceiver。

DataOutputStream mirrorOut  // stream to next target

DataInputStream mirrorIn    // reply from next target

DataOutputStream replyOut   // stream to prev target

Socket mirrorSock           // socket to next target

BlockReceiver blockReceiver // responsible for data handling

然后调用 blockReceiver.receiveBlock进行块的操作。操作完后关闭流和socket。

BlockReceiver.receiveBlock会先start一个PacketResponder线程，然后一直调用receivePacket直到读完该块数据为止。然后往下一个节点写入0表示块写入完成。等待到packets的所有响应都发送完毕。调用FSDataset进行finalizeBlock。最后关闭PacketResponder线程。

BlockReceiver.receivePacket调用BlockReceiver.readNextPacket接收一个packet(最终是调用readToBuf将packet读到buf中)。然后将packet写到下一个节点(在将数据都写给下一个节点后，写入一个0表示块结束)。一个packet格式如下图：

只有pipeline最后一个节点或发送端是namenode时才需要对发过来的数据进行校验verifyChunks。接着会把块数据和块元数据文件写入到tmp目录下。本地写入完成后向responder.ackQueue入队一个响应，以便向上一个DN或client响应块操作。

在pipeline上的DataNode都有个BlockReceiver.PacketResponder线程，用于向上一个datanode或客户端发送响应消息以及心跳保活。

pipeline上的心跳保活机制是由后往前进行的:最后一个DataNode D 发给上一个DataNode C，C接收到发现是一个心跳消息，就像B发送一个心跳...

PacketResponder.run从下一个DN中读取一个PipelineAck，得到packet号seqno(可以是心跳标识、错误标志、packet标识)。如果是心跳就将这个ack回送到上一个DN。如果是packet号，就先从ackQueue读出一个Packet，验证这个Packet.seqno是否与之前读到的ack.seqno相等(表示是同一个块)，如果这个数据包是块中最后一个，就用notifyNamenodeReceivedBlock向namenode发送通知【所有改发DataNode的操作，需要把信息更新刡NameNode上】，同时调用finalizeBlock将块和元数据文件移到到current下。然后从之前读出的ack.replies加上当前的操作状态OP_STATUS_SUCCESS，包装成一个PipelineAck,写回到上一个DN或client。

上一个DN或client发送过来的数据包操作成功后，通过将Packet(seqno,lastPacketInBlock)加入到PacketResponder.ackQueue中，PacketResponder从ackQueue中取出Packet，得知要对pipeline的写数据包的情况向上一个DN或client通报。根据从下一个DN得到的后续DN的写数据包响应消息，向上一个DN或client发送一个PipelineAck。

当datanode接收到的是最后一个packet时，在PacketResponder.run内调用finalizeBlock将Block移到current下，并调用notifyNamenodeReceivedBlock(block,DataNode.EMPTY_DEL_HINT);向namenode通知块已写完。 

===============================================================

读元数据文件操作DataXceiver.readMetadata：

先根据请求者传过来的blockId、genStamp找到相应的块信息，然后获得块元数据文件输入流，将元数据读取到buf中，建立一个到请求者的输出流，依次写入OP_STATUS_SUCCESS、元数据文件大小、元数据。最后关闭流。

=============================================================

复制块文件操作DataXceiver.copyBlock：

调用BlockSender.sendBlock将块发送出去。

BlockSender构造函数：

BlockSender(Block block, long startOffset, long length,boolean corruptChecksumOk, boolean chunkOffsetOK,boolean verifyChecksum, DataNode datanode);

readBlock中：

new BlockSender(block, startOffset, length,true, true, false, datanode, clientTraceFmt);

copyBlock中：

new BlockSender(block, 0, -1, false, false, false,datanode);

读块是从块指定位置开始读特定长度，而copyBlock是整个块进行复制。

=============================================================

块文件的替换操作DataXceiver.replaceBlock：

替换块只发生在指定的一个DataNode上，而writeBlock是在一个pipeline上。

替换块的流程大致是：得到替换块请求者的信息，向请求者发送复制块请求OP_COPY_BLOCK，创建并调用BlockReceiver.receiveBlock用于从请求者那读取块，然后向namenode发送一个块变更通知，更新namenode上的块信息。

格式对比：

BlockReceiver(Block block, DataInputStream in, String inAddr,String myAddr, boolean isRecovery, String clientName,DatanodeInfo srcDataNode, DataNode datanode);

replaceBlock中的创建BlockReceiver的参数：

new BlockReceiver(block, proxyReply, proxySock.getRemoteSocketAddress().toString(),proxySock.getLocalSocketAddress().toString(),false, "", null, datanode);

writeBlock中的创建BlockReceiver的参数：

new BlockReceiver(block, in,s.getRemoteSocketAddress().toString(),s.getLocalSocketAddress().toString(),isRecovery, client, srcDataNode, datanode);

proxyReply和in 不一样，返是因为发起请求的节点和提供数据的节点并不是同一个。写数据块发起请求方也提供数据，

替换数据块请求方不提供数据，而是提供了一个数据源（proxySource参数），由replaceBlock发起一个拷贝数据块的请求，

建立数据源。对亍拷贝数据块操作，isRecovery=false，client=""， srcDataNode=null。client=""表示这不是客户端的写块。

块被替换后，要将被替换掉的块从namenode中删除，这是通过datanode.notifyNamenodeReceivedBlock(block, sourceID)实现的。

=============================================================

读取块校验操作DataXceiver.getBlockChecksum：

流程：根据块元数据文件中的数据进行利用MD5算法计算校验和，然后将校验和发送出去。

转载地址：https://blog.csdn.net/u013494310/article/details/19476891 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！