es源码笔记-如何选择协调节点

协调节点作为es节点中的一个节点，默认情况下es集群中所有的节点都能当协调节点，主要作用于请求转发，请求响应处理等轻量级操作。

但是在生产环境中，当客户端通过REST API向es服务端发起一个请求时，会有以下几个问题？
1、服务端有多个节点的情况下，客户端该发给哪个节点处理？
2、处理的节点是不是一成不变的？
3、节点连接失败，该如何处理？
等等。。。

es的RestClient作为一个成熟的客户端，这些问题必然考虑在内了。下面我们跟踪源码，看看es是如何解决这些问题的？

es的RestClient

ES提供了两个JAVA REST client 版本

Java Low Level REST Client: 低级别的REST客户端，通过http与集群交互，用户需自己编组请求JSON串，及解析响应JSON串。兼容所有ES版本。
Java High Level REST Client: 高级别的REST客户端，基于低级别的REST客户端，增加了编组请求JSON串、解析响应JSON串等相关api。

例子：

//高
RestHighLevelClient client = new RestHighLevelClient(
        RestClient.builder(
                new HttpHost("localhost", 9200, "http"),
                new HttpHost("localhost", 9201, "http")));
//低
RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200, "http"),
        new HttpHost("localhost", 9201, "http")).build();

流程

1、创建restClient

创建restClient 由RestClientBuilder实现，该类采用构建器模式创建restClient，具体代码如下：org.elasticsearch.client.RestClientBuilder

private NodeSelector nodeSelector = NodeSelector.ANY;
public static RestClientBuilder builder(HttpHost... hosts) {      
        List<Node> nodes = Arrays.stream(hosts).map(Node::new).collect(Collectors.toList());
        return new RestClientBuilder(nodes);
    }
    public RestClient build() {      
        CloseableHttpAsyncClient httpClient = AccessController.doPrivileged(
            (PrivilegedAction<CloseableHttpAsyncClient>) this::createHttpClient);
        RestClient restClient = new RestClient(httpClient, defaultHeaders, nodes,
                pathPrefix, failureListener, nodeSelector, strictDeprecationMode);
        httpClient.start();
        return restClient;
    }

从上面代码可以看出两个关键的地方：
1、在创建RestClient时，可以自定义nodeSelector，默认情况下是ANY，下面会详细说
2、创建成功后，启动httpClient，可以看出底层还是通过httpClient来通信的

2、NodeSelector

上面提到NodeSelector在RestClient创建时需要传递进来，那么NodeSelector有什么用处呢？
NodeSelector是节点选择器，通过该选择器，客户端可以解决服务器端多节点选择以及节点均衡处理等等问题。
截止到7.3版本，restclient提供了四种选择器，分别是：HasAttributeNodeSelector，PreferHasAttributeNodeSelector，ANY，SKIP_DEDICATED_MASTERS，都实现了select（）方法

其中前面两种可以根据用户配置选择指定的节点，但是会造成节点轮训不均匀以及节点挂了以后导致不可用等问题
ANY：是默认选择器，select（）方法是个空实现，即所有的节点都可以做为协调节点
SKIP_DEDICATED_MASTERS：过滤掉master，data，Ingest节点

下面来看它具体如何处理的？
restClient接收到请求后，交给org.elasticsearch.client.RestClient.performRequest(Request)处理，最终通过selectNodes（）方法进行协调节点的选择，代码流程如下：

public Response performRequest(Request request) throws IOException {
        InternalRequest internalRequest = new InternalRequest(request);
        return performRequest(nextNodes(), internalRequest, null);
    }
private NodeTuple<Iterator<Node>> nextNodes() throws IOException {
        NodeTuple<List<Node>> nodeTuple = this.nodeTuple;
        Iterable<Node> hosts = selectNodes(nodeTuple, blacklist, lastNodeIndex, nodeSelector);
        return new NodeTuple<>(hosts.iterator(), nodeTuple.authCache);
    }

上面的代码有几个参数需要注意：

NodeTuple：存储所有的节点，包括加入黑名单的节点
blacklist：黑名单列表，这个列表用于存放连接失败后的节点，但需要注意的是，黑名单是有时间限制的，即默认情况下一分钟，超过一分钟后，黑名单中的节点重新加入到livingNodes节点列表中。可以修改MIN_CONNECTION_TIMEOUT_NANOS参数减短或者增加节点黑名单时间
nodeSelector：选择器，上面提过了，这里就不说了
lastNodeIndex：节点旋转距离，这个有什么用呢？
举个例子：假如列表中有[t, a, n, k, s]，lastNodeIndex为1
返回的结果将为：[s, t, a, n, k]
即保证节点列表中节点都被使用到
具体算法实现是在java.util.Collections.rotate(List<?>, int)方法中

private static <T> void rotate1(List<T> list, int distance) {
        int size = list.size();
        if (size == 0)
            return;
        distance = distance % size;
        if (distance < 0)
            distance += size;
        if (distance == 0)
            return;

        for (int cycleStart = 0, nMoved = 0; nMoved != size; cycleStart++) {
            T displaced = list.get(cycleStart);
            int i = cycleStart;
            do {
                i += distance;
                if (i >= size)
                    i -= size;
                displaced = list.set(i, displaced);
                nMoved ++;
            } while (i != cycleStart);
        }
    }

具体的选择过程

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.
         */
//1、拿到活跃的节点列表，包括黑名单中已经到时间的节点，默认是一分钟
        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));
            }
        }
      //2、如果有活跃的节点列表，则通过nodeSelector选择一个节点
        if (false == livingNodes.isEmpty()) {
           //nodeSelector选出合适的节点
            List<Node> selectedLivingNodes = new ArrayList<>(livingNodes);
            nodeSelector.select(selectedLivingNodes);
            if (false == selectedLivingNodes.isEmpty()) {
              //2、1选择成功后，旋转列表，确保下次请求时选择不同的节点
                Collections.rotate(selectedLivingNodes, lastNodeIndex.getAndIncrement());
                return selectedLivingNodes;
            }
        }

       //3、如果没有活跃的节点列表，则从死亡节点（连接失败的）列表中选择连接失败时间最早的节点
        if (false == deadNodes.isEmpty()) {
            final List<DeadNode> selectedDeadNodes = new ArrayList<>(deadNodes);        
            nodeSelector.select(() -> new DeadNodeIteratorAdapter(selectedDeadNodes.iterator()));
            if (false == selectedDeadNodes.isEmpty()) {
                return singletonList(Collections.min(selectedDeadNodes).node);
            }
        }
//4、如果都没有，则抛异常
        throw new IOException("NodeSelector [" + nodeSelector + "] rejected all nodes, "
                + "living " + livingNodes + " and dead " + deadNodes);
    }

具体流程，代码上已经写了注释了，就不再详细说了。下面来说说黑名单列表以及第3点

（1）黑名单列表

当节点连接失败时，这个节点将被加入到黑名单列表中，并设置黑名单时间为1分钟（默认）

private Response performRequest(final NodeTuple<Iterator<Node>> nodeTuple,
                                    final InternalRequest request,
                                    Exception previousException) throws IOException {
    
        try {
            httpResponse = client.execute(context.requestProducer, context.asyncResponseConsumer, context.context, null).get();
        } catch(Exception e) {
          //加入黑名单中
            onFailure(context.node);
         
    }

private void onFailure(Node node) {
        while(true) {
            DeadHostState previousDeadHostState =
                blacklist.putIfAbsent(node.getHost(), new DeadHostState(DeadHostState.DEFAULT_TIME_SUPPLIER));          
        }     
    }

过期时间判断

boolean shallBeRetried() {
        return timeSupplier.get() - deadUntilNanos > 0;
    }

timeSupplier是当前时间，deadUntilNanos 在黑名单停留时间，当当前时间大于黑名单停留时间，那么这个节点将可以复活了

（2）死亡节点列表比较

当没有活跃的节点时，即所有的节点都连接失败，这种情况不常见。
那么选择器将从死亡节点列表中挑选一个死亡时间最长的节点，也就是该节点在黑名单列表停留时间最长，说明该节点被恢复的几率更大，优先被选择
org.elasticsearch.client.RestClient.DeadNode

@Override
        public int compareTo(DeadNode rhs) {
            return deadness.compareTo(rhs.deadness);
        }

 @Override
    public int compareTo(DeadHostState other) {      
        return Long.compare(deadUntilNanos, other.deadUntilNanos);
    }

总结

1、RestClient通过以上的策略保证了集群中各个节点都能均匀被调用，不会导致某个节点被高负载使用
2、但是有个不足的地方，没有做到真正的负载均衡，比如一些配置好的节点应该比配置差的节点调用多点，频繁失败的节点应该降低它的调用次数等。