Hadoop3.2.1 【 HDFS 】源码分析 : DataXceiver 解析 [一]

左手的ㄟ右手 2022-12-23 01:55 188阅读 0赞

# 一. 前言 #

DataXceiverServer主要用于监听并接收流式请求， 然后创建并启动DataXceiver对象。 DataXceiver是Receiver的子类， DataTransferProtocol真正的响应操作都是在DataXceiver类中实现的。

# 二. run()方法 #

DataXceiver的执行逻辑主要是在run()方法中完成的。 DataXceiver.run()方法首先根据传入的Peer对象获取此层网络的输入／ 输出流， 并对输入／ 输出流进行装饰， 然后调用父类Receiver的initialize()方法执行初始化操作。

接下来run()方法会循环读取IO流中的流式接口请求， 并调用对应的方法响应请求。这里注意， 对于操作符Op的解析是通过调用父类Receiver.readOp()方法完成的， 对于操作符的处理是通过调用父类Receiver.processOp()方法完成的， processOp()方法会根据Op调用  
DataXceiver对应的处理方法.

/** * Read/write data from/to the DataXceiverServer. */
      @Override
      public void run() { 
        int opsProcessed = 0;
        Op op = null;
    
        try { 
          synchronized(this) { 
            xceiver = Thread.currentThread();
          }
          dataXceiverServer.addPeer(peer, Thread.currentThread(), this);
          peer.setWriteTimeout(datanode.getDnConf().socketWriteTimeout);
          //获取底层的输入流
          InputStream input = socketIn;
          try { 
            IOStreamPair saslStreams = datanode.saslServer.receive(peer, socketOut,
              socketIn, datanode.getXferAddress().getPort(),
              datanode.getDatanodeId());
    
            // 对输入流进行装饰
            input = new BufferedInputStream(saslStreams.in, smallBufferSize);
            // 获取底层的输出流
            socketOut = saslStreams.out;
          } catch (InvalidMagicNumberException imne) { 
            if (imne.isHandshake4Encryption()) { 
              LOG.info("Failed to read expected encryption handshake from client " +
                  "at {}. Perhaps the client " +
                  "is running an older version of Hadoop which does not support " +
                  "encryption", peer.getRemoteAddressString(), imne);
            } else { 
              LOG.info("Failed to read expected SASL data transfer protection " +
                  "handshake from client at {}" +
                  ". Perhaps the client is running an older version of Hadoop " +
                  "which does not support SASL data transfer protection",
                  peer.getRemoteAddressString(), imne);
            }
            return;
          }
          //调用父类的initialize ()方法完成Receiver的初始化操作
          super.initialize(new DataInputStream(input));
          
          // We process requests in a loop, and stay around for a short timeout.
          // This optimistic behaviour allows the other end to reuse connections.
          // Setting keepalive timeout to 0 disable this behavior.
          do { 
            updateCurrentThreadName("Waiting for operation #" + (opsProcessed + 1));
    
            try { 
              if (opsProcessed != 0) { 
                assert dnConf.socketKeepaliveTimeout > 0;
                peer.setReadTimeout(dnConf.socketKeepaliveTimeout);
              } else { 
                peer.setReadTimeout(dnConf.socketTimeout);
              }
              //调用Receiver.readOp()从输入流中解析操作符
              op = readOp();
            } catch (InterruptedIOException ignored) { 
              // Time out while we wait for client rpc
              break;
            } catch (EOFException | ClosedChannelException e) { 
              // Since we optimistically expect the next op, it's quite normal to
              // get EOF here.
              LOG.debug("Cached {} closing after {} ops. " +
                  "This message is usually benign.", peer, opsProcessed);
              break;
            } catch (IOException err) { 
              incrDatanodeNetworkErrors();
              throw err;
            }
    
            // restore normal timeout
            if (opsProcessed != 0) { 
              peer.setReadTimeout(dnConf.socketTimeout);
            }
    
            opStartTime = monotonicNow();
    
            //调用processOp()处理这个流式请求
            processOp(op);
            ++opsProcessed;
          } while ((peer != null) &&
              (!peer.isClosed() && dnConf.socketKeepaliveTimeout > 0));
        } catch (Throwable t) { 
          String s = datanode.getDisplayName() + ":DataXceiver error processing "
              + ((op == null) ? "unknown" : op.name()) + " operation "
              + " src: " + remoteAddress + " dst: " + localAddress;
          if (op == Op.WRITE_BLOCK && t instanceof ReplicaAlreadyExistsException) { 
            // For WRITE_BLOCK, it is okay if the replica already exists since
            // client and replication may write the same block to the same datanode
            // at the same time.
            if (LOG.isTraceEnabled()) { 
              LOG.trace(s, t);
            } else { 
              LOG.info("{}; {}", s, t.toString());
            }
          } else if (op == Op.READ_BLOCK && t instanceof SocketTimeoutException) { 
            String s1 =
                "Likely the client has stopped reading, disconnecting it";
            s1 += " (" + s + ")";
            if (LOG.isTraceEnabled()) { 
              LOG.trace(s1, t);
            } else { 
              LOG.info("{}; {}", s1, t.toString());
            }
          } else if (t instanceof InvalidToken ||
              t.getCause() instanceof InvalidToken) { 
            // The InvalidToken exception has already been logged in
            // checkAccess() method and this is not a server error.
            LOG.trace(s, t);
          } else { 
            LOG.error(s, t);
          }
        } finally { 
          collectThreadLocalStates();
          LOG.debug("{}:Number of active connections is: {}",
              datanode.getDisplayName(), datanode.getXceiverCount());
          updateCurrentThreadName("Cleaning up");
          if (peer != null) { 
            dataXceiverServer.closePeer(peer);
            IOUtils.closeStream(in);
          }
        }
      }

--------------------

参考:  
Hadoop 2.X HDFS源码剖析 – 徐鹏