聊聊jest的NodeChecker（jest webpack）

序

本文主要研究一下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