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