Hadoop-2.7.0 HA实现细节分析之HealthMonitor

竺绍辉
2023-12-01

一、HealthMonitor是什么

      HealthMonitor是一个周期性工作的后台线程,它在一个循环中周期性的同HA服务进行心跳,负责跟踪NameNode服务的健康状况,并在健康状况变化时调用failover控制器的回调方法。

二、HealthMonitor是如何实现的

      1、成员变量

            HealthMonitor内部有如下主要成员变量:

            1)后台工作线程:真正卖力干活的

  private Daemon daemon;// 后台工作线程
            2)一些时间间隔、rpc超时参数等配置信息,见注释

  // 重连、周期性检查、失连后的睡眠时间等时间间隔
  private long connectRetryInterval;
  private long checkIntervalMillis;
  private long sleepAfterDisconnectMillis;

  // rpc超时时间
  private int rpcTimeout;
    // 配置信息
  private final Configuration conf;

            3)连接代理相关

  /** The connected proxy */
  // 连接代理
  private HAServiceProtocol proxy;

  /** The HA service to monitor */
  // HA服务到监视器的对象,通过它来获取连接代理proxy,代表一个客户端HA管理命令的目标。
  private final HAServiceTarget targetToMonitor;

            4)监视器s,状态发生变化时需要周知监视器s,故它们都实现了相同的接口

  /**
   * Listeners for state changes
   * 状态变更监视器列表,里面的元素是实现Callback的对象
   */
  private List<Callback> callbacks = Collections.synchronizedList(
      new LinkedList<Callback>());

  /**
   * 服务状态监视器列表,里面的元素是实现ServiceStateCallback的对象
   */
  private List<ServiceStateCallback> serviceStateCallbacks = Collections
      .synchronizedList(new LinkedList<ServiceStateCallback>());

            5)状态

  // 服务状态,默认为正在初始化
  private State state = State.INITIALIZING;
  /**
   * 最后一次服务状态,默认为正在初始化
   */
  private HAServiceStatus lastServiceState = new HAServiceStatus(
      HAServiceState.INITIALIZING);

      2、状态&接口

            2.1、状态

            HealthMonitor定义了几个状态,如下:

  @InterfaceAudience.Private
  public enum State {
    /**
     * The health monitor is still starting up.
     * 健康监视器正在启动
     */
    INITIALIZING,

    /**
     * The service is not responding to health check RPCs.
     * 健康监测RPCs服务没有响应
     */
    SERVICE_NOT_RESPONDING,

    /**
     * The service is connected and healthy.
     * 服务已连接且健康
     */
    SERVICE_HEALTHY,
    
    /**
     * The service is running but unhealthy.
     * 服务正在运行但是不健康
     */
    SERVICE_UNHEALTHY,
    
    /**
     * The health monitor itself failed unrecoverably and can
     * no longer provide accurate information.
     * 健康监视器自己发生不可恢复故障且不能再提供准确信息
     */
    HEALTH_MONITOR_FAILED;
  }
            分别对应了HealthMonitor监视服务健康状况过程中的一些状态,如下:

            1)INITIALIZING:健康监视器正在启动;

            2)SERVICE_NOT_RESPONDING:健康监测RPCs服务没有响应,有可能是连接、rpc通讯问题等等;

            3)SERVICE_HEALTHY:服务已连接且健康,这是一种比较理想的结果;

            4)SERVICE_UNHEALTHY:服务正在运行但是不健康,这个结果也不赖,至少知道了服务不健康,也属于正常的检测结果;

            5)HEALTH_MONITOR_FAILED:健康监视器自己发生不可恢复故障且不能再提供准确信息,糟糕透顶,监视线程本身出故障了。
            2.2、监视器回调接口

            HealthMonitor定义了两个接口,Callback和ServiceStateCallback,分别是在状态发生变更和服务状态发生变更时的,需要周知所有监听器的回调方法,如下:

  /**
   * Callback interface for state change events.
   * 状态变更事件的回调接口,需要实现enteredState()方法,即进入一种新的状态
   * 
   * This interface is called from a single thread which also performs
   * the health monitoring. If the callback processing takes a long time,
   * no further health checks will be made during this period, nor will
   * other registered callbacks be called.
   * 这个接口被一个执行健康监测的特定线程调用。如果回调处理需要很长时间,在此期间没有进一步的健康检查进行,其他注册回调函数也不会被调用。
   * 
   * If the callback itself throws an unchecked exception, no other
   * callbacks following it will be called, and the health monitor
   * will terminate, entering HEALTH_MONITOR_FAILED state.
   * 
   * 
   */
  static interface Callback {
    void enteredState(State newState);
  }

  /**
   * Callback interface for service states.
   * 服务状态的回调接口
   */
  static interface ServiceStateCallback {
    void reportServiceStatus(HAServiceStatus status);
  }
      至于上述状态和接口如何调用,在下面会详解。

      3、内部工作流程

       HealthMonitor的工作流程,主要看内部工作线程的执行,如下:

  /**
   * 监视器后台线程
   *
   */
  private class MonitorDaemon extends Daemon {
    
   /**
    * 私有内部类的私有构造函数  
    */
   private MonitorDaemon() {
      super();
      
      // 设置线程名称
      setName("Health Monitor for " + targetToMonitor);
      
      // 设置异常处理:调用enterState()方法,并传入健康监控器自己失败的状态
      setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
          LOG.fatal("Health monitor failed", e);
          enterState(HealthMonitor.State.HEALTH_MONITOR_FAILED);
        }
      });
    }
    
   /**
    * 线程核心run()方法
    */
    @Override
    public void run() {
      while (shouldRun) {// 在一个while循环内,循环的条件是shouldRun标志位
        try { 
          loopUntilConnected();// 循环直到连接上
          doHealthChecks();// 执行健康检查
        } catch (InterruptedException ie) {
          Preconditions.checkState(!shouldRun,
              "Interrupted but still supposed to run");
        }
      }
    }
  }
      首先,监视后台线程的核心run()方法,在一个while循环内,循环的条件是shouldRun标志位:

      1)循环直到连接上:不停的连接,直到连接上;

      2)执行健康检查。

      如此简单的两步,下面我们需要看下以下问题:

      1)如何连接?

      2)如何执行健康检查,得到检查结果后如何处理?


      1、如何连接?

       分析loopUntilConnected()方法如下:

  /**
   * 循环直到连接上
   */
  private void loopUntilConnected() throws InterruptedException {
    // 尝试连接
	tryConnect();
	
	// tryConnect()成功的话,proxy应该不为null
	// proxy为null,说明还需要重试,在一个while循环内进行,循环的条件就是proxy为null
    while (proxy == null) {
      // 线程休眠一段时间
      Thread.sleep(connectRetryInterval);
      
      // 再次尝试连接
      tryConnect();
    }
    
    // 最后的assert需要确保proxy不为null
    assert proxy != null;
  }


  private void tryConnect() {
    
	// 仅当代理proxy为空时才尝试连接
	Preconditions.checkState(proxy == null);
    
    try {
      synchronized (this) {// 同步代码块内创建代理proxy
        proxy = createProxy();
      }
    } catch (IOException e) {
      
      // 代理proxy创建发生异常时,proxy设置为null,方便下次重试,并且调用enterState()方法,确定状态为健康监测RPCs服务没有响应
      LOG.warn("Could not connect to local service at " + targetToMonitor +
          ": " + e.getMessage());
      proxy = null;
      enterState(State.SERVICE_NOT_RESPONDING);
    }
  }
  
  /**
   * Connect to the service to be monitored. Stubbed out for easier testing.
   */
  protected HAServiceProtocol createProxy() throws IOException {
	// 通过目标到监视器的对象targetToMonitor获取代理
    return targetToMonitor.getProxy(conf, rpcTimeout);
  }
        很简单,通过targetToMonitor获得代理,获取成功的话,直接返回,失败的话线程休眠一段时间,继续重试。



        2、如何执行健康检查,得到检查结果后如何处理?

        分析doHealthChecks()方法如下:

        在一个while循环内,循环的依据同样是shouldRun标志位为true,且线程会周期性休眠:

        1)HA服务状态status设置为null;

        2)标志位healthy默认为false,即不健康;

        3)获取服务代理的服务状态,并由代理执行健康检查;

        4)1、如果健康检查能够正确返回,标志位healthy设置为true,表明服务健康;

               2、抛出了异常:

                     2.1、如果是健康检查失败异常,调用enterState()方法,确定状态为服务正在运行但是不健康;

                     2.1、否则调用enterState()方法,确定状态为健康监测RPCs服务没有响应,停止并清空代理,线程休眠一段时间,避免异常情况下没有必要的重复尝试;

         5)设置上次服务状态;

         6)如果检测结果为健康,则调用enterState()方法,确定状态为服务已连接且健康;

         7)工作线程周期性休眠。

        可以看到,健康状况的检查是通过代理完成的,且,如果有检查结果(无论是正常还是不正常,这里的正常是指返回了明确的结果,不正常只通讯、连接或者线程本身出现问题),会通过enterState()方法通知监听器s,如下:

  private synchronized void enterState(State newState) {
    
	// 如果状态变更,调用所有监听器的enteredState()方法
	if (newState != state) {
      LOG.info("Entering state " + newState);
      state = newState;
      synchronized (callbacks) {
        for (Callback cb : callbacks) {
          cb.enteredState(newState);
        }
      }
    }
  }
        而整体HA服务状态也会通过setLastServiceStatus()方法,设置lastServiceState成员变量,并通知服务状态监听器,如下:

  // 设置上次服务状态,同步所有ServiceStateCallback服务状态监听器的服务状态
  private synchronized void setLastServiceStatus(HAServiceStatus status) {
    this.lastServiceState = status;
    for (ServiceStateCallback cb : serviceStateCallbacks) {
      cb.reportServiceStatus(lastServiceState);
    }
  }
        这也就是上述状态和接口存在的意义。而监听器的注册和注销则是通过如下实现的:

  /**
   * 以下四个为增减状态、服务状态监视器方法
   */
  public void addCallback(Callback cb) {
    this.callbacks.add(cb);
  }
  
  public void removeCallback(Callback cb) {
    callbacks.remove(cb);
  }

  public synchronized void addServiceStateCallback(ServiceStateCallback cb) {
    this.serviceStateCallbacks.add(cb);
  }

  public synchronized void removeServiceStateCallback(ServiceStateCallback cb) {
    serviceStateCallbacks.remove(cb);
  }

三、HealthMonitor的初始化及启动

        HealthMonitor的构造及启动是在ZKFailoverController的initHM()方法内完成的,它是整个Hadoop HDFS HA中的一个控制组件,初始化及启动如下:

  // ------------------------------------------
  // Begin actual guts of failover controller
  // ------------------------------------------
  
  private void initHM() {
    healthMonitor = new HealthMonitor(conf, localTarget);
    healthMonitor.addCallback(new HealthCallbacks());
    healthMonitor.addServiceStateCallback(new ServiceStateCallBacks());
    healthMonitor.start();
  }
        构造一个HealthMonitor实例,传入的代理为localTarget,并注册了一个state change的监听器HealthCallbacks和service state change的监听器ServiceStateCallBacks,然后调用start()方法启动。

        至于HealthMonitor如何与其它组件一起工作的,如何获得的代理,代理怎么执行监控检查,监听器如何处理健康检查的结果,请关注后续文章。

 类似资料: