序
本文主要研究一下 nacos 的 HttpHealthCheckProcessor
HealthCheckProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckProcessor.java
public interface HealthCheckProcessor {
/**
* Run check task for service
*
* @param task check task
*/
void process(HealthCheckTask task);
/**
* Get check task type, refer to enum HealthCheckType
*
* @return check type
*/
String getType();}
- HealthCheckProcessor 接口定义了 process、getType 方法
HttpHealthCheckProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HttpHealthCheckProcessor.java
@Component
public class HttpHealthCheckProcessor implements HealthCheckProcessor {
@Autowired
private SwitchDomain switchDomain;
@Autowired
private HealthCheckCommon healthCheckCommon;
private static AsyncHttpClient asyncHttpClient;
private static final int CONNECT_TIMEOUT_MS = 500;
static {
try {AsyncHttpClientConfig.Builder builder = new AsyncHttpClientConfig.Builder();
builder.setMaximumConnectionsTotal(-1);
builder.setMaximumConnectionsPerHost(-1);
builder.setAllowPoolingConnection(false);
builder.setFollowRedirects(false);
builder.setIdleConnectionTimeoutInMs(CONNECT_TIMEOUT_MS);
builder.setConnectionTimeoutInMs(CONNECT_TIMEOUT_MS);
builder.setCompressionEnabled(false);
builder.setIOThreadMultiplier(1);
builder.setMaxRequestRetry(0);
builder.setUserAgent("VIPServer");
asyncHttpClient = new AsyncHttpClient(builder.build());
} catch (Throwable e) {SRV_LOG.error("[HEALTH-CHECK] Error while constructing HTTP asynchronous client", e);
}
}
@Override
public String getType() {return "HTTP";}
@Override
public void process(HealthCheckTask task) {List<Instance> ips = task.getCluster().allIPs(false);
if (CollectionUtils.isEmpty(ips)) {return;}
if (!switchDomain.isHealthCheckEnabled()) {return;}
Cluster cluster = task.getCluster();
for (Instance ip : ips) {
try {if (ip.isMarked()) {if (SRV_LOG.isDebugEnabled()) {SRV_LOG.debug("http check, ip is marked as to skip health check, ip: {}" + ip.getIp());
}
continue;
}
if (!ip.markChecking()) {SRV_LOG.warn("http check started before last one finished, service: {}:{}:{}",
task.getCluster().getService().getName(), task.getCluster().getName(), ip.getIp());
healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getHttpHealthParams());
continue;
}
AbstractHealthChecker.Http healthChecker = (AbstractHealthChecker.Http) cluster.getHealthChecker();
int ckPort = cluster.isUseIPPort4Check() ? ip.getPort() : cluster.getDefCkport();
URL host = new URL("http://" + ip.getIp() + ":" + ckPort);
URL target = new URL(host, healthChecker.getPath());
AsyncHttpClient.BoundRequestBuilder builder = asyncHttpClient.prepareGet(target.toString());
Map<String, String> customHeaders = healthChecker.getCustomHeaders();
for (Map.Entry<String, String> entry : customHeaders.entrySet()) {if ("Host".equals(entry.getKey())) {builder.setVirtualHost(entry.getValue());
continue;
}
builder.setHeader(entry.getKey(), entry.getValue());
}
builder.execute(new HttpHealthCheckCallback(ip, task));
MetricsMonitor.getHttpHealthCheckMonitor().incrementAndGet();
} catch (Throwable e) {ip.setCheckRT(switchDomain.getHttpHealthParams().getMax());
healthCheckCommon.checkFail(ip, task, "http:error:" + e.getMessage());
healthCheckCommon.reEvaluateCheckRT(switchDomain.getHttpHealthParams().getMax(), task, switchDomain.getHttpHealthParams());
}
}
}
//......
}
- HttpHealthCheckProcessor 实现了 HealthCheckProcessor 接口,其 static 方法初始化了 AsyncHttpClient;其 getType 返回的是 HTTP;其 process 方法会遍历 instances,然后对于非 marked 及 markChecking 的执行 health check,并注册 HttpHealthCheckCallback;对于非 markChecking 的或者出现异常的则执行 healthCheckCommon.reEvaluateCheckRT
HttpHealthCheckCallback
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HttpHealthCheckProcessor.java
private class HttpHealthCheckCallback extends AsyncCompletionHandler<Integer> {
private Instance ip;
private HealthCheckTask task;
private long startTime = System.currentTimeMillis();
public HttpHealthCheckCallback(Instance ip, HealthCheckTask task) {
this.ip = ip;
this.task = task;
}
@Override
public Integer onCompleted(Response response) throws Exception {ip.setCheckRT(System.currentTimeMillis() - startTime);
int httpCode = response.getStatusCode();
if (HttpURLConnection.HTTP_OK == httpCode) {healthCheckCommon.checkOK(ip, task, "http:" + httpCode);
healthCheckCommon.reEvaluateCheckRT(System.currentTimeMillis() - startTime, task, switchDomain.getHttpHealthParams());
} else if (HttpURLConnection.HTTP_UNAVAILABLE == httpCode || HttpURLConnection.HTTP_MOVED_TEMP == httpCode) {
// server is busy, need verification later
healthCheckCommon.checkFail(ip, task, "http:" + httpCode);
healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getHttpHealthParams());
} else {
//probably means the state files has been removed by administrator
healthCheckCommon.checkFailNow(ip, task, "http:" + httpCode);
healthCheckCommon.reEvaluateCheckRT(switchDomain.getHttpHealthParams().getMax(), task, switchDomain.getHttpHealthParams());
}
return httpCode;
}
@Override
public void onThrowable(Throwable t) {ip.setCheckRT(System.currentTimeMillis() - startTime);
Throwable cause = t;
int maxStackDepth = 50;
for (int deepth = 0; deepth < maxStackDepth && cause != null; deepth++) {
if (cause instanceof SocketTimeoutException
|| cause instanceof ConnectTimeoutException
|| cause instanceof org.jboss.netty.channel.ConnectTimeoutException
|| cause instanceof TimeoutException
|| cause.getCause() instanceof TimeoutException) {healthCheckCommon.checkFail(ip, task, "http:timeout:" + cause.getMessage());
healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getHttpHealthParams());
return;
}
cause = cause.getCause();}
// connection error, probably not reachable
if (t instanceof ConnectException) {healthCheckCommon.checkFailNow(ip, task, "http:unable2connect:" + t.getMessage());
healthCheckCommon.reEvaluateCheckRT(switchDomain.getHttpHealthParams().getMax(), task, switchDomain.getHttpHealthParams());
} else {healthCheckCommon.checkFail(ip, task, "http:error:" + t.getMessage());
healthCheckCommon.reEvaluateCheckRT(switchDomain.getHttpHealthParams().getMax(), task, switchDomain.getHttpHealthParams());
}
}
}
- HttpHealthCheckCallback 实现了 async http client 的 AsyncCompletionHandler;其 onCompleted 方法对于 HTTP_OK 执行 healthCheckCommon.checkOK;对于 HTTP_UNAVAILABLE 及 HTTP_MOVED_TEMP 执行 healthCheckCommon.checkFail,其余的执行 healthCheckCommon.checkFailNow;其 onThrowable 针对不同条件执行 healthCheckCommon.checkFail 或者 healthCheckCommon.checkFailNow
小结
HttpHealthCheckProcessor 实现了 HealthCheckProcessor 接口,其 static 方法初始化了 AsyncHttpClient;其 getType 返回的是 HTTP;其 process 方法会遍历 instances,然后对于非 marked 及 markChecking 的执行 health check,并注册 HttpHealthCheckCallback;对于非 markChecking 的或者出现异常的则执行 healthCheckCommon.reEvaluateCheckRT
doc
- HttpHealthCheckProcessor