Hadoop3.2.1 【 HDFS 】源码分析 : DataXceiverServer 解析

港控/mmm° 2022-12-23 00:37 250阅读 0赞

# 一. 前言 #

DataXceiverServer对象用于在Datanode上监听流式接口的请求， 每当有Client通过Sender类发起流式接口请求时， DataXceiverServer就会监听并接收这个请求， 然后创建一个DataXceiver对象用于响应这个请求并执行对应的操作。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

# 二.DataXceiverServer的初始化 #

在DataNode的初始化代码中， 会创建一个DataXceiverServer对象监听所有流式接口请求， Datanode会调用Datanode.initDataXceiver()方法来完成DataXceiverServer对象的构造。  
initDataXceiver()方法会首先创建TcpPeerServer对象（对ServerSocket的封装） ， 它能通过accept()方法返回Peer对象（封装了Socket对象， 提供通信的输入／ 输出流） 。  
TcpPeerServer的监听地址是通过dfs.datanode.address配置项配置的， TcpPeerServer会将Socket设置为No\_Delay的方式， 然后设置receiveBuffer为128KB(默认为8KB） ， 同时将DataXceiverServer放入线程组dataXceiverServer中， 最后将这个线程组设置为守护线程。

注： （1） ThreadGroup表示一个线程的集合， 可以通过线程组的函数对一组线程进行操作， 例如setDeamon()。  
（2） 守护线程： 如果虚拟机中只有守护线程在运行， 则虚拟机退出； 如果存在一个用户线程， 则守护线程仍然执行。

initDataXceiver()方法的第二部分代码则是与短路操作相关的， 如果用户通过设置dfs.client.read.shortcircuit或者dfs.client.domain.socket.data.traffic配置项开启了短路操作\[默认不开启\]， 那么DataNode会创建一个localDataXceiverServer响应本地的短路读取请求。 这里要特别注意， localDataXceiverServer底层的PeerServer为DomainPeerServer——也就是通信时并不使用基于TCP连接的传统的TcpPeerServer（底层封装Socket） ， 而是使用DomainPeerServer（底层基于DomainSocket） 。

> 注： 在Linux中， 有种技术叫作UNIX Domain Socket。 UNIX Domain Socket是一种进程间的通信方式， 它使得同一台机器上的两个进程能以Socket的方式通信。 它带来的另一大好处是， 利用它， 两个进程除了可以传递普通数据外， 还可以在进程间传递文件描述符。 通过这种技术， 就可以实现DataNode与DFSClient的本地读取功能。

private void initDataXceiver() throws IOException { 
        // find free port or use privileged port provided
        TcpPeerServer tcpPeerServer;
        if (secureResources != null) { 
          tcpPeerServer = new TcpPeerServer(secureResources);
        } else { 
          // 设置TCP接收缓冲区 ipc.server.listen.queue.size 128
          int backlogLength = getConf().getInt(
              CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_KEY,
              CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_DEFAULT);
          // TcpPeerServer(/0.0.0.0:9866)
          tcpPeerServer = new TcpPeerServer(dnConf.socketWriteTimeout,  DataNode.getStreamingAddr(getConf()), backlogLength);
    
        }
    
        if (dnConf.getTransferSocketRecvBufferSize() > 0) { 
    
          tcpPeerServer.setReceiveBufferSize(  dnConf.getTransferSocketRecvBufferSize());
        }
        // /0.0.0.0:9866
        streamingAddr = tcpPeerServer.getStreamingAddr();
        LOG.info("Opened streaming server at {}", streamingAddr);
        // 构造DataXceiverServer对象 并设置分组为: dataXceiverServer
        this.threadGroup = new ThreadGroup("dataXceiverServer");
        xserver = new DataXceiverServer(tcpPeerServer, getConf(), this);
        //将DataXceiverServer线程组设置为守护线程
        this.dataXceiverServer = new Daemon(threadGroup, xserver);
        this.threadGroup.setDaemon(true); // auto destroy when empty
    
        //短路读取情况
        //dfs.client.domain.socket.data.traffic : false
        if (getConf().getBoolean(
            HdfsClientConfigKeys.Read.ShortCircuit.KEY,
            HdfsClientConfigKeys.Read.ShortCircuit.DEFAULT) ||
            getConf().getBoolean(
                HdfsClientConfigKeys.DFS_CLIENT_DOMAIN_SOCKET_DATA_TRAFFIC,
                HdfsClientConfigKeys
                  .DFS_CLIENT_DOMAIN_SOCKET_DATA_TRAFFIC_DEFAULT)) { 
          //构造DomainPeerServer (底层基于DomainSocket,用于本进程间通信）
          DomainPeerServer domainPeerServer =  getDomainPeerServer(getConf(), streamingAddr.getPort());
    
          if (domainPeerServer != null) { 
            //构造localDataXceiverServer
            this.localDataXceiverServer = new Daemon(threadGroup,
                new DataXceiverServer(domainPeerServer, getConf(), this));
            LOG.info("Listening on UNIX domain socket: {}",
                domainPeerServer.getBindPath());
          }
        }
        //创建ShortCircuitRegistry对象
        this.shortCircuitRegistry = new ShortCircuitRegistry(getConf());
      }

# 三. run()方法 #

DataXceiverServer的功能都是在run()方法中实现的。 DataXceiverServer.run()方法的逻辑非常简单， 它会循环调用peerServer的accept()方法监听， 如果有新的连接请求则创建Peer对象， 并构造一个DataXceiver线程服务这个流式请求——也就是DataXceiverServer只负责连接的建立以及构造并启动DataXceiver， 流式接口请求则是由DataXceiver响应的，真正的操作都是由DataXceiver来进行的。

@Override
      public void run() { 
        Peer peer = null;
    
        //循环执行逻辑
        while (datanode.shouldRun && !datanode.shutdownForUpgrade) { 
          try { 
            //接受连接请求
            peer = peerServer.accept();
    
            // Make sure the xceiver count is not exceeded
            // 通过 ThreadGroup 来控制的, 默认值为 4096
            int curXceiverCount = datanode.getXceiverCount();
            if (curXceiverCount > maxXceiverCount) { 
              throw new IOException("Xceiver count " + curXceiverCount
                  + " exceeds the limit of concurrent xcievers: "
                  + maxXceiverCount);
            }
    
            //创建一个DataXceiver响应请求
            new Daemon(datanode.threadGroup,
                DataXceiver.create(peer, datanode, this))
                .start();
          } catch (SocketTimeoutException ignored) { 
            //Socket超时异常直接忽略
            // wake up to see if should continue to run
          } catch (AsynchronousCloseException ace) { 
            // another thread closed our listener socket - that's expected during shutdown,
            // but not in other circumstances
            if (datanode.shouldRun && !datanode.shutdownForUpgrade) { 
              LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ace);
            }
          } catch (IOException ie) { 
            IOUtils.cleanup(null, peer);
            LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ie);
          } catch (OutOfMemoryError ie) { 
            IOUtils.cleanup(null, peer);
            // DataNode can run out of memory if there is too many transfers.
            // Log the event, Sleep for 30 seconds, other transfers may complete by
            // then.
            LOG.error("DataNode is out of memory. Will retry in 30 seconds.", ie);
            try { 
              Thread.sleep(30 * 1000);
            } catch (InterruptedException e) { 
              // ignore
            }
          } catch (Throwable te) { 
            LOG.error(datanode.getDisplayName() + ":DataXceiverServer: Exiting due to: ", te);
            //其他异常则直接关闭Datanode
            datanode.shouldRun = false;
          }
        }
    
        // 清理操作： 退出主循环， 执行关闭操作， 将peerServer关闭
        // Close the server to stop reception of more requests.
        try { 
          peerServer.close();
          closed = true;
        } catch (IOException ie) { 
          LOG.warn(datanode.getDisplayName()
              + " :DataXceiverServer: close exception", ie);
        }
    
        // if in restart prep stage, notify peers before closing them.
        if (datanode.shutdownForUpgrade) { 
          restartNotifyPeers();
          // Each thread needs some time to process it. If a thread needs
          // to send an OOB message to the client, but blocked on network for
          // long time, we need to force its termination.
          LOG.info("Shutting down DataXceiverServer before restart");
          // Allow roughly up to 2 seconds.
          for (int i = 0; getNumPeers() > 0 && i < 10; i++) { 
            try { 
              Thread.sleep(200);
            } catch (InterruptedException e) { 
              // ignore
            }
          }
        }
        // Close all peers.
        closeAllPeers();
      }

AsynchronousCloseException， 当通过DataXceiverServer.kill()关闭服务器时， 会触发这种异常。  
这是因为DataXceiverServer常规的关闭方法是将dataNode.shouldRun字段设置为false， 这样run()方法通过循环时的检查就可以正常退出。 但是当dataNode.shouldRun字段已经被设置为false， 而DataXceiverServer却在accept()中阻塞时， 就需要调用kill()方法强行ServerSocket关闭。 这种情况就会抛出AsynchronousCloseException异常， 这时就需要强行结束DataXceiverServer的服务了。

void kill() { 
        assert (datanode.shouldRun == false || datanode.shutdownForUpgrade) :
          "shoudRun should be set to false or restarting should be true"
          + " before killing";
        try { 
          this.peerServer.close();
          this.closed = true;
        } catch (IOException ie) { 
          LOG.warn(datanode.getDisplayName() + ":DataXceiverServer.kill(): ", ie);
        }
      }

当DataXceiverServer退出主循环后， 就会进行DataXceiverServer的关闭清理操作， 这种设计模式在Hadoop RPC框架的实现中也使用了。 在线程的run()方法中携带了清理程序， 通过标志位来进行流程控制， 退出循环之后在run()方法中进行程序的清理.

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/12fbc33ec10f4f4caeadf22271500a4a.png