从一个生产的问题分析ElasticSearch负载均衡算法

/*** Sends a request to the Elasticsearch cluster that the client points to.* Blocks until the request is completed and returns its response or fails* by throwing an exception. Selects a host out of the provided ones in a* round-robin fashion. Failing hosts are marked dead and retried after a* certain amount of time (minimum 1 minute, maximum 30 minutes), depending* on how many times they previously failed (the more failures, the later* they will be retried). In case of failures all of the alive nodes (or* dead nodes that deserve a retry) are retried until one responds or none* of them does, in which case an {@link IOException} will be thrown.** This method works by performing an asynchronous call and waiting* for the result. If the asynchronous call throws an exception we wrap* it and rethrow it so that the stack trace attached to the exception* contains the call site. While we attempt to preserve the original* exception this isn't always possible and likely haven't covered all of* the cases. You can get the original exception from* {@link Exception#getCause()}.** @param request the request to perform* @return the response returned by Elasticsearch* @throws IOException in case of a problem or the connection was aborted* @throws ClientProtocolException in case of an http protocol error* @throws ResponseException in case Elasticsearch responded with a status code that indicated an error*/

public Response performRequest(Request request) throws IOException {     InternalRequest internalRequest = new InternalRequest(request);     return performRequest(nextNodes(), internalRequest, null); }

RequestContext context = request.createContextForNextAttempt(nodeTuple.nodes.next(), nodeTuple.authCache);

static Iterable<Node> selectNodes(NodeTuple<List<Node>> nodeTuple, Map<HttpHost, DeadHostState> blacklist,                                      AtomicInteger lastNodeIndex, NodeSelector nodeSelector) throws IOException {        /\*         \* Sort the nodes into living and dead lists.         \*/        List<Node> livingNodes = new ArrayList<>(Math.max(0, nodeTuple.nodes.size() - blacklist.size()));        List<DeadNode> deadNodes = new ArrayList<>(blacklist.size());        for (Node node : nodeTuple.nodes) {            DeadHostState deadness = blacklist.get(node.getHost());            if (deadness == null || deadness.shallBeRetried()) {                livingNodes.add(node);            } else {                deadNodes.add(new DeadNode(node, deadness));            }        }
        if (false == livingNodes.isEmpty()) {            /\*             \* Normal state: there is at least one living node. If the             \* selector is ok with any over the living nodes then use them             \* for the request.             \*/            List<Node> selectedLivingNodes = new ArrayList<>(livingNodes);            nodeSelector.select(selectedLivingNodes);            if (false == selectedLivingNodes.isEmpty()) {                /\*                 \* Rotate the list using a global counter as the distance so subsequent                 \* requests will try the nodes in a different order.                 \*/                Collections.rotate(selectedLivingNodes, lastNodeIndex.getAndIncrement());                return selectedLivingNodes;            }        }

/\*\* \* Select the {@link Node}s to which to send requests. This is called with \* a mutable {@link Iterable} of {@linkplain Node}s in the order that the \* rest client would prefer to use them and implementers should remove \* nodes from the that should not receive the request. Implementers may \* iterate the nodes as many times as they need. \* <p> \* This may be called twice per request: first for "living" nodes that \* have not been blacklisted by previous errors. If the selector removes \* all nodes from the list or if there aren't any living nodes then the \* {@link RestClient} will call this method with a list of "dead" nodes. \* <p> \* Implementers should not rely on the ordering of the nodes. \*/void select(Iterable<Node> nodes);

try {            httpResponse = client.execute(context.requestProducer, context.asyncResponseConsumer, context.context, null).get();        } catch(Exception e) {            RequestLogger.logFailedRequest(logger, request.httpRequest, context.node, e);            onFailure(context.node);            Exception cause = extractAndWrapCause(e);            addSuppressedException(previousException, cause);            if (nodeTuple.nodes.hasNext()) {                return performRequest(nodeTuple, request, cause);            }            if (cause instanceof IOException) {                throw (IOException) cause;            }            if (cause instanceof RuntimeException) {                throw (RuntimeException) cause;            }            throw new IllegalStateException("unexpected exception type: must be either RuntimeException or IOException", cause);        }

/\*\* \* Indicates whether it's time to retry to failed host or not. \* \* @return true if the host should be retried, false otherwise \*/boolean shallBeRetried() {    return timeSupplier.get() - deadUntilNanos > 0;}

DeadHostState(Supplier timeSupplier) {this.failedAttempts = 1;this.deadUntilNanos = timeSupplier.get() + MIN_CONNECTION_TIMEOUT_NANOS;this.timeSupplier = timeSupplier;}

/\*\* \* Build the dead state of a host given its previous dead state. Useful when a host has been failing before, hence \* it already failed for one or more consecutive times. The more failed attempts we register the longer we wait \* to retry that same host again. Minimum is 1 minute (for a node the only failed once created \* through {@link #DeadHostState(Supplier)}), maximum is 30 minutes (for a node that failed more than 10 consecutive times) \* \* @param previousDeadHostState the previous state of the host which allows us to increase the wait till the next retry attempt \*/DeadHostState(DeadHostState previousDeadHostState) {    long timeoutNanos = (long)Math.min(MIN\_CONNECTION\_TIMEOUT\_NANOS \* 2 \* Math.pow(2, previousDeadHostState.failedAttempts \* 0.5 - 1),            MAX\_CONNECTION\_TIMEOUT\_NANOS);    this.deadUntilNanos = previousDeadHostState.timeSupplier.get() + timeoutNanos;    this.failedAttempts = previousDeadHostState.failedAttempts + 1;    this.timeSupplier = previousDeadHostState.timeSupplier;}