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 删除。

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
ClickHouse之MaterializeMySQL引擎(十)
ClickHouse 20.8.2.3 版本新增加了 MaterializeMySQL 的 database 引擎,该 database 能 映射到 MySQL 中的某个 database ,并自动在 ClickHouse 中创建对应的ReplacingMergeTree。ClickHouse 服务做为 MySQL 副本,读取 Binlog 并执行 DDL 和 DML 请求,实现了基于 MySQL Binlog 机制的业务数据库实时同步功能。
Java技术债务
2022/08/09
6920
MySQL 主从同步配置
先在docker下创建几个 mysql server,docker-compose.xml 如下:
前Thoughtworks-杨焱
2021/12/07
2.6K0
golang源码分析:mysql同步工具gravity(1)
https://github.com/moiot/gravity是摩拜单车开源的一款mysql同步工具:它是由一系列简单插件组成的同步工具,提升了足够的可扩展性的同时,也保证了架构的简洁性。下面先分析下它的原理以及如何使用,然后分析下它的源码。
golangLeetcode
2023/09/06
5570
golang源码分析:mysql同步工具gravity(1)
从 MySQL 到 ClickHouse 实时数据同步 —— MaterializeMySQL + Materialized View
本篇演示使用 ClickHouse 的 MaterializeMySQL 数据库引擎和物化视图,实时将 MySQL 库表中的数据同步到 ClickHouse 的库表中。相关软件版本如下:
用户1148526
2024/04/20
4.1K0
从 MySQL 到 ClickHouse 实时数据同步 —— MaterializeMySQL + Materialized View
MySQL5.7基于GTID的主从配置
从MySQL 5.6.5 开始新增了一种基于 GTID 的复制方式。通过 GTID 保证了每个在主库上提交的事务在集群中有一个唯一的ID。这种方式强化了数据库的主备一致性,故障恢复以及容错能力。 GTID (Global Transaction ID)是全局事务ID,当在主库上提交事务或者被从库应用时,可以定位和追踪每一个事务,对DBA来说意义就很大了,我们可以适当的解放出来,不用手工去可以找偏移量的值了,而是通过CHANGE MASTER TO MASTER_HOST='xxx', MASTER_AUTO_POSITION=1的即可方便的搭建从库,在故障修复中也可以采用MASTER_AUTO_POSITION=‘X’的方式。
陳斯托洛夫斯記
2022/10/27
5290
MySQL5.7基于GTID的主从配置
clickhouse同步mysql数据_clickhouse查询
大家好,我是架构君,一个会写代码吟诗的架构师。今天说一说clickhouse同步mysql数据_clickhouse查询,希望能够帮助大家进步!!!
Java架构师必看
2022/08/14
5.3K0
clickhouse同步mysql数据_clickhouse查询
使用Docker快速体验MySQL半同步复制机制
近期研究了下MySQL的半同步复制机制,想要体验一下。搭建环境是件麻烦事,然后就想到用Docker快速搭建环境。
Coder Sam
2018/09/14
3.1K4
MySQL传统点位复制在线转为GTID模式复制
MySQL传统点位复制在5.7版本前是主要的主从复制模式,而随着MySQL5.6版本引入GTID,并且MySQL5.7进行各方面的优化以后,在mySQL5.7(尤其是MySQL5.7.6)版本后GTID模式的主从复制方式成为一个新的选择方式。要使用GTID模式,首先也需知其优缺点,其主要的优缺点如下:
俊才
2019/11/18
2K0
MySQL传统点位复制在线转为GTID模式复制
MySQL基于GTIDs的MySQL Replication
1、GTIDs(Global transaction identifiers)全局事务标识符,是mysql 5.6新加入的一项技术
星哥玩云
2022/08/18
4190
ClickHouse王炸功能即将来袭?
众所周知,MySQL 的用户群体很大,为了能够增强数据的实时性,很多解决方案会利用 binlog 将数据写入到 ClickHouse。
Nauu
2020/07/31
2.7K1
MySQL 8.X 主从配置
Desc ✏️ 本文以基于GTID的复制示例 GTID(Global Transaction ID)是对于一个已提交事务的唯一编号,并且是一个全局(主从复制)唯一的编号。 它的官方定义如下: GTID = source_id :transaction_id 7E11FA47-31CA-19E1-9E56-C43AA21293967:29 什么是sever_uuid,和Server-id 区别? 核心特性: 全局唯一,具备幂等性 GTID核心参数 重要参数: gtid-mode=on
星哥玩云
2022/08/18
2300
Mysql之Mycat读写分离及分库分表
​ Mycat的原理中最重要的一个动词是“拦截”,它拦截了用户发送过来的SQL语句,首先对SQL语句做了一些特定的分析:如分片分析、路由分析、读写分离分析、缓存分析等, 然后将此SQL发往后端的真实数据库,并将返回的结果做适当的处理,最终再返回给用户。 如下图: ​ 上述图片里,Orders表被分为三个分片datanode(简称dn),这三个分片是分布在两台MySQL Server上(DataHost),即datanode=database@datahost方式, 因此你可以用一台到N台服务器来分片,分片规则为(sharding rule)典型的字符串枚举分片规则,一个规则的定义是分片字段(sharding column)+分片函数(rule function), ​ 这里的分片字段为prov而分片函数为字符串枚举方式。 ​ 当Mycat收到一个SQL时,会先解析这个SQL,查找涉及到的表,然后看此表的定义,如果有分片规则,则获取到SQL里分片字段的值,并匹配分片函数,得到该SQL对应的分片列表,
iginkgo18
2021/04/26
9500
Mysql之Mycat读写分离及分库分表
MYSQL无备份情况下恢复误删除的user权限表
前几天客户反馈,误删除了权限表,导致无法连接到实例中了,但是又没有备份,咨询要怎么去恢复;
SEian.G
2021/03/03
2.2K0
MySQL主从复制详解
在MySQL中,主从架构应该是最基础、最常用的一种架构了。后续的读写分离、多活高可用架构等大多都依赖于主从复制。主从复制也是我们学习MySQL过程中必不可少的一部分,关于主从复制的文章有很多,笔者也来凑凑热闹,写写这方面的内容吧,同时分享下自己的经验和方法。
MySQL技术
2020/07/28
4150
MySQL主从复制详解
MySQL 多主一从配置
MySQL 的多主一从是指,多个master机,单个slave的模式,是在MySQL5.7以后支持的
前Thoughtworks-杨焱
2021/12/07
2.1K1
mysqldump 快速搭建特定库主从架构(GTID)
相关知识点参考 基于mysqldump搭建gtid主从 MySQL GTID 错误处理汇总 配置MySQL GTID 主从复制 使用mysqldump导出数据库
Leshami
2018/08/08
1.5K0
MySql5.7-多源复制(多主单从) 原
#binlog format有三种形式:Statement、Mixed、Row,默认设置为mixed
拓荒者
2019/03/11
3.1K0
MySql5.7-多源复制(多主单从)
                                                                            原
Mysql主从复制
Mysql主从复制也可以称为Mysql主从同步,它是构建数据库高可用集群架构的基础。它通过将一台主机的数据复制到其他一台或者多台主机上,并重新应用日志(realy log)中的SQL语句来实现复制功能。Mysql支持单向,双向,链式级联,异步复制,复制过程中一台服务器充当主库(master),而一个或者多个服务器充当从库(slave)
mikelLam
2022/10/31
1.4K0
MySQL Fabric实验(三)HA与Sharding
实验步骤: 1. 安装虚拟机         使用VirtualBox安装四个CentOS release 6.4虚拟机,安装Python 2.6或以上版本,关闭iptables和selinux。虚拟机和网卡说明如下表所示。
用户1148526
2022/05/07
1.1K0
MySQL Fabric实验(三)HA与Sharding
mysql-MGR集群搭建
MGR是MySQL数据库未来发展的一个重要方向。 MGR基础结构要求: 引擎必须为innodb,因为需事务支持在commit时对各节点进行冲突检查 每个表必须有主键,在进行事务冲突检测时需要利用主键值对比 必须开启binlog且为row格式 开启GTID,且主从状态信息存于表中(--master-info-repository=TABLE 、--relay-log-info-repository=TABLE),--log-slave-updates打开 一致性检测设置--transaction-write-set-extraction=XXHASH64 MGR使用限制: RP和普通复制binlog校验不能共存,需设置--binlog-checksum=none 不支持gap lock(间隙锁),隔离级别需设置为read_committed 不支持对表进行锁操作(lock /unlock table),不会发送到其他节点执行 ,影响需要对表进行加锁操作的情况,列入mysqldump全表备份恢复操作 不支持serializable(序列化)隔离级别 DDL语句不支持原子性,不能检测冲突,执行后需自行校验是否一致 不支持外键:多主不支持,单主模式不存在此问题 最多支持9个节点:超过9台server无法加入组
章工运维
2023/08/24
4250
mysql-MGR集群搭建
相关推荐
ClickHouse之MaterializeMySQL引擎(十)
更多 >
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
本文部分代码块支持一键运行,欢迎体验
本文部分代码块支持一键运行,欢迎体验