about云开发

 找回密码
 立即注册

QQ登录

只需一步,快速开始

[连载型] 注册免费送体验金平台

[复制链接]
levycui 发表于 5 天前 | 显示全部楼层 |阅读模式
本帖最后由 levycui 于 2020-1-16 08:59 编辑

问题导读:
1、ES的写入基本流程有哪些过程?
2、如何理解bulk请求分发?
3、bulk写入流程有哪些?
4、如何理解协调节点处理并转发请求原理?



1 前言

Elasticsearch(ES)是一个基于Lucene的分布式存储和搜索分析系统,本文希望从源码的角度分析ES在保证数据的可靠性、实时性和一致性前提下,其写入的具体流程。

写入也是整个ES系统里面,最主要的流程之一,便于更好的理解ES的内部原理和逻辑,关于ES数据存储结构请参考:【Elasticsearch】原理-Elasticsearch数据存储结构与写入流程。

2 写入基本流程

    图片来自官网,源代码取自6.7.1版本:

2020-01-14_180358.jpg
ES的写入采用一主多副的模式,写操作一般会经过三种节点:协调节点、主分片所在节点、副本分片所在节点。
客户端发送请求到Node1(相当于协调节点),协调节点收到请求之后,确认写入的文档属于分片P0,于是将请求转发给P0所在的节点Node3,Node3写完成之后将请求转发到P0所属的副本R0所在的节点Node1和Node2。

什么时候给客户端返回成功呢?

特别注意: 取决于wait_for_active_shards参数:需要确认的分片数,默认为1,即主分片写入成功就返回客户端结果。
[Java] 纯文本查看 复制代码
    /**
     * The number of active shard copies to check for before proceeding with a write operation.
     */
    public static final Setting<ActiveShardCount> SETTING_WAIT_FOR_ACTIVE_SHARDS =
        new Setting<>("index.write.wait_for_active_shards",
                      "1",
                      ActiveShardCount::parseString,
                      Setting.Property.Dynamic,
                      Setting.Property.IndexScope);


以上是写入的大体流程,整个详细的流程,通过源码进行分析。

3 写入源码分析

ES的写入官方提供了两种写入方式:index,逐条写入;Bulk,批量写入。对于这两种方式,ES都会转化成Bulk写入。

3.1 bulk请求分发

ES的写入请求一般会进过两层处理,首先的Rest层(进行请求参数解析),另一层是Transport层(进行实际的请求处理)。在每一层处理前都有一次请求分发:
2020-01-14_180512.jpg

客户端发送过来的HTTP请求由HttpServerTransport初步处理后进入RestController模块进行实际的分发过程:
[Java] 纯文本查看 复制代码
    public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
        if (request.rawPath().equals("/favicon.ico")) {
            handleFavicon(request, channel);
            return;
        }
        try {
                //找出所有可能的handlers,然后分发这些请求
            tryAllHandlers(request, channel, threadContext);
        } catch (Exception e) {
          .......
        }
    }


上面dispatchRequest方法,会通过tryAllHandlers方法找出所有可能的handlers,并分发请求,代码如下:
[Java] 纯文本查看 复制代码
    void tryAllHandlers(final RestRequest request, final RestChannel channel, final ThreadContext threadContext) throws Exception {
        for (String key : headersToCopy) {
            String httpHeader = request.header(key);
            if (httpHeader != null) {
                threadContext.putHeader(key, httpHeader);
            }
        }
     .....
        // 获取所有可能的Handler,并尝试分发request请求
        Iterator<MethodHandlers> allHandlers = getAllHandlers(request);
        for (Iterator<MethodHandlers> it = allHandlers; it.hasNext(); ) {
            final Optional<RestHandler> mHandler = Optional.ofNullable(it.next()).flatMap(mh -> mh.getHandler(request.method()));
            //进行request请求分发,如果分发成功,则返回true
            requestHandled = dispatchRequest(request, channel, client, mHandler);
            if (requestHandled) {
                break;
            }
        }
        .....
    }


首先根据request找到其对应的handler,然后在dispatchRequest中调用handler的handleRequest方法处理请求。那么getHandler是如何根据请求找到对应的handler的呢?这块的逻辑如下:
[Java] 纯文本查看 复制代码
    Iterator<MethodHandlers> getAllHandlers(final RestRequest request) {
        final Map<String, String> originalParams = new HashMap<>(request.params());
        return handlers.retrieveAll(getPath(request), () -> {
            request.params().clear();
            request.params().putAll(originalParams);
            return request.params();
        });
    }
    public void registerHandler(RestRequest.Method method, String path, RestHandler handler) {
        if (handler instanceof BaseRestHandler) {
            usageService.addRestHandler((BaseRestHandler) handler);
        }
        handlers.insertOrUpdate(path, new MethodHandlers(path, handler, method), (mHandlers, newMHandler) -> {
            return mHandlers.addMethods(handler, method);
        });
    }


ES会通过RestController的registerHandler方法,提前把handler注册到对应http请求方法(GET、PUT、POST、DELETE等)的handlers列表。这样用户请求到达时,就可以通过RestController的getHandler方法,并根据http请求方法和路径取出对应的handler。对于bulk操作,其请求对应的handler是RestBulkAction,该类会在其构造函数中将其注册到RestController,代码如下:

[Java] 纯文本查看 复制代码
    public RestBulkAction(Settings settings, RestController controller) {
        super(settings);
        controller.registerHandler(POST, "/_bulk", this);
        controller.registerHandler(PUT, "/_bulk", this);
        controller.registerHandler(POST, "/{index}/_bulk", this);
        controller.registerHandler(PUT, "/{index}/_bulk", this);
        controller.registerHandler(POST, "/{index}/{type}/_bulk", this);
        controller.registerHandler(PUT, "/{index}/{type}/_bulk", this);
        this.allowExplicitIndex = MULTI_ALLOW_EXPLICIT_INDEX.get(settings);
    }


RestBulkAction会将RestRequest解析并转化为BulkRequest,然后再对BulkRequest做处理,这块的逻辑在prepareRequest方法中,部分代码如下:

[Java] 纯文本查看 复制代码
    public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
        //根据RestRequest构建bulkRequest
        ......
        //处理bulkRequest请求
        return channel -> client.bulk(bulkRequest, new RestStatusToXContentListener<>(channel));
    }


NodeClient在处理BulkRequest请求时,会将请求的action转化为对应Transport层的action,然后再由Transport层的action来处理BulkRequest,action转化的代码如下:


[Java] 纯文本查看 复制代码
    public <    Request extends ActionRequest,
                Response extends ActionResponse
            > Task executeLocally(GenericAction<Request, Response> action, Request request, TaskListener<Response> listener) {
        return transportAction(action).execute(request, listener);
    }
    private <    Request extends ActionRequest,
                Response extends ActionResponse
            > TransportAction<Request, Response> transportAction(GenericAction<Request, Response> action) {
        .....
        //actions是个action到transportAction的Map,这个映射关系是在节点启动时初始化的
        TransportAction<Request, Response> transportAction = actions.get(action);
        ......
        return transportAction;
    }        


然后进入TransportAction,TransportAction#execute(Request request, ActionListener listener) -> TransportAction#execute(Task task, Request request, ActionListener listener) -> TransportAction#proceed(Task task, String actionName, Request request, ActionListener listener)。TransportAction会调用一个请求过滤链来处理请求,如果相关的插件定义了对该action的过滤处理,则先会执行插件的处理逻辑,然后再进入TransportAction的处理逻辑,过滤链的处理逻辑如下:

[Java] 纯文本查看 复制代码
        public void proceed(Task task, String actionName, Request request, ActionListener<Response> listener) {
            int i = index.getAndIncrement();
            try {
                if (i < this.action.filters.length) {
                        //应用插件的逻辑
                    this.action.filters[i].[/i]apply(task, actionName, request, listener, this);
                } else if (i == this.action.filters.length) {
                        //执行TransportAction的逻辑
                    this.action.doExecute(task, request, listener);
                } else {
                   ......
                }
            } catch(Exception e) {
               .....
            }
        }


对于Bulk请求,这里的TransportAction对应的具体对象是TransportBulkAction的实例,到此,Rest层转化为Transport层的流程完成,下节将详细介绍TransportBulkAction的处理逻辑。

3.2 bulk写入流程

    代码入口:TransportBulkAction#doExecute(Task task, BulkRequest bulkRequest, ActionListener listener)。

3.2.1 pipeline预处理

首先判断bulk请求中是否指定了pipeline参数,则先使用相应的pipeline进行处理。如果本节点不具备预处理(Ingest)的资格,则将请求转发到有资格的节点。如果没有Ingest节点则继续往下走。

3.2.2 创建索引

判断是否需要创建索引,即needToChec方法,返回的autoCreateIndex的开关,默认是true,即自动创建索引是打开的;

[Java] 纯文本查看 复制代码
    boolean needToCheck() {
        return autoCreateIndex.needToCheck();
    }
    public static final Setting<AutoCreate> AUTO_CREATE_INDEX_SETTING =
        new Setting<>("action.auto_create_index", "true", AutoCreate::new, Property.NodeScope, Setting.Property.Dynamic);


如果自动创建索引已关闭,则直接准备下一步操作:

[Java] 纯文本查看 复制代码
 executeBulk(task, bulkRequest, startTime, listener, responses, emptyMap());


如果需要自动创建索引,则需要遍历bulk的所有index,然后检查index是否需要自动创建,对于不存在的index,则会加入到自动创建的集合中,然后会调用createIndex方法创建index。index的创建由master来把控,master会根据分片分配和均衡的算法来决定在哪些data node上创建index对应的shard,然后将信息同步到data node上,由data node来执行具体的创建动作。

[Java] 纯文本查看 复制代码
  // Step 1: 对bulkRequest进行过滤,获取所有的索引名。主要为opType和versionType,其中opType为索引操作类型,支持INDEX、CREATE,UPDATE,DELETE四种。DELETE请求如果索引不存在,不应该创建索引,除非external versioning正在使用。
            final Set<String> indices = bulkRequest.requests.stream()
                .filter(request -> request.opType() != DocWriteRequest.OpType.DELETE
                    || request.versionType() == VersionType.EXTERNAL
                    || request.versionType() == VersionType.EXTERNAL_GTE)
                .map(DocWriteRequest::index)
                .collect(Collectors.toSet());
// Step 2: 对各个索引进行检查,indicesThatCannotBeCreated用来存储无法创建索引的信息Map,autoCreateIndices用来存储可以自动创建索引的Set。
//索引是否可以正常自动创建,主要检查:1.是否存在该索引或别名(存在则无法创建);2.该索引是否被允许自动创建(二次检查,为了防止check信息丢失);3.动态mapping是否被禁用(如果被禁用,则无法创建);4.创建索引的匹配规则是否存在并可以正常匹配(如果表达式非空,且该索引无法匹配上,则无法创建)。
            final Map<String, IndexNotFoundException> indicesThatCannotBeCreated = new HashMap<>();
            Set<String> autoCreateIndices = new HashSet<>();
            ClusterState state = clusterService.state();
            for (String index : indices) {
                boolean shouldAutoCreate;
                try {
                    shouldAutoCreate = shouldAutoCreate(index, state);
                } catch (.....) { .....}
                if (shouldAutoCreate) {
                    autoCreateIndices.add(index);
                }
            }
// Step 3: 如果没有索引需要创建,直接executeBulk到下一步;如果存在需要创建的索引,则逐个创建索引,并监听结果,成功计数器减1.失败的话,将bulkRequest中对应的request的value值设置为null,计数器减1,当所有索引执行"创建索引"操作结束后,即计数器为0时,进入executeBulk。
            if (autoCreateIndices.isEmpty()) {
                executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
            } else {
                final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
                for (String index : autoCreateIndices) {
                    createIndex(index, bulkRequest.timeout(), new ActionListener<CreateIndexResponse>() {
                        @Override
                        public void onResponse(CreateIndexResponse result) {
                            if (counter.decrementAndGet() == 0) {
                                executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
                            }
                        }
                        @Override
                        public void onFailure(Exception e) {
                                                ..........
                    });
                }


3.2.3 协调节点处理并转发请求

创建完index之后,index的各shard已在数据节点上建立完成,接着协调节点将会转发写入请求到文档对应的primary shard。进入到BulkOperation#doRun中。
首先会检查集群无BlockException后(存在BlockedException会不断重试,直至超时),然后遍历BulkRequest的所有子请求,然后根据请求的操作类型生成相应的逻辑,对于写入请求,会首先根据IndexMetaData信息,resolveRouting方法为每条IndexRequest生成路由信息,并通过process方法按需生成doc id(不指定的话默认是UUID)。

[Java] 纯文本查看 复制代码
            for (int i = 0; i < bulkRequest.requests.size(); i++) {
                DocWriteRequest docWriteRequest = bulkRequest.requests.get(i);
                                .......
                Index concreteIndex = concreteIndices.resolveIfAbsent(docWriteRequest);
                try {
                    switch (docWriteRequest.opType()) {
                        case CREATE:
                        case INDEX:
                            .......
                            indexRequest.resolveRouting(metaData);
                            indexRequest.process(indexCreated, mappingMd, concreteIndex.getName());
                            break;
                                                        ......
                    }
                } catch (.......) {.......}
            }


然后根据每个IndexRequest请求的路由信息(如果写入时未指定路由,则es默认使用doc id作为路由)得到所要写入的目标shard id,并将DocWriteRequest封装为BulkItemRequest且添加到对应shardId的请求列表中。代码如下:

[Java] 纯文本查看 复制代码
                        //requestsByShard的key是shard id,value是对应的单个doc写入请求(会被封装成BulkItemRequest)的集合
            Map<ShardId, List<BulkItemRequest>> requestsByShard = new HashMap<>();
            for (int i = 0; i < bulkRequest.requests.size(); i++) {
                    //从bulk请求中得到每个doc写入请求
                DocWriteRequest request = bulkRequest.requests.get(i);
                ......
                String concreteIndex = concreteIndices.getConcreteIndex(request.index()).getName();
                //根据路由,找出doc写入的目标shard id
                ShardId shardId = clusterService.operationRouting().indexShards(clusterState, concreteIndex, request.id(),
                    request.routing()).shardId();
                List<BulkItemRequest> shardRequests = requestsByShard.computeIfAbsent(shardId, shard -> new ArrayList<>());
                shardRequests.add(new BulkItemRequest(i, request));
            }


计算ShardId的代码如下所示:这里的partitionOffset是根据参数index.routing_partition_size获取的,默认为1,写入时指定id,可能导致分布不均,可调大该参数,让分片id可变范围更大,分布更均匀。routingFactor默认为1,主要是在做spilt和shrink时改变。

[Java] 纯文本查看 复制代码
    private static int calculateScaledShardId(IndexMetaData indexMetaData, String effectiveRouting, int partitionOffset) {
        final int hash = Murmur3HashFunction.hash(effectiveRouting) + partitionOffset;
        return Math.floorMod(hash, indexMetaData.getRoutingNumShards()) / indexMetaData.getRoutingFactor();
    }


上一步已经找出每个shard及其所需执行的doc写入请求列表的对应关系,这里就相当于将请求按shard进行了拆分,接下来会将每个shard对应的所有请求封装为BulkShardRequest并交由TransportShardBulkAction来处理:即将相同shard id的请求合并,并转发TransportShardBulkAction请求。

[Java] 纯文本查看 复制代码
            for (Map.Entry<ShardId, List<BulkItemRequest>> entry : requestsByShard.entrySet()) {
                final ShardId shardId = entry.getKey();
                final List<BulkItemRequest> requests = entry.getValue();
                // 对每个shard id及对应的BulkItemRequest集合,合并为一个BulkShardRequest
                BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, bulkRequest.getRefreshPolicy(),
                    requests.toArray(new BulkItemRequest[requests.size()]));
                ......
                if (task != null) {
                    bulkShardRequest.setParentTask(nodeId, task.getId());
                // 处理请求(在listener中等待响应,响应都是按shard返回的,如果一个shard中有部分请求失败,将异常填到response中,所有请求完成,即计数器为0,调用finishHim(),整体请求做成功处理):
                shardBulkAction.execute(bulkShardRequest, new ActionListener<BulkShardResponse>() {
                                 ........
                });
            }


3.2.4 向主分片发送请求

转发TransportShardBulkAction请求,最后进入到TransportReplicationAction#doExecute方法,然后进入到TransportReplicationAction.ReroutePhase#doRun方法。这里会通过ClusterState获取到primary shard的路由信息,然后得到primay shard所在的node,如果node为当前协调节点则直接将请求发往本地,否则发往远端:

[Java] 纯文本查看 复制代码
            setPhase(task, "routing"); //标识为routing阶段
            final ClusterState state = observer.setAndGetObservedState();
            .......
            } else {
                // 获取主分片所在的shard路由信息,得到主分片所在的node节点
                final IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
                .........
                final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
                if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
                        //是当前节点,继续执行
                    performLocalAction(state, primary, node, indexMetaData);
                } else {
                        //不是当前节点,转发到对应的node上进行处理
                    performRemoteAction(state, primary, node);
                }
            }


如果分片在当前节点,task当前阶段置为“waiting_on_primary”,否则为“rerouted”,两者都走到同一入口,即performAction(…), 在performAction方法中,会调用TransportService的sendRequest方法,将请求发送出去。
如果对端返回异常,比如对端节点故障或者primary shard挂了,对于这些异常,协调节点会有重试机制,重试的逻辑为等待获取最新的集群状态,然后再根据集群的最新状态(通过集群状态可以拿到新的primary shard信息)重新执行上面的doRun逻辑;如果在等待集群状态更新时超时,则会执行最后一次重试操作(执行doRun)。这块的代码如下:

[Java] 纯文本查看 复制代码
        void retry(Exception failure) {
            if (observer.isTimedOut()) {
                // 超时时已经做过最后一次尝试,这里将不再重试,超时默认1min
                finishAsFailed(failure);
                return;
            }
            setPhase(task, "waiting_for_retry");
            request.onRetry();
            observer.waitForNextChange(new ClusterStateObserver.Listener() {
                @Override
                public void onNewClusterState(ClusterState state) {
                    run(); //会调用doRun
                }
                .......
                @Override
                public void onTimeout(TimeValue timeout) { //超时,做最后一次重试
                    // Try one more time...
                    run(); //会调用doRun
                }
            });
        }



下一篇:Elasticsearch源码:写入流程原理分析(二)

作者:少加点香菜
来源:https://blog.csdn.net/wudingmei1023/article/details/103934342
最新经典文章,欢迎关注公众号


您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /4 下一条

QQ|小黑屋|about云开发-学问论坛|社区 ( 京ICP备12023829号 )

GMT+8, 2020-1-19 10:11 , Processed in 1.125000 second(s), 29 queries , Gzip On.

Powered by Discuz! X3.4 Licensed

© 2018 Comsenz Inc.Designed by u179

快速回复 返回顶部 返回列表