ES的transportClient初始化过程解析

我不是女神ヾ 2022-03-09 03:23 258阅读 0赞

# TransportClient的初始化流程 #

## 创建client的代码 ##

在初始化TransportClient的时候，我们将嗅探模式开启，即client.transport.sniff设置为true.

Settings settings = Settings.settingsBuilder()
                    .put("client.transport.sniff", true).put("cluster.name", clusterName).build();

然后通过TransportClient client = TransportClient.builder().settings(settings).build(); 获取客户端。  
再向客户端添加ES节点  
client.addTransportAddresses(addrs);  
在测试环境的ES集群包含3个节点**node1,node2,node3**。  
我们将这3个节点都添加给client,即addrs中包含了node1,node2,node3的信息。

## 初始化客户端流程 ##

*  1、 addrs中的节点信息将会被放入listedNodes中。
 *  2、 TransportClientNodesService.NodeSampler 的sample()中去建立tcp连接。

NodeSampler 有2个子类SniffNodesSampler，SimpleNodeSampler。如果开启了嗅探模式用SniffNodesSampler，否则用SimpleNodeSampler：

if (this.settings.getAsBoolean("client.transport.sniff", false)) {
                this.nodesSampler = new TransportClientNodesService.SniffNodesSampler();
            } else {
                this.nodesSampler = new TransportClientNodesService.SimpleNodeSampler();
            }

## 嗅探模式建立连接的过程 ##

由于我们用的是嗅探模式，所以只关注SniffNodesSampler。

### 1、 将listedNodes的节点信息加载到nodesToPing中 ###

### 2、 与nodesToPing中的每个节点都建立一个tcp连接。 ###

Iterator i$x = nodesToPing.iterator();
    
                while(i$x.hasNext()) {
                    final DiscoveryNode listedNode = (DiscoveryNode)i$x.next();
                    TransportClientNodesService.this.threadPool.executor("management").execute(new Runnable() {
                        public void run() {
                            try {
                                if (!TransportClientNodesService.this.transportService.nodeConnected(listedNode)) {
                                    try {
                                        if (TransportClientNodesService.this.nodes.contains(listedNode)) {
                                            TransportClientNodesService.this.logger.trace("connecting to cluster node [{}]", new Object[]{listedNode});
                                            TransportClientNodesService.this.transportService.connectToNode(listedNode);
                                        } else {
                                            TransportClientNodesService.this.logger.trace("connecting to listed node (light) [{}]", new Object[]{listedNode});
                                            TransportClientNodesService.this.transportService.connectToNodeLight(listedNode);
                                        }

这里有2中连接，第一种是**light connect**,另外一种是**fully connect**。Light connect只建立1个channel，而fullyconnect建立了13个channel用于通信。13个channel的分布如下：

*  recovery：做数据恢复recovery，默认个数2个；
 *  bulk：用于bulk请求，默认个数3个；
 *  med/reg：典型的搜索和单doc索引，默认个数6个；
 *  state:如集群state的发送等，默认个数1个；
 *  ping：就是node之间的ping咯。默认个数1个；

this.connectionsPerNodeRecovery = this.settings.getAsInt("transport.netty.connections_per_node.recovery", settings.getAsInt("transport.connections_per_node.recovery", 2));
            this.connectionsPerNodeBulk = this.settings.getAsInt("transport.netty.connections_per_node.bulk", settings.getAsInt("transport.connections_per_node.bulk", 3));
            this.connectionsPerNodeReg = this.settings.getAsInt("transport.netty.connections_per_node.reg", settings.getAsInt("transport.connections_per_node.reg", 6));
            this.connectionsPerNodeState = this.settings.getAsInt("transport.netty.connections_per_node.high", settings.getAsInt("transport.connections_per_node.state", 1));
            this.connectionsPerNodePing = this.settings.getAsInt("transport.netty.connections_per_node.ping", settings.getAsInt("transport.connections_per_node.ping", 1));

**首次都会建立light connect。**

### 3、 通过light connect获取集群的state信息，解析出datanode，生成newNodes ###

TransportClientNodesService.this.transportService.sendRequest(listedNode, "cluster:monitor/state", (TransportRequest)TransportClientNodesService.this.headers.applyTo(Requests.clusterStateRequest().clear().nodes(true).local(true)), TransportRequestOptions.options().withType(Type.STATE).withTimeout(TransportClientNodesService.this.pingTimeout), new BaseTransportResponseHandler<ClusterStateResponse>() {
                                    
    
                                    public void handleResponse(ClusterStateResponse response) {
                                        clusterStateResponses.put(listedNode, response);
                                        latch.countDown();
                                    }
                                
                }
    
                
                HashSet<DiscoveryNode> newNodes = new HashSet();
                HashSet<DiscoveryNode> newFilteredNodes = new HashSet();
                Iterator i$xx = clusterStateResponses.entrySet().iterator();
    
                while(true) {
                    while(i$xx.hasNext()) {
                        Entry<DiscoveryNode, ClusterStateResponse> entry = (Entry)i$xx.next();
                        if (!TransportClientNodesService.this.ignoreClusterName && !TransportClientNodesService.this.clusterName.equals(((ClusterStateResponse)entry.getValue()).getClusterName())) {
                            TransportClientNodesService.this.logger.warn("node {} not part of the cluster {}, ignoring...", new Object[]{((ClusterStateResponse)entry.getValue()).getState().nodes().localNode(), TransportClientNodesService.this.clusterName});
                            newFilteredNodes.add(entry.getKey());
                        } else {
                            Iterator i$xxx = ((ClusterStateResponse)entry.getValue()).getState().nodes().dataNodes().values().iterator();
    
                            while(i$xxx.hasNext()) {
                                ObjectCursor<DiscoveryNode> cursor = (ObjectCursor)i$xxx.next();
                                newNodes.add(cursor.value);
                            }
                        }
                    }

### 4、 与newNodes中的每个节点都建立full connect,并且将newNodes的节点放入nodes中。 ###

1~4这个过程默认没5秒钟会执行一次，因此集群的节点变化对client是可以感知到的。下图是transportClient 嗅探模式下创建连接的过程：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxcXEwMTk5MTgx_size_16_color_FFFFFF_t_70]

综上所述client最终会与listedNodes中的每个节点建立1个lightConnect ，与ES集群中每个节点建立一个full connect 。1个lightConnect=1个tcp连接；1个fullConnect=13个tcp连接。因此在本文的例子中transportClient将与集群中的每个节点建立14个tcp连接。

## 请求路由 ##

建立了fullConnect的节点都会放入nodes中用于处理请求。Client将会用round-robin的算法决定nodes中哪个节点处理请求。代码如下：

private int getNodeNumber() {
            int index = this.randomNodeGenerator.incrementAndGet();
            if (index < 0) {
                index = 0;
                this.randomNodeGenerator.set(0);
            }
            return index;
    }
    public <Response> void execute(TransportClientNodesService.NodeListenerCallback<Response> callback, ActionListener<Response> listener) {
            List<DiscoveryNode> nodes = this.nodes;
            this.ensureNodesAreAvailable(nodes);
            int index = this.getNodeNumber();
            TransportClientNodesService.RetryListener<Response> retryListener = new TransportClientNodesService.RetryListener(callback, listener, nodes, index);
            DiscoveryNode node = (DiscoveryNode)nodes.get(index % nodes.size());
            try {
                callback.doWithNode(node, retryListener);
            } catch (Throwable var8) {
                listener.onFailure(var8);
            }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxcXEwMTk5MTgx_size_16_color_FFFFFF_t_70]: /images/20220309/e6677e1e26fc4139be14baeabcee4d7b.png