共计 11173 个字符，预计需要花费 28 分钟才能阅读完成。
序
本文主要研究一下 jest 的 NodeChecker
NodeChecker
jest-common-6.3.1-sources.jar!/io/searchbox/client/config/discovery/NodeChecker.java
public class NodeChecker extends AbstractScheduledService {private final static Logger log = LoggerFactory.getLogger(NodeChecker.class);
private final static String PUBLISH_ADDRESS_KEY = "http_address";
private final static String PUBLISH_ADDRESS_KEY_V5 = "publish_address"; // The one that under "http" node
private final static Pattern INETSOCKETADDRESS_PATTERN = Pattern.compile("(?:inet\\[)?(?:(?:[^:]+)?\\/)?([^:]+):(\\d+)\\]?");
private final NodesInfo action;
protected JestClient client;
protected Scheduler scheduler;
protected String defaultScheme;
protected Set<String> bootstrapServerList;
protected Set<String> discoveredServerList;
public NodeChecker(JestClient jestClient, ClientConfig clientConfig) {action = new NodesInfo.Builder()
.withHttp()
.addNode(clientConfig.getDiscoveryFilter())
.build();
this.client = jestClient;
this.defaultScheme = clientConfig.getDefaultSchemeForDiscoveredNodes();
this.scheduler = Scheduler.newFixedDelaySchedule(
0l,
clientConfig.getDiscoveryFrequency(),
clientConfig.getDiscoveryFrequencyTimeUnit());
this.bootstrapServerList = ImmutableSet.copyOf(clientConfig.getServerList());
this.discoveredServerList = new LinkedHashSet<String>();}
@Override
protected void runOneIteration() throws Exception {
JestResult result;
try {result = client.execute(action);
} catch (CouldNotConnectException cnce) {
// Can't connect to this node, remove it from the list
log.error("Connect exception executing NodesInfo!", cnce);
removeNodeAndUpdateServers(cnce.getHost());
return;
// do not elevate the exception since that will stop the scheduled calls.
// throw new RuntimeException("Error executing NodesInfo!", e);
} catch (Exception e) {log.error("Error executing NodesInfo!", e);
client.setServers(bootstrapServerList);
return;
// do not elevate the exception since that will stop the scheduled calls.
// throw new RuntimeException("Error executing NodesInfo!", e);
}
if (result.isSucceeded()) {LinkedHashSet<String> httpHosts = new LinkedHashSet<String>();
JsonObject jsonMap = result.getJsonObject();
JsonObject nodes = (JsonObject) jsonMap.get("nodes");
if (nodes != null) {for (Entry<String, JsonElement> entry : nodes.entrySet()) {JsonObject host = entry.getValue().getAsJsonObject();
JsonElement addressElement = null;
if (host.has("version")) {int majorVersion = Integer.parseInt(Splitter.on('.').splitToList(host.get("version").getAsString()).get(0));
if (majorVersion >= 5) {JsonObject http = host.getAsJsonObject("http");
if (http != null && http.has(PUBLISH_ADDRESS_KEY_V5))
addressElement = http.get(PUBLISH_ADDRESS_KEY_V5);
}
}
if (addressElement == null) {
// get as a JsonElement first as some nodes in the cluster may not have an http_address
if (host.has(PUBLISH_ADDRESS_KEY)) addressElement = host.get(PUBLISH_ADDRESS_KEY);
}
if (addressElement != null && !addressElement.isJsonNull()) {String httpAddress = getHttpAddress(addressElement.getAsString());
if(httpAddress != null) httpHosts.add(httpAddress);
}
}
}
if (log.isDebugEnabled()) {log.debug("Discovered {} HTTP hosts: {}", httpHosts.size(), Joiner.on(',').join(httpHosts));
}
discoveredServerList = httpHosts;
client.setServers(discoveredServerList);
} else {log.warn("NodesInfo request resulted in error: {}", result.getErrorMessage());
client.setServers(bootstrapServerList);
}
}
protected void removeNodeAndUpdateServers(final String hostToRemove) {log.warn("Removing host {}", hostToRemove);
discoveredServerList.remove(hostToRemove);
if (log.isInfoEnabled()) {log.info("Discovered server pool is now: {}", Joiner.on(',').join(discoveredServerList));
}
if (!discoveredServerList.isEmpty()) {client.setServers(discoveredServerList);
} else {client.setServers(bootstrapServerList);
}
}
@Override
protected Scheduler scheduler() {return scheduler;}
@Override
protected ScheduledExecutorService executor() {
final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder()
.setDaemon(true)
.setNameFormat(serviceName())
.build());
// Add a listener to shutdown the executor after the service is stopped. This ensures that the
// JVM shutdown will not be prevented from exiting after this service has stopped or failed.
// Technically this listener is added after start() was called so it is a little gross, but it
// is called within doStart() so we know that the service cannot terminate or fail concurrently
// with adding this listener so it is impossible to miss an event that we are interested in.
addListener(new Listener() {@Override public void terminated(State from) {executor.shutdown();
}
@Override public void failed(State from, Throwable failure) {executor.shutdown();
}}, MoreExecutors.directExecutor());
return executor;
}
/**
* Converts the Elasticsearch reported publish address in the format "inet[<hostname>:<port>]" or
* "inet[<hostname>/<hostaddress>:<port>]" to a normalized http address in the form "http://host:port".
*/
protected String getHttpAddress(String httpAddress) {Matcher resolvedMatcher = INETSOCKETADDRESS_PATTERN.matcher(httpAddress);
if (resolvedMatcher.matches()) {return defaultScheme + resolvedMatcher.group(1) + ":" + resolvedMatcher.group(2);
}
return null;
}
}- NodeChecker 继承了 AbstractScheduledService，它的构造器根据 clientConfig 的 discoveryFrequency 及 discoveryFrequencyTimeUnit 新建了 fixedDelayScheduler 来执行 node checker；它实现了 runOneIteration 方法，该方法主要是发送 NodesInfo 请求 (
GET /_nodes/_all/http) - 如果请求抛出 CouldNotConnectException 则调用 removeNodeAndUpdateServers 方法移除该 host；如果抛出其他的 Exception 则将 client 的 servers 重置为 bootstrapServerList
- 如果请求成功则解析 body，如果 nodes 下面有 version 且大于等于 5 则取 http 节点下面的 PUBLISH_ADDRESS_KEY_V5(
publish_address) 属性值添加到 discoveredServerList；旧版本的则从 nodes 下面的 PUBLISH_ADDRESS_KEY(http_address) 属性值添加到 discoveredServerList
NodesInfo 返回实例
{
"_nodes" : {
"total" : 1,
"successful" : 1,
"failed" : 0
},
"cluster_name" : "docker-cluster",
"nodes" : {
"RmyGhZEbTjC7JCQFVS3HWQ" : {
"name" : "RmyGhZE",
"transport_address" : "172.17.0.2:9300",
"host" : "172.17.0.2",
"ip" : "172.17.0.2",
"version" : "6.6.2",
"build_flavor" : "oss",
"build_type" : "tar",
"build_hash" : "3bd3e59",
"roles" : [
"master",
"data",
"ingest"
],
"http" : {
"bound_address" : ["0.0.0.0:9200"],
"publish_address" : "192.168.99.100:9200",
"max_content_length_in_bytes" : 104857600
}
}
}
}- 如果是 5 版本及以上的则在 nodes 下面有 http 属性，里头有 publish_address 属性用于返回该 node 的 publish address
JestHttpClient
jest-6.3.1-sources.jar!/io/searchbox/client/http/JestHttpClient.java
public class JestHttpClient extends AbstractJestClient {
//......
@Override
public <T extends JestResult> T execute(Action<T> clientRequest) throws IOException {return execute(clientRequest, null);
}
public <T extends JestResult> T execute(Action<T> clientRequest, RequestConfig requestConfig) throws IOException {HttpUriRequest request = prepareRequest(clientRequest, requestConfig);
CloseableHttpResponse response = null;
try {response = executeRequest(request);
return deserializeResponse(response, request, clientRequest);
} catch (HttpHostConnectException ex) {throw new CouldNotConnectException(ex.getHost().toURI(), ex);
} finally {if (response != null) {
try {response.close();
} catch (IOException ex) {log.error("Exception occurred while closing response stream.", ex);
}
}
}
}
@Override
public <T extends JestResult> void executeAsync(final Action<T> clientRequest, final JestResultHandler<? super T> resultHandler) {executeAsync(clientRequest, resultHandler, null);
}
public <T extends JestResult> void executeAsync(final Action<T> clientRequest, final JestResultHandler<? super T> resultHandler, final RequestConfig requestConfig) {synchronized (this) {if (!asyncClient.isRunning()) {asyncClient.start();
}
}
HttpUriRequest request = prepareRequest(clientRequest, requestConfig);
executeAsyncRequest(clientRequest, resultHandler, request);
}
protected <T extends JestResult> HttpUriRequest prepareRequest(final Action<T> clientRequest, final RequestConfig requestConfig) {String elasticSearchRestUrl = getRequestURL(getNextServer(), clientRequest.getURI(elasticsearchVersion));
HttpUriRequest request = constructHttpMethod(clientRequest.getRestMethodName(), elasticSearchRestUrl, clientRequest.getData(gson), requestConfig);
log.debug("Request method={} url={}", clientRequest.getRestMethodName(), elasticSearchRestUrl);
// add headers added to action
for (Entry<String, Object> header : clientRequest.getHeaders().entrySet()) {request.addHeader(header.getKey(), header.getValue().toString());
}
return request;
}
protected String getNextServer() {return serverPoolReference.get().getNextServer();}
//......
}- JestHttpClient 继承了 AbstractJestClient，它的 execute 及 executeAsync 方法都调用了 prepareRequest 来构造 HttpUriRequest；该方法会先调用 getNextServer 方法来获取要请求的 elasticSearchServer 的地址；而 getNextServer 方法则是调用的 serverPoolReference.get().getNextServer()
AbstractJestClient
jest-common-6.3.1-sources.jar!/io/searchbox/client/AbstractJestClient.java
public abstract class AbstractJestClient implements JestClient {
private final AtomicReference<ServerPool> serverPoolReference =
new AtomicReference<ServerPool>(new ServerPool(ImmutableSet.<String>of()));
//......
public void setServers(Set<String> servers) {if (servers.equals(serverPoolReference.get().getServers())) {if (log.isDebugEnabled()) {log.debug("Server pool already contains same list of servers: {}",
Joiner.on(',').join(scrubServerURIs(servers)));
}
return;
}
if (log.isInfoEnabled()) {log.info("Setting server pool to a list of {} servers: [{}]",
servers.size(), Joiner.on(',').join(scrubServerURIs(servers)));
}
serverPoolReference.set(new ServerPool(servers));
if (servers.isEmpty()) {log.warn("No servers are currently available to connect.");
}
}
//......
}- AbstractJestClient 有一个 serverPoolReference 属性，它是 AtomicReference，其泛型为 ServerPool；setServers 方法则是创建新的 ServerPool，然后更新 serverPoolReference
ServerPool
jest-common-6.3.1-sources.jar!/io/searchbox/client/AbstractJestClient.java
private static final class ServerPool {
private final List<String> serversRing;
private final AtomicInteger nextServerIndex = new AtomicInteger(0);
public ServerPool(final Set<String> servers) {this.serversRing = ImmutableList.copyOf(servers);
}
public Set<String> getServers() {return ImmutableSet.copyOf(serversRing);
}
public String getNextServer() {if (serversRing.size() > 0) {
try {return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
} catch (IndexOutOfBoundsException outOfBoundsException) {
// In the very rare case where nextServerIndex overflowed, this will end up with a negative number,
// resulting in an IndexOutOfBoundsException.
// We should then start back at the beginning of the server list.
// Note that this might happen on several threads at once, in which the reset might happen a few times
log.info("Resetting next server index");
nextServerIndex.set(0);
return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
}
}
else {throw new NoServerConfiguredException("No Server is assigned to client to connect");
}
}
public int getSize() {return serversRing.size();
}
}- ServerPool 有个 AtomicInteger 类型的 nextServerIndex，getNextServer 方法则是通过 nextServerIndex.getAndIncrement() % serversRing.size() 来确定取的 serversRing 这个 List 的 index，其实现的是 Round Robin 策略；极端情况下出现 IndexOutOfBoundsException 的话，则会重置 nextServerIndex 为 0，然后继续按 Round Robin 策略取下一个 server
小结
- NodeChecker 继承了 AbstractScheduledService，它的构造器根据 clientConfig 的 discoveryFrequency 及 discoveryFrequencyTimeUnit 新建了 fixedDelayScheduler 来执行 node checker；它实现了 runOneIteration 方法，该方法主要是发送 NodesInfo 请求 (
GET /_nodes/_all/http)，然后获取 nodes 的 PUBLISH_ADDRESS 来更新 discoveredServerList - JestHttpClient 继承了 AbstractJestClient，它的 execute 及 executeAsync 方法都调用了 prepareRequest 来构造 HttpUriRequest；该方法会先调用 getNextServer 方法来获取要请求的 elasticSearchServer 的地址；而 getNextServer 方法则是调用的 serverPoolReference.get().getNextServer()；AbstractJestClient 有一个 serverPoolReference 属性，它是 AtomicReference，其泛型为 ServerPool；setServers 方法则是创建新的 ServerPool，然后更新 serverPoolReference
- ServerPool 有个 AtomicInteger 类型的 nextServerIndex，getNextServer 方法则是通过 nextServerIndex.getAndIncrement() % serversRing.size() 来确定取的 serversRing 这个 List 的 index，其实现的是 Round Robin 策略；极端情况下出现 IndexOutOfBoundsException 的话，则会重置 nextServerIndex 为 0，然后继续按 Round Robin 策略取下一个 server
doc
- NodeChecker
正文完
