序
本文主要研究一下 nacos 的 HealthCheckCommon
HealthCheckCommon
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckCommon.java
@Component
public class HealthCheckCommon {
@Autowired
private DistroMapper distroMapper;
@Autowired
private SwitchDomain switchDomain;
@Autowired
private ServerListManager serverListManager;
@Autowired
private PushService pushService;
private static LinkedBlockingDeque<HealthCheckResult> healthCheckResults = new LinkedBlockingDeque<>(1024 * 128);
private static ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("com.taobao.health-check.notifier");
return thread;
}
});
public void init() {executorService.schedule(new Runnable() {
@Override
public void run() {List list = Arrays.asList(healthCheckResults.toArray());
healthCheckResults.clear();
List<Server> sameSiteServers = serverListManager.getServers();
if (sameSiteServers == null || sameSiteServers.size() <= 0) {return;}
for (Server server : sameSiteServers) {if (server.getKey().equals(NetUtils.localServer())) {continue;}
Map<String, String> params = new HashMap<>(10);
params.put("result", JSON.toJSONString(list));
if (Loggers.SRV_LOG.isDebugEnabled()) {Loggers.SRV_LOG.debug("[HEALTH-SYNC] server: {}, healthCheckResults: {}",
server, JSON.toJSONString(list));
}
HttpClient.HttpResult httpResult = HttpClient.httpPost("http://" + server.getKey()
+ RunningConfig.getContextPath() + UtilsAndCommons.NACOS_NAMING_CONTEXT
+ "/api/healthCheckResult", null, params);
if (httpResult.code != HttpURLConnection.HTTP_OK) {Loggers.EVT_LOG.warn("[HEALTH-CHECK-SYNC] failed to send result to {}, result: {}",
server, JSON.toJSONString(list));
}
}
}
}, 500, TimeUnit.MILLISECONDS);
}
//......
public void reEvaluateCheckRT(long checkRT, HealthCheckTask task, SwitchDomain.HealthParams params) {//......}
public void checkOK(Instance ip, HealthCheckTask task, String msg) {//......}
public void checkFail(Instance ip, HealthCheckTask task, String msg) {//......}
public void checkFailNow(Instance ip, HealthCheckTask task, String msg) {//......}
//......
}
- HealthCheckCommon 的 init 方法注册了一个延时任务,往其他 server 同步 healthCheckResults;它主要提供了 reEvaluateCheckRT、checkOK、checkFail、checkFailNow 方法
reEvaluateCheckRT
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckCommon.java
public void reEvaluateCheckRT(long checkRT, HealthCheckTask task, SwitchDomain.HealthParams params) {task.setCheckRTLast(checkRT);
if (checkRT > task.getCheckRTWorst()) {task.setCheckRTWorst(checkRT);
}
if (checkRT < task.getCheckRTBest()) {task.setCheckRTBest(checkRT);
}
checkRT = (long) ((params.getFactor() * task.getCheckRTNormalized()) + (1 - params.getFactor()) * checkRT);
if (checkRT > params.getMax()) {checkRT = params.getMax();
}
if (checkRT < params.getMin()) {checkRT = params.getMin();
}
task.setCheckRTNormalized(checkRT);
}
- reEvaluateCheckRT 方法首先更新 checkRTLast,然后判断是否更新 checkRTWorst、checkRTBest,之后根据 factor 及 checkRTNormalized 参数重置 checkRT,最后更新 checkRTNormalized
checkOK
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckCommon.java
public void checkOK(Instance ip, HealthCheckTask task, String msg) {Cluster cluster = task.getCluster();
try {if (!ip.isHealthy() || !ip.isMockValid()) {if (ip.getOKCount().incrementAndGet() >= switchDomain.getCheckTimes()) {if (distroMapper.responsible(cluster, ip)) {ip.setHealthy(true);
ip.setMockValid(true);
Service service = cluster.getService();
service.setLastModifiedMillis(System.currentTimeMillis());
pushService.serviceChanged(service);
addResult(new HealthCheckResult(service.getName(), ip));
Loggers.EVT_LOG.info("serviceName: {} {POS} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
} else {if (!ip.isMockValid()) {ip.setMockValid(true);
Loggers.EVT_LOG.info("serviceName: {} {PROBE} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
}
}
} else {Loggers.EVT_LOG.info("serviceName: {} {OTHER} {IP-ENABLED} pre-valid: {}:{}@{} in {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), ip.getOKCount(), msg);
}
}
} catch (Throwable t) {Loggers.SRV_LOG.error("[CHECK-OK] error when close check task.", t);
}
ip.getFailCount().set(0);
ip.setBeingChecked(false);
}
- checkOK 对于非 healthy 或者 mockValid 的 instance 会设置其为 healthy 及 mockValid,然后通过 pushService.serviceChanged 发布变更事件,并添加 HealthCheckResult 到 healthCheckResults 中
checkFail
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckCommon.java
public void checkFail(Instance ip, HealthCheckTask task, String msg) {Cluster cluster = task.getCluster();
try {if (ip.isHealthy() || ip.isMockValid()) {if (ip.getFailCount().incrementAndGet() >= switchDomain.getCheckTimes()) {if (distroMapper.responsible(cluster, ip)) {ip.setHealthy(false);
ip.setMockValid(false);
Service service = cluster.getService();
service.setLastModifiedMillis(System.currentTimeMillis());
addResult(new HealthCheckResult(service.getName(), ip));
pushService.serviceChanged(service);
Loggers.EVT_LOG.info("serviceName: {} {POS} {IP-DISABLED} invalid: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
} else {Loggers.EVT_LOG.info("serviceName: {} {PROBE} {IP-DISABLED} invalid: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
}
} else {Loggers.EVT_LOG.info("serviceName: {} {OTHER} {IP-DISABLED} pre-invalid: {}:{}@{} in {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), ip.getFailCount(), msg);
}
}
} catch (Throwable t) {Loggers.SRV_LOG.error("[CHECK-FAIL] error when close check task.", t);
}
ip.getOKCount().set(0);
ip.setBeingChecked(false);
}
- checkFail 对于 healthy 或者 mockValid 的 instance 会设置其 healthy 及 mockValid 为 false,然后通过 pushService.serviceChanged 发布变更事件,并添加 HealthCheckResult 到 healthCheckResults 中
checkFailNow
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckCommon.java
public void checkFailNow(Instance ip, HealthCheckTask task, String msg) {Cluster cluster = task.getCluster();
try {if (ip.isHealthy() || ip.isMockValid()) {if (distroMapper.responsible(cluster, ip)) {ip.setHealthy(false);
ip.setMockValid(false);
Service service = cluster.getService();
service.setLastModifiedMillis(System.currentTimeMillis());
pushService.serviceChanged(service);
addResult(new HealthCheckResult(service.getName(), ip));
Loggers.EVT_LOG.info("serviceName: {} {POS} {IP-DISABLED} invalid-now: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
} else {if (ip.isMockValid()) {ip.setMockValid(false);
Loggers.EVT_LOG.info("serviceName: {} {PROBE} {IP-DISABLED} invalid-now: {}:{}@{}, region: {}, msg: {}",
cluster.getService().getName(), ip.getIp(), ip.getPort(), cluster.getName(), UtilsAndCommons.LOCALHOST_SITE, msg);
}
}
}
} catch (Throwable t) {Loggers.SRV_LOG.error("[CHECK-FAIL-NOW] error when close check task.", t);
}
ip.getOKCount().set(0);
ip.setBeingChecked(false);
}
- checkFailNow 对于 healthy 或者 mockValid 的 instance 会设置其 healthy 及 mockValid 为 false,然后通过 pushService.serviceChanged 发布变更事件,并添加 HealthCheckResult 到 healthCheckResults 中;与 checkFail 不同的是它对于非自己负责的 instance 会立马标记 mockVlid 为 false
小结
HealthCheckCommon 的 init 方法注册了一个延时任务,往其他 server 同步 healthCheckResults;它主要提供了 reEvaluateCheckRT、checkOK、checkFail、checkFailNow 方法
doc
- HealthCheckCommon