Hadoop-2.7.0 HA实现细节分析之HealthMonitor
一、HealthMonitor是什么
HealthMonitor是一个周期性工作的后台线程，它在一个循环中周期性的同HA服务进行心跳，负责跟踪NameNode服务的健康状况，并在健康状况变化时调用failover控制器的回调方法。
二、HealthMonitor是如何实现的
1、成员变量
HealthMonitor内部有如下主要成员变量：
1）后台工作线程：真正卖力干活的
private Daemon daemon;// 后台工作线程 // 重连、周期性检查、失连后的睡眠时间等时间间隔
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;
}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");
}
}
}
}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);
}
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);
}
}
}
} // 设置上次服务状态，同步所有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如何与其它组件一起工作的，如何获得的代理，代理怎么执行监控检查，监听器如何处理健康检查的结果，请关注后续文章。