zoukankan      html  css  js  c++  java
  • 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 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);
        }
  • 相关阅读:
    遇到的相关问题总结
    AI测试相关文章
    常用模块文档地址
    09-微服务接口:怎么用Mock解决混乱的调用关系
    03-思维方式:用一个案例彻底理解接口测试的关键逻辑
    1-基础:跳出细节看全局,接口测试到底是在做什么?
    21-Python并发编程之Futures
    使用原生php读写excel文件
    在for、foreach循环体中添加数组元素
    eval函数的坑
  • 原文地址:https://www.cnblogs.com/davidwang456/p/10102927.html
Copyright © 2011-2022 走看看