ES群集仅配置了一个节点。客户端能否自动发现群集中的所有节点？是怎么发现的？配置以下节点：单个节点如何配置es客户端实现负载均衡？在一个es节点挂起后，es客户端如何删除该节点？Es客户端节点检测分为两种模式。有什么区别？
传输客户端3360 ES客户端外部API类传输客户端nodeService 3360维护节点节点类ScheduledNodeSampler :定期维护正常节点类NettyTransport :节点采样器初始化代码：节点嗅探类
用于数据传输
设置。builder builder=settings . settings builder()。put('cluster.name '，clusterName)。put('client.transport.sniff '，true)；Settings设置=builder . build()；transport client client=transport client . builder()。设置(settings)。build()；for(transport address transport address : transport addresses){ client . addtransportaddress(transport address)；}ES通过builder模式构造基本配置参数；客户端由build构造，包括构造客户端、初始化线程池、构造TransportClientNodesService、启动定时任务、定制嗅探类型。添加集群的可用地址，比如我在集群中只有一个节点；构建客户端
调用构建API
构建代码
其中，简单描述一下依赖注入：Guice是Google用于Java开发的开源依赖注入框架(有兴趣的可以了解一下，这里不做重点介绍)。请参考以下详细信息：
Google Guice开始Guice博客1Guice博客2
初始化TransportClientNodesService
在上图中，modules.createInjector实例化了TransportClientNodeService并将其注入到TransportClient中。可以看到，TransportClient中的大多数API都由TransportClientNodeService表示：
运输客户代码
Guice是由注释注入的。
Guice注释注入代码
上图中，集群名，线程池等。都是注入的，重点是下面这段代码：这段代码选择节点嗅探器的类型来嗅探同一个集群中的所有节点，SniffNodesSampler还是只专注于配置文件配置的SimpleNodeSampler。
if(this . settings . getas boolean(' client . transport . sniff '，false)){ this . nodessampler=new SniffNodesSampler()；} else { this . nodessampler=new simple node sampler()；}特点,SniffNodesSampler:客户端会主动发现集群中的其他节点并创建全连接(什么是全连接？后来)SimpleNodeSampler: ping listed nodes中的所有节点，不同的是这里创建的所有节点都是LightConnectTransportClientNodesService维护三种节点存储数据结构：
//添加到discovered 1 private volatilelistednodes=collections . empty list()的节点；2 private volatile listnodes=collections . empty list()；volatileListfilteredNodes=Co
llections.emptyList();代表配置文件中主动加入的节点；代表参与请求的节点；过滤掉的不能进行请求处理的节点；
负载均衡code
如上图，我们发现每次 execute 的时候，是从 nodes 这个数据结构中获取节点，然后通过简单的 rouund-robbin 获取节点服务器，核心代码如下：
private final AtomicInteger randomNodeGenerator = new AtomicInteger();......private int getNodeNumber() { int index = randomNodeGenerator.incrementAndGet(); if (index < 0) { index = 0; randomNodeGenerator.set(0); } return index;}然后通过netty的channel将数据写入，核心代码如下：
public void sendRequest(final DiscoveryNode node, final long requestId, final String action, final TransportRequest request, TransportRequestOptions options) throws IOException, TransportException {1 Channel targetChannel = nodeChannel(node, options); if (compress) { options = TransportRequestOptions.builder(options).withCompress(true).build(); } byte status = 0; status = TransportStatus.setRequest(status); ReleasableBytesStreamOutput bStream = new ReleasableBytesStreamOutput(bigArrays); boolean addedReleaseListener = false; try { bStream.skip(NettyHeader.HEADER_SIZE); StreamOutput stream = bStream; // only compress if asked, and, the request is not bytes, since then only // the header part is compressed, and the "body" can't be extracted as compressed if (options.compress() && (!(request instanceof BytesTransportRequest))) { status = TransportStatus.setCompress(status); stream = CompressorFactory.defaultCompressor().streamOutput(stream); } // we pick the smallest of the 2, to support both backward and forward compatibility // note, this is the only place we need to do this, since from here on, we use the serialized version // as the version to use also when the node receiving this request will send the response with Version version = Version.smallest(this.version, node.version()); stream.setVersion(version); stream.writeString(action); ReleasablePagedBytesReference bytes; ChannelBuffer buffer; // it might be nice to somehow generalize this optimization, maybe a smart "paged" bytes output // that create paged channel buffers, but its tricky to know when to do it (where this option is // more explicit). if (request instanceof BytesTransportRequest) { BytesTransportRequest bRequest = (BytesTransportRequest) request; assert node.version().equals(bRequest.version()); bRequest.writeThin(stream); stream.close(); bytes = bStream.bytes(); ChannelBuffer headerBuffer = bytes.toChannelBuffer(); ChannelBuffer contentBuffer = bRequest.bytes().toChannelBuffer(); buffer = ChannelBuffers.wrappedBuffer(NettyUtils.DEFAULT_GATHERING, headerBuffer, contentBuffer); } else { request.writeTo(stream); stream.close(); bytes = bStream.bytes(); buffer = bytes.toChannelBuffer(); } NettyHeader.writeHeader(buffer, requestId, status, version);2 ChannelFuture future = targetChannel.write(buffer); ReleaseChannelFutureListener listener= new ReleaseChannelFutureListener(bytes); future.addListener(listener); addedReleaseListener = true; transportServiceAdapter.onRequestSent(node, requestId, action, request, options); } finally { if (!addedReleaseListener) { Releasables.close(bStream.bytes()); } }}其中最重要的就是1和2
1代表拿到一个连接；2代表通过拿到的连接写数据；这时候就会有新的问题
nodes的数据是何时写入的？连接是什么时候创建的？
protected void doSample() { // the nodes we are going to ping include the core listed nodes that were added // and the last round of discovered nodes SetnodesToPing = Sets.newHashSet(); for (DiscoveryNode node : listedNodes) { nodesToPing.add(node); }for (DiscoveryNode node : nodes) { nodesToPing.add(node); } final CountDownLatch latch = new CountDownLatch(nodesToPing.size()); final ConcurrentMapclusterStateResponses = ConcurrentCollections.newConcurrentMap(); for (final DiscoveryNode listedNode : nodesToPing) { threadPool.executor(ThreadPool.Names.MANAGEMENT).execute(new Runnable() { @Override public void run() { try { if (!transportService.nodeConnected(listedNode)) { try { // if its one of the actual nodes we will talk to, not to listed nodes, fully connect if (nodes.contains(listedNode)) { logger.trace("connecting to cluster node <{}>", listedNode); transportService.connectToNode(listedNode); } else { // its a listed node, light connect to it... logger.trace("connecting to listed node (light) <{}>", listedNode); transportService.connectToNodeLight(listedNode); } } catch (Exception e) { logger.debug("failed to connect to node <{}>, ignoring...", e, listedNode); latch.countDown(); return; } }//核心是在这里，刚刚开始初始化的时候，可能只有配置的一个节点，这个时候会通过这个地址发送一个state状态监测 //"cluster:monitor/state" transportService.sendRequest(listedNode, ClusterStateAction.NAME, headers.applyTo(Requests.clusterStateRequest().clear().nodes(true).local(true)), TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STAE).withTimeout(pingTimeout).build(), new BaseTransportResponseHandler() { @Override public ClusterStateResponse newInstance() { return new ClusterStateResponse(); } @Override public String executor() { return ThreadPool.Names.SAME; } @Override public void handleResponse(ClusterStateResponse response) {/*通过回调，会在这个地方返回集群中类似下边所有节点的信息{ "version" : 27, "state_uuid" : "YSI9d_HiQJ-FFAtGFCVOlw", "master_node" : "TXHHx-XRQaiXAxtP1EzXMw", "blocks" : { }, "nodes" : { "poxubF0LTVue84GMrZ7rwA" : { "name" : "node1", "transport_address" : "1.1.1.1:8888", "attributes" : { "data" : "false", "master" : "true" } }, "9Cz8m3GkTza7vgmpf3L65Q" : { "name" : "node2", "transport_address" : "1.1.1.2:8889", "attributes" : { "master" : "false" } } }, "metadata" : { "cluster_uuid" : "_na_", "templates" : { }, "indices" : { } }, "routing_table" : { "indices" : { } }, "routing_nodes" : { "unassigned" : < >, "nodes" : { "lZqD-WExRu-gaSUiCXaJcg" : < >, "hR6PbFrgQVSY0MHajNDmgA" : < >, } }}*/ clusterStateResponses.put(listedNode, response); latch.countDown(); } @Override public void handleException(TransportException e) { logger.info("failed to get local cluster state for {}, disconnecting...", e, listedNode); transportService.disconnectFromNode(listedNode); latch.countDown(); } });} catch (Throwable e) { logger.info("failed to get local cluster state info for {}, disconnecting...", e, listedNode); transportService.disconnectFromNode(listedNode); latch.countDown(); }}});} try { latch.await(); } catch (InterruptedException e) { return; } HashSetnewNodes = new HashSet<>(); HashSetnewFilteredNodes = new HashSet<>(); for (Map.Entryentry : clusterStateResponses.entrySet()) { if (!ignoreClusterName &&!clusterName.equals(entry.getValue().getClusterName())) { logger.warn("node {} not part of the cluster {}, ignoring...", entry.getValue().getState().nodes().localNode(), clusterName); newFilteredNodes.add(entry.getKey()); continue; }//接下来在这个地方拿到所有的data nodes 写入到nodes节点里边 for (ObjectCursorcursor : entry.getValue().getState().nodes().dataNodes().values()){ newNodes.add(cursor.value);}} nodes = validateNewNodes(newNodes); filteredNodes = Collections.unmodifiableList(new ArrayList<(newFilteredNodes)); }其中调用时机分为两部分：
client.addTransportAddress(transportAddress);ScheduledNodeSampler，默认每隔5s会进行一次对各个节点的请求操作；
创建连接code
这个时候发现，如果是light则创建轻连接，也就是，否则创建fully connect，其中包括：recovery：做数据恢复recovery，默认个数2个；
bulk：用于bulk请求，默认个数3个；med/reg：典型的搜索和单doc索引，默认个数6个；high:如集群state的发送等，默认个数1个；ping：就是node之间的ping咯。默认个数1个；对应的代码为：
public void start() { List<Channel> newAllChannels = new ArrayList<>(); newAllChannels.addAll(Arrays.asList(recovery)); newAllChannels.addAll(Arrays.asList(bulk)); newAllChannels.addAll(Arrays.asList(reg)); newAllChannels.addAll(Arrays.asList(state)); newAllChannels.addAll(Arrays.asList(ping)); this.allChannels = Collections.unmodifiableList(newAllChannels);}END
