elasticSearch6源码分析(3)cluster模块
2024-08-26 17:25:10
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 duplicates
Map<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
@Override
protected synchronized void doStart() {
clusterApplierService.start();
masterService.start();
}
分别调用
2.3.1clusterApplierService
@Override
protected 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
@Override
protected 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);
}
最新文章
- JavaScript面试时候的坑洼沟洄——数据类型
- C#资源文件管理
- EXCEL 建立工作薄与工作表
- Hadoop-MapReduce之自定义数据类型
- HDU2088JAVA2
- ACM_2057
- AltiumDesigner导入AutoCAD文件DXF,DWG格式
- 《CSS设计指南》阅读笔记
- RabbitMQ消息队列(六):使用主题进行消息分发
- 从零学习Fluter(九):正式使用flutter进入商业开发
- 关于springboot2.x 的 RedisCacheManager变化
- Dostoevsky: Better Space-Time Trade-Offs for LSM-Tree Based Key-Value Stores via Adaptive Removal of Superfluous Merging 阅读笔记
- System.Web.Optimization对脚本和样式表的压缩操作
- URL参数编码
- JavaSE学习总结(三)——Java语言编程练习、格式化字符与常量
- php安装及配置笔记
- 获取SQL Server数据库中的表和字段描述
- nginx根据http_user_agent防DDOS
- jQuery实现ToolTip元素定位显示功能示例
- 快速上手Vue