elasticSearch6源码分析(3)cluster模块

1. cluser概述

One of the main roles of the master is to decide which shards to allocate to which nodes, and when to move shards between nodes in order to rebalance the cluster.

2. ClusterModule模块的作用

Configures classes and services that affect the entire cluster

3.重点类介绍：

  3.1 配置类Decider

　　各种Decider关系如下

 

以EnableAllocationDecider.java为例：

public static final Setting<Allocation> CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING =new Setting<>("cluster.routing.allocation.enable", Allocation.ALL.toString(), Allocation::parse,Property.Dynamic, Property.NodeScope);public static final Setting<Allocation> INDEX_ROUTING_ALLOCATION_ENABLE_SETTING =new Setting<>("index.routing.allocation.enable", Allocation.ALL.toString(), Allocation::parse,Property.Dynamic, Property.IndexScope);public static final Setting<Rebalance> CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING =new Setting<>("cluster.routing.rebalance.enable", Rebalance.ALL.toString(), Rebalance::parse,Property.Dynamic, Property.NodeScope);public static final Setting<Rebalance> INDEX_ROUTING_REBALANCE_ENABLE_SETTING =new Setting<>("index.routing.rebalance.enable", Rebalance.ALL.toString(), Rebalance::parse,Property.Dynamic, Property.IndexScope);private volatile Rebalance enableRebalance;private volatile Allocation enableAllocation;

 

Decider的使用

ClusterModule.java public static Collection<AllocationDecider> createAllocationDeciders(Settings settings, ClusterSettings clusterSettings,List<ClusterPlugin> clusterPlugins) {// collect deciders by class so that we can detect duplicatesMap<Class, AllocationDecider> deciders = new LinkedHashMap<>();addAllocationDecider(deciders, new MaxRetryAllocationDecider(settings));addAllocationDecider(deciders, new ResizeAllocationDecider(settings));addAllocationDecider(deciders, new ReplicaAfterPrimaryActiveAllocationDecider(settings));addAllocationDecider(deciders, new RebalanceOnlyWhenActiveAllocationDecider(settings));addAllocationDecider(deciders, new ClusterRebalanceAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new ConcurrentRebalanceAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new EnableAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new NodeVersionAllocationDecider(settings));addAllocationDecider(deciders, new SnapshotInProgressAllocationDecider(settings));addAllocationDecider(deciders, new RestoreInProgressAllocationDecider(settings));addAllocationDecider(deciders, new FilterAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new SameShardAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new DiskThresholdDecider(settings, clusterSettings));addAllocationDecider(deciders, new ThrottlingAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new ShardsLimitAllocationDecider(settings, clusterSettings));addAllocationDecider(deciders, new AwarenessAllocationDecider(settings, clusterSettings));clusterPlugins.stream().flatMap(p -> p.createAllocationDeciders(settings, clusterSettings).stream()).forEach(d -> addAllocationDecider(deciders, d));return deciders.values();}

3.2 配置类Allocator

实现类ShardsAllocator

使用

private static ShardsAllocator createShardsAllocator(Settings settings, ClusterSettings clusterSettings,List<ClusterPlugin> clusterPlugins) {Map<String, Supplier<ShardsAllocator>> allocators = new HashMap<>();allocators.put(BALANCED_ALLOCATOR, () -> new BalancedShardsAllocator(settings, clusterSettings));for (ClusterPlugin plugin : clusterPlugins) {plugin.getShardsAllocators(settings, clusterSettings).forEach((k, v) -> {if (allocators.put(k, v) != null) {throw new IllegalArgumentException("ShardsAllocator [" + k + "] already defined");}});}String allocatorName = SHARDS_ALLOCATOR_TYPE_SETTING.get(settings);Supplier<ShardsAllocator> allocatorSupplier = allocators.get(allocatorName);if (allocatorSupplier == null) {throw new IllegalArgumentException("Unknown ShardsAllocator [" + allocatorName + "]");}return Objects.requireNonNull(allocatorSupplier.get(),"ShardsAllocator factory for [" + allocatorName + "] returned null");}

3.3 服务类

ClusterService.java @Overrideprotected synchronized void doStart() {clusterApplierService.start();masterService.start();}

分别调用

2.3.1clusterApplierService

@Overrideprotected synchronized void doStart() {Objects.requireNonNull(nodeConnectionsService, "please set the node connection service before starting");Objects.requireNonNull(state.get(), "please set initial state before starting");addListener(localNodeMasterListeners);threadPoolExecutor = EsExecutors.newSinglePrioritizing(nodeName + "/" + CLUSTER_UPDATE_THREAD_NAME,daemonThreadFactory(settings, CLUSTER_UPDATE_THREAD_NAME),threadPool.getThreadContext(),threadPool.scheduler());}

3.3.2 masterService

@Overrideprotected synchronized void doStart() {Objects.requireNonNull(clusterStatePublisher, "please set a cluster state publisher before starting");Objects.requireNonNull(clusterStateSupplier, "please set a cluster state supplier before starting");threadPoolExecutor = EsExecutors.newSinglePrioritizing(nodeName + "/" + MASTER_UPDATE_THREAD_NAME,daemonThreadFactory(settings, MASTER_UPDATE_THREAD_NAME),threadPool.getThreadContext(),threadPool.scheduler());taskBatcher = new Batcher(logger, threadPoolExecutor);}

 

转载于:https://www.cnblogs.com/davidwang456/p/10102927.html

总结

以上是生活随笔为你收集整理的elasticSearch6源码分析(3)cluster模块的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。