Loading [MathJax]/jax/output/CommonHTML/config.js
前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >聊聊elasticsearch的SeedHostsResolver

聊聊elasticsearch的SeedHostsResolver

原创
作者头像
code4it
发布于 2019-04-26 14:23:43
发布于 2019-04-26 14:23:43
80800
代码可运行
举报
文章被收录于专栏:码匠的流水账码匠的流水账
运行总次数:0
代码可运行

本文主要研究一下elasticsearch的SeedHostsResolver

ConfiguredHostsResolver

elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
    public interface ConfiguredHostsResolver {
        /**
         * Attempt to resolve the configured unicast hosts list to a list of transport addresses.
         *
         * @param consumer Consumer for the resolved list. May not be called if an error occurs or if another resolution attempt is in
         *                 progress.
         */
        void resolveConfiguredHosts(Consumer<List<TransportAddress>> consumer);
    }
  • ConfiguredHostsResolver接口定义了resolveConfiguredHosts方法用于解析配置的transport address列表

SeedHostsResolver

elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/SeedHostsResolver.java

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public class SeedHostsResolver extends AbstractLifecycleComponent implements ConfiguredHostsResolver {
    public static final Setting<Integer> LEGACY_DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING =
        Setting.intSetting("discovery.zen.ping.unicast.concurrent_connects", 10, 0, Setting.Property.NodeScope,
            Setting.Property.Deprecated);
    public static final Setting<TimeValue> LEGACY_DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT =
        Setting.positiveTimeSetting("discovery.zen.ping.unicast.hosts.resolve_timeout", TimeValue.timeValueSeconds(5),
            Setting.Property.NodeScope, Setting.Property.Deprecated);
    public static final Setting<Integer> DISCOVERY_SEED_RESOLVER_MAX_CONCURRENT_RESOLVERS_SETTING =
        Setting.intSetting("discovery.seed_resolver.max_concurrent_resolvers", 10, 0, Setting.Property.NodeScope);
    public static final Setting<TimeValue> DISCOVERY_SEED_RESOLVER_TIMEOUT_SETTING =
        Setting.positiveTimeSetting("discovery.seed_resolver.timeout", TimeValue.timeValueSeconds(5), Setting.Property.NodeScope);private static final Logger logger = LogManager.getLogger(SeedHostsResolver.class);private final Settings settings;
    private final AtomicBoolean resolveInProgress = new AtomicBoolean();
    private final TransportService transportService;
    private final SeedHostsProvider hostsProvider;
    private final SetOnce<ExecutorService> executorService = new SetOnce<>();
    private final TimeValue resolveTimeout;
    private final String nodeName;
    private final int concurrentConnects;public SeedHostsResolver(String nodeName, Settings settings, TransportService transportService,
                             SeedHostsProvider seedProvider) {
        this.settings = settings;
        this.nodeName = nodeName;
        this.transportService = transportService;
        this.hostsProvider = seedProvider;
        resolveTimeout = getResolveTimeout(settings);
        concurrentConnects = getMaxConcurrentResolvers(settings);
    }public static int getMaxConcurrentResolvers(Settings settings) {
        if (LEGACY_DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING.exists(settings)) {
            if (DISCOVERY_SEED_RESOLVER_MAX_CONCURRENT_RESOLVERS_SETTING.exists(settings)) {
                throw new IllegalArgumentException("it is forbidden to set both ["
                    + DISCOVERY_SEED_RESOLVER_MAX_CONCURRENT_RESOLVERS_SETTING.getKey() + "] and ["
                    + LEGACY_DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING.getKey() + "]");
            }
            return LEGACY_DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING.get(settings);
        }
        return DISCOVERY_SEED_RESOLVER_MAX_CONCURRENT_RESOLVERS_SETTING.get(settings);
    }public static TimeValue getResolveTimeout(Settings settings) {
        if (LEGACY_DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT.exists(settings)) {
            if (DISCOVERY_SEED_RESOLVER_TIMEOUT_SETTING.exists(settings)) {
                throw new IllegalArgumentException("it is forbidden to set both ["
                    + DISCOVERY_SEED_RESOLVER_TIMEOUT_SETTING.getKey() + "] and ["
                    + LEGACY_DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT.getKey() + "]");
            }
            return LEGACY_DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT.get(settings);
        }
        return DISCOVERY_SEED_RESOLVER_TIMEOUT_SETTING.get(settings);
    }/**
     * Resolves a list of hosts to a list of transport addresses. Each host is resolved into a transport address (or a collection of
     * addresses if the number of ports is greater than one). Host lookups are done in parallel using specified executor service up
     * to the specified resolve timeout.
     *
     * @param executorService  the executor service used to parallelize hostname lookups
     * @param logger           logger used for logging messages regarding hostname lookups
     * @param hosts            the hosts to resolve
     * @param limitPortCounts  the number of ports to resolve (should be 1 for non-local transport)
     * @param transportService the transport service
     * @param resolveTimeout   the timeout before returning from hostname lookups
     * @return a list of resolved transport addresses
     */
    public static List<TransportAddress> resolveHostsLists(
        final ExecutorService executorService,
        final Logger logger,
        final List<String> hosts,
        final int limitPortCounts,
        final TransportService transportService,
        final TimeValue resolveTimeout) {
        Objects.requireNonNull(executorService);
        Objects.requireNonNull(logger);
        Objects.requireNonNull(hosts);
        Objects.requireNonNull(transportService);
        Objects.requireNonNull(resolveTimeout);
        if (resolveTimeout.nanos() < 0) {
            throw new IllegalArgumentException("resolve timeout must be non-negative but was [" + resolveTimeout + "]");
        }
        // create tasks to submit to the executor service; we will wait up to resolveTimeout for these tasks to complete
        final List<Callable<TransportAddress[]>> callables =
            hosts
                .stream()
                .map(hn -> (Callable<TransportAddress[]>) () -> transportService.addressesFromString(hn, limitPortCounts))
                .collect(Collectors.toList());
        final List<Future<TransportAddress[]>> futures;
        try {
            futures = executorService.invokeAll(callables, resolveTimeout.nanos(), TimeUnit.NANOSECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return Collections.emptyList();
        }
        final List<TransportAddress> transportAddresses = new ArrayList<>();
        final Set<TransportAddress> localAddresses = new HashSet<>();
        localAddresses.add(transportService.boundAddress().publishAddress());
        localAddresses.addAll(Arrays.asList(transportService.boundAddress().boundAddresses()));
        // ExecutorService#invokeAll guarantees that the futures are returned in the iteration order of the tasks so we can associate the
        // hostname with the corresponding task by iterating together
        final Iterator<String> it = hosts.iterator();
        for (final Future<TransportAddress[]> future : futures) {
            final String hostname = it.next();
            if (!future.isCancelled()) {
                assert future.isDone();
                try {
                    final TransportAddress[] addresses = future.get();
                    logger.trace("resolved host [{}] to {}", hostname, addresses);
                    for (int addressId = 0; addressId < addresses.length; addressId++) {
                        final TransportAddress address = addresses[addressId];
                        // no point in pinging ourselves
                        if (localAddresses.contains(address) == false) {
                            transportAddresses.add(address);
                        }
                    }
                } catch (final ExecutionException e) {
                    assert e.getCause() != null;
                    final String message = "failed to resolve host [" + hostname + "]";
                    logger.warn(message, e.getCause());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    // ignore
                }
            } else {
                logger.warn("timed out after [{}] resolving host [{}]", resolveTimeout, hostname);
            }
        }
        return Collections.unmodifiableList(transportAddresses);
    }
​
    @Override
    protected void doStart() {
        logger.debug("using max_concurrent_resolvers [{}], resolver timeout [{}]", concurrentConnects, resolveTimeout);
        final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(settings, "[unicast_configured_hosts_resolver]");
        executorService.set(EsExecutors.newScaling(nodeName + "/" + "unicast_configured_hosts_resolver",
            0, concurrentConnects, 60, TimeUnit.SECONDS, threadFactory, transportService.getThreadPool().getThreadContext()));
    }
​
    @Override
    protected void doStop() {
        ThreadPool.terminate(executorService.get(), 10, TimeUnit.SECONDS);
    }
​
    @Override
    protected void doClose() {
    }
​
    @Override
    public void resolveConfiguredHosts(Consumer<List<TransportAddress>> consumer) {
        if (lifecycle.started() == false) {
            logger.debug("resolveConfiguredHosts: lifecycle is {}, not proceeding", lifecycle);
            return;
        }if (resolveInProgress.compareAndSet(false, true)) {
            transportService.getThreadPool().generic().execute(new AbstractRunnable() {
                @Override
                public void onFailure(Exception e) {
                    logger.debug("failure when resolving unicast hosts list", e);
                }
​
                @Override
                protected void doRun() {
                    if (lifecycle.started() == false) {
                        logger.debug("resolveConfiguredHosts.doRun: lifecycle is {}, not proceeding", lifecycle);
                        return;
                    }
​
                    List<TransportAddress> providedAddresses
                        = hostsProvider.getSeedAddresses((hosts, limitPortCounts)
                        -> resolveHostsLists(executorService.get(), logger, hosts, limitPortCounts,
                        transportService, resolveTimeout));
​
                    consumer.accept(providedAddresses);
                }
​
                @Override
                public void onAfter() {
                    resolveInProgress.set(false);
                }
​
                @Override
                public String toString() {
                    return "SeedHostsResolver resolving unicast hosts list";
                }
            });
        }
    }
}
  • SeedHostsResolver继承了AbstractLifecycleComponent,同时实现了ConfiguredHostsResolver接口;它提供了getMaxConcurrentResolvers、getResolveTimeout、resolveHostsLists(使用线程池并发执行transportService.addressesFromString)这几个静态方法
  • doStart方法使用EsExecutors.newScaling创建了EsThreadPoolExecutor;doStop方法则使用ThreadPool.terminate来终止线程池
  • resolveConfiguredHosts方法首先将resolveInProgress从false设置为true,之后通过transportService.getThreadPool()执行hostsProvider.getSeedAddresses,执行完成则设置resolveInProgress为false

小结

  • ConfiguredHostsResolver接口定义了resolveConfiguredHosts方法用于解析配置的transport address列表
  • SeedHostsResolver继承了AbstractLifecycleComponent,同时实现了ConfiguredHostsResolver接口;它提供了getMaxConcurrentResolvers、getResolveTimeout、resolveHostsLists(使用线程池并发执行transportService.addressesFromString)这几个静态方法
  • doStart方法使用EsExecutors.newScaling创建了EsThreadPoolExecutor;doStop方法则使用ThreadPool.terminate来终止线程池;resolveConfiguredHosts方法首先将resolveInProgress从false设置为true,之后通过transportService.getThreadPool()执行hostsProvider.getSeedAddresses,执行完成则设置resolveInProgress为false

doc

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
聊聊elasticsearch的SeedHostsProvider
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/SeedHostsProvider.java
code4it
2019/04/27
5730
聊聊elasticsearch的SeedHostsProvider
聊聊elasticsearch的PeerFinder
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java
code4it
2019/04/29
6890
聊聊elasticsearch的PeerFinder
聊聊elasticsearch的LagDetector
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/cluster/coordination/LagDetector.java
code4it
2019/05/13
8680
聊聊elasticsearch的LagDetector
聊聊elasticsearch的MasterFaultDetection
本文主要研究一下elasticsearch的MasterFaultDetection
code4it
2019/05/11
7290
聊聊elasticsearch的MasterFaultDetection
聊聊elasticsearch的TransportProxyClient
本文主要研究一下elasticsearch的TransportProxyClient
code4it
2019/04/19
7410
聊聊elasticsearch的TransportProxyClient
聊聊elasticsearch的MembershipAction
elasticsearch-6.7.1/server/src/main/java/org/elasticsearch/discovery/zen/MembershipAction.java
code4it
2019/04/25
7730
聊聊elasticsearch的MembershipAction
聊聊Elasticsearch的DiscoveryPlugin
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/plugins/DiscoveryPlugin.java
code4it
2019/05/27
5540
聊聊Elasticsearch的DiscoveryPlugin
聊聊elasticsearch的NodesFaultDetection
本文主要研究一下elasticsearch的NodesFaultDetection
code4it
2019/05/10
5950
聊聊elasticsearch的NodesFaultDetection
聊聊elasticsearch的ElectMasterService
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/discovery/zen/ElectMasterService.java
code4it
2019/05/12
8510
聊聊elasticsearch的ElectMasterService
聊聊Elasticsearch的MonitorService
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/monitor/MonitorService.java
code4it
2019/05/26
7360
聊聊Elasticsearch的MonitorService
ES系列(二):基于多播的集群发现实现原理解析
ES作为超强悍的搜索引擎,除了需要具有齐全的功能支持,超高的性能,还必须要有任意扩展的能力。一定程度上,它是一个大数据产品。而要做扩展性,集群自然少不了。然而单独的集群又是不够的,能够做的事情太少(比如通常的集群为负载均衡式对等集群),所以它需要自己组建合适自己的集群。也就是服务需要自动发现,自动协调集群实例。当然,这只是扩展性的第一步。
烂猪皮
2021/05/11
1K0
聊聊elasticsearch的ZenPing
elasticsearch-7.0.0/server/src/main/java/org/elasticsearch/discovery/zen/ZenPing.java
code4it
2019/04/28
6020
聊聊elasticsearch的ZenPing
Elasticsearch源码分析三之线程池
我们先来看一看ThreadPool的签名:public class ThreadPool implements Scheduler,ThreadPool实现了Scheduler,我们就来围绕Scheduler和ThreadPool中的方法和属性进行分析。首先来看一个类的结构图:
山行AI
2020/03/11
1.3K0
Elasticsearch源码分析三之线程池
蛋疼的ElasticSearch(三)之配置elasticsearch-analysis-ik和集群
1.下载https://github.com/medcl/elasticsearch-analysis-ik
用户2032165
2018/11/09
1.5K0
Elasticsearch源码分析七之集群选举流程分析
org.elasticsearch.node.Node#start方法中有ZenDiscovery初始化的部分:
山行AI
2020/03/12
1.2K0
Elasitcsearch底层系列之 Node启动过程源码解析
Elasticsearch 是一款开源的分布式搜索引擎,提供了近实时的查询能力和强大的聚合分析能力。与Elastic官方提供的其他组件(Beats、Logstash、Kibana)组合成Elastic Stack,提供了多种使用场景下数据摄入、清洗、存储、查询、可视化的完整解决方案,在搜索、日志分析、统计分析等领域有广泛应用。
morningchen
2018/12/29
2.6K0
Elasitcsearch底层系列之 Node启动过程源码解析
聊聊Elasticsearch的FixedExecutorBuilder
本文主要研究一下Elasticsearch的FixedExecutorBuilder
code4it
2019/06/15
5400
聊聊Elasticsearch的FixedExecutorBuilder
【ElasticSearch篇】--ElasticSearch从初识到安装和应用
ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎,基于RESTful web接口,在企业中全文搜索时,特别常用。
LhWorld哥陪你聊算法
2018/09/13
5730
【ElasticSearch篇】--ElasticSearch从初识到安装和应用
Elasticsearch源码分析二之Node节点创建与启动流程分析
紧接着昨天的Bootstrap的初始化来进行开篇,对应的是org.elasticsearch.bootstrap.Bootstrap#setup方法,详见代码片段:
山行AI
2020/03/11
1.1K0
【ES三周年】elasticsearch目录及配置参数说明
1.elasticsearch目录及其作用bin存放ES启动、关闭等脚本文件存放执行文件,例如启动脚本、密钥工具等configES的配置文件所在的目录,其中有三个主要的配置文件elasticsearch.yml是用于ES的最主要的配置文件jvm.options用于配置ES JVM设置log4j2.properties用于配置ES 日志记录的属性jdkES自带的JDK目录data默认的索引数据存储位置,实际中一般需要自行更改logs存储ES的运行日志默认的日志存放位置,实际中一般需要自行更改libES运行所需
张同学tty
2023/04/06
2.3K1
【ES三周年】elasticsearch目录及配置参数说明
相关推荐
聊聊elasticsearch的SeedHostsProvider
更多 >
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
本文部分代码块支持一键运行,欢迎体验
本文部分代码块支持一键运行,欢迎体验