Elasticsearch Document Index API详解、原理与示例

时间 2020-12-04

标签 java 算法数据库 json 服务器 restful 并发 app 异步 elasticsearch 栏目日志分析繁體版

原文原文链接

本节将开始介绍Document API，本节将重点介绍ElasticSearch Doucment Index API(新增索引)。
从上节可知，ElasticSearch Index Rest Hign level Index Api声明以下：java

public final IndexResponse index(IndexRequest indexRequest, RequestOptions options) throws IOException
public final void indexAsync(IndexRequest indexRequest, RequestOptions options，ActionListener<IndexResponse> listener)
上述两个API,一个同步调用，一个异步调用，同步调用方法直接组装IndexResponse并返回，而异步方法经过回调ActionListener，并将执行结果(IndexResponse)传入回调方法。从中能够看出，Index API的核心是IndexRequest与RequestOptions。接下来咱们重点关注这两个对象。
一、IndexRequest 详解
IndexRequest的完整类继承层次以下：

咱们目前重点关注与索引相关的IndexRequest、ReplicatedWriteRequest、ReplicationRequest便可。

接下来对各个属性作个简单介绍（相关的内部实现机制将会在源码分析篇重点深刻分析，目前点到为止）。
protected ShardId shardId：目标执行的分片信息。（指定该名字哪些分片上执行）
protected ActiveShardCount waitForActiveShards = ActiveShardCount.DEFAULT，设置处理索引操做以前必须处于激活状态到的副本数量。
private long routedBasedOnClusterVersion = 0 请求基于的集群路由版本号，若是该版本后大于当前集群版本，则拒绝该请求。
private RefreshPolicy refreshPolicy = RefreshPolicy.NONE 索引刷新机制，该部分将在下文详细分析。
protected String index：索引库，相似关系型数据库的database。
private String type：类型名，相似于关系数据库的table(表)。
private String id ：文档ID，所谓的文档，相似于关系数据库的行，id,相似于关系数据库的主键ID。
private String routing：分片值，默认为id的值，elasticsearch的分片路由算法为( hashcode(routing) % primary_sharding_count(主分片个数) )。
private String parent：暂未知（与父子任务相关）
private BytesReference source：source，document的原始数据（被索引的原始数据，有效载荷）。
private OpType opType = OpType.INDEX：操做类型，例如INDEX、CREATE、UPDATE、DELETE。
private long version = Versions.MATCH_ANY：数据版本。
private VersionType versionType = VersionType.INTERNAL：版本类型，分为内部版本、外部版本，默认为内部版本。
private XContentType contentType：source的数据contentType，主要包含(XContentType.JSON、XContentType.SMILE、XContentType.YAML、XContentType.CBOR)，默认为XContentType.JSON。
private String pipeline：管道，暂未知。
private long autoGeneratedTimestamp = UNSET_AUTO_GENERATED_TIMESTAMP 利用基于事件戳建立自增ID。
private boolean isRetry = false 是否重试，默认为false。
二、RequestOption详解

RequestOptions，其实就是与Http相关的请求参数，http request header，由于Rest Hign Level Client其本质是Http请求。

三、Index BytesReference source构造详解
下面是4中构建JSON document的4种形式：算法

java的json字符串的byte[]或json字符串
java.util.Map
使用第三方JSON类库构建json字符串或其byte[]。
使用Elasticsearch自身提供的XContentFactory.jsonBuilder()类库。
3.1 json字符串

String json = "{" +
                    "\"user\":\"kimchy\"," +
                    "\"postDate\":\"2013-01-30\"," +
                    "\"message\":\"trying out Elasticsearch\"" +
            "}";
IndexResponse response = client.prepareIndex("twitter", "tweet")
    .setSource(json, XContentType.JSON)
    .get();

3.2 Map数据库

Map<String, Object> json = new HashMap<String, Object>();
json.put("user","kimchy");
json.put("postDate",new Date());
json.put("message","trying out Elasticsearch");
IndexResponse response = client.prepareIndex("twitter", "tweet")
        .setSource(json)
        .get();

3.3 使用第三方类库json

import com.fasterxml.jackson.databind.*;
// instance a json mapper
ObjectMapper mapper = new ObjectMapper(); // create once, reuse
// generate json
byte[] json = mapper.writeValueAsBytes(yourbeaninstance);
IndexResponse response = client.prepareIndex("twitter", "tweet")
        .setSource(json, XContentType.JSON)
        .get();

3.四、使用ElasticSearch自带类库服务器

import static org.elasticsearch.common.xcontent.XContentFactory.*;
IndexResponse response = client.prepareIndex("twitter", "tweet", "1")
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "kimchy")
                        .field("postDate", new Date())
                        .field("message", "trying out Elasticsearch")
                    .endObject()
                  )
        .get();

四、Index API使用Demorestful

Demo在单机版本的ElasticSearch服务器中运行。并发

package persistent.prestige.elasticsearchdemo;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
public class IndexApiDemo {
    public static void main(String[] args) {
        RestHighLevelClient client = EsClient.getClient();
        try {
            IndexRequest request = new IndexRequest();
            request.index("twitter");
            request.type("_doc");
            request.id("1");
            Map<String, String> source = new HashMap<>();
            source.put("user", "dingw");
            source.put("post_date", "2009-11-16T14:12:12");
            source.put("message", "trying out Elasticsearch");
            request.source(source);
            try {
                IndexResponse result = client.index(request, RequestOptions.DEFAULT);
                System.out.println(result);
            } catch (IOException e) {
                e.printStackTrace();
            }
        } finally {
            EsClient.close(client);
        }
    }
}

运行结果以下：app

{
    "_shards" : {
        "total" : 2,
        "failed" : 0,
        "successful" : 1
    },
    "_index" : "twitter",
    "_type" : "_doc",
    "_id" : "1",
    "_version" : 1,
    "_seq_no" : 0,
    "_primary_term" : 1,
    "result" : "created"
}

接下来的内容将详细介绍Index API返回结果相关的扩展知识，让你们更加全面的了解Index API内部运行的机制。异步

五、Index API 内部实现机制
5.1 _shards 返回字段概述
_shards 结构体将反馈索引在副本级的复制信息。elasticsearch

total：表示本次请求应该在多少个分片上执行（包含主分片 + 副本）。
successful：表示本次请求成功执行的分片个数。
failed：表示本次请求为成功执行请求的分片个数。

注：索引操做成功的标志是successful大于0。当索引操做成功返回时，复制分片（副本）可能不会所有启动（默认状况下，只有主服务器是必需的，可是这种行为能够被更改）。
当前单机环境，total为2表示，一个分片存在1主一从，但同一个复制组内的分片不会分布在同一个机器上，故只启动了主分片，复制分片未启动；successful为1表示在主分片上已成功执行，failed为0表示没有执行失败的分片。

5.2 自动建立索引
使用Index API，若是索引不存在，则会自动建立对应的索引（类型映射类型为动态映射机制，具体关于字段映射，将会在Mapping章节中详细介绍）。Elasticsearch数据的组织形式为(index/type/document)。索引的管理（增删改查等API在后续文中会描述）。
自动索引建立能够经过配置来禁用。经过在全部节点的配置文件中添加action.auto_create_index=false来禁用。经过配置index.mapper.dynamic=false能够禁用索引的映射自动建立。配置是否禁用自动建立索引可基于模式的白名单/黑名单列表模式，例如action.auto_create_index=aaa,-bbb,+ccc,- 分别表明 aaa开头的索引自动建立，bbb开头的索引禁止自动建立，禁用索引自动建立。

5.3 版本工做机制
每一个索引文档都有一个版本号。关联的版本号做为对索引API请求的响应的一部分返回。索引请求若是指定了版本号这个参数（IndexRequest#version)时，索引API可选择性地容许乐观并发控制机制，所谓乐观并发控制就是若是待操做的索引文档的版本号若是与IndexRequest#version版本不相同，则本次操做失败。版本控制彻底是实时的，若是未提供版本，则无需验证版本信息而当即执行。
默认状况下使用内部版本控制，从1开始，每次更新自增1,（包含删除）。可选地，版本号能够用外部值来补充（例如，若是在数据库中维护）。为了启用这个功能，IndexRequest#versionType应该被设置为外部（VersionType.EXTERNAL或VersionType.EXTERNAL_GTE）。外部版本号的取值范围为[0,9.2 e+18)。若是使用外部版本号，系统会检查传递给索引请求的版本号是否大于当前存储文档的版本号，而不是检查匹配的版本号。若是所提供的值小于或等于存储文档的版本号，则会出现版本冲突，索引操做将失败。
警告：外部版本控制支持0做为有效版本号。这容许版本与外部版本控制系统同步，其中版本号从0开始，而不是1。它有一个反作用，即版本号为零的文档不能使用更新的查询API进行更新，也不能使用查询API的Delete来删除，只要它们的版本号等于零。
外部版本号一个最佳实践，使用源数据库中数据的版本号，就不须要维护对源数据库的更改所执行的异步索引操做的严格排序。即便使用来自数据库的数据来更新Elasticsearch索引的简单状况，若是使用外部版本控制，也会简化，由于若是索引操做出于某种缘由而不正常，则只使用最新的版本便可。

5.4 版本类型
ElasticSearch支持以下版本类型：

internal
内部版本号，只有当请求版本号与数据版本号想等时才能够执行对应的动做。
external or external_gt
默认外部版本号，当请求版本号大于数据存储版本号时才能够执行对应动做，若是数据不存在，则使用指定版本号。
external_gte
外部版本号，当请求版本号大于等于数据存储版本号时能够执行对应动做，若是数据不存在，则使用指定版本号。
注意，外部版本号一般基于数据库，其思想更是基于乐观锁，对于版本号相等的更新动做须要特别谨慎，故外部版本号一般建议external（ external_gt）。外部版本号的取值范围[0,9.2 e+18)。
5.5 操做类型(IndexRequest#opType)
索引API也接受一个opType，它能够用来强制建立动做。当使用create时，若是该id中的文档已经存在于索引中，索引操做将会败。
OpType以下可选值：
OpType.INDEX
索引，若是文档已存在，覆盖，内部版本号+1。
OpType.CREATE
建立，若是文档已存在，返回错误。
OpType.UPDATE
更新操做。
OpType.DELETE
删除操做。

5.6 自动ID生成
索引动做能够不指定文档ID，ElasticSearch会自动建立ID,此时的opType属性会自动设置为OpType.CREATE。其Restfull请求又原先的PUT变动为POST，固然咱们在使用Rest Hign Level API时无需关注restfull请求类型，都是经过index方法发生调用，内部会自动封装相应的http请求。

5.7 路由
默认状况下，路由字段是经过使用文档的id值的散列来控制的，其路由算法(hash(路由字段) % (primary count))来定位所在的主分片（复制组）。ElasticSearch提供了显示指定路由字段的方法，经过routing来指定路由值，索引API经过IndexRequest#routing()方法来指定路由值。
当设置显式映射(Mapping)时，能够选择使用路由字段来指导索引操做从文档自己提取路由值。若是路由映射被定义并设置为required，那么若是没有提供或提取路由值，则索引操做将失败。

5.8 分布式
索引操做首先根据路由规则将请求转发到主分片，并在包含此分片的的实际节点上执行。在主分片完成操做以后，若是须要，更新将被分发到对应复制组中的副本所在的节点上执行。其执行逻辑已在上篇《Elasticsearch Document API之文档读写概要设计》中写模型一节中详细介绍，在此不重复介绍。

5.9 等待活动的分片数(Wait For Active Shards )
为了提升对系统写操做的弹性，引入了(wait for active shards)机制，就是在进行索引操做以前，先校验当前活跃的分片（副本数量），若是当前的激活分片数量不足，则先等待更多的分片启动直到有新的分片加入或等待超时。默认状况下，写操做只需等待主分片处于激活状态便可（即wait_for_active_shards=一、请求参数waitForActiveShards=1）。这个默认值能够经过设置索引的配置(Setting)index.write.wait_for_active_shards来动态改变。一样也能够经过请求参数IndexRequest#waitForActiveShards(waitForActiveShards)来动态改变。
wait_for_active_shards的数据类型为正整数，取值范围为[1,number_of_replicas+1]。
例如，假设咱们有一个由三个节点组成的集群，A、B和C，咱们建立一个索引，其中的副本数量（number_of_replicas）设置为3（3个副本+1个主分片，比节点数量多一个）。若是咱们尝试索引操做，默认状况下，只要主节点处于激活，则索引操做会在主节点上执行，而后转发到其余复制组。这意味着，即便B和C宕机(主分片在A节点上)，索引操做仍然会在主分片上执行。若是wait_for_active_shards设置为3（而且全部3个节点都正常），索引操做能继续执行而无需等待，由于集群中有3个活动节点，每一个节点都持有分片的副本。可是，若是咱们将wait_for_active_shards设置为all或4，索引操做将不会继续，陷入等待，由于咱们目前集群中没有4个副本。除非集群中出现一个新的节点来承载第4个副本，不然该操做将超时。
须要注意的是，这种设置大大减小了没必要要的写操做（能避免无谓的写处理，若是分片数量不足，则不执行索引动做），可是它并无彻底消除这种可能性，由于这种检查发生在写操做开始以前。一旦写操做开始，复制在任意数量的碎片副本上仍然可能失败，可是在主服务器上仍然成功。写操做响应的分片部分（5.1节所示）揭示了复制成功/失败的分片副本的数量，数据在主分片、副本之间数据的最终一致性处理在《Elasticsearch Document API之文档读写概要设计》写模型异常处理部分有相应的处理机制。
ActiveShardCount取值常量说明：

public static final ActiveShardCount DEFAULT = new ActiveShardCount(-2)：默认，只需主分片激活便可。
public static final ActiveShardCount ALL = new ActiveShardCount(-1)：所有，number_of_replicas+1 个副本都在线。
public static final ActiveShardCount NONE = new ActiveShardCount(0) ：不校验。
public static final ActiveShardCount ONE = new ActiveShardCount(1)：一个。
int 正整数,取值范围为[1,number_of_replicas+1]。

5.10 刷新机制
Index API 、Update API、Delete API、Bulk API都支持RefreshPolicy设置刷新策略（private RefreshPolicy refreshPolicy），以便控制上述API所产生的变化对查询API的可见性策略。其可选值以下：

空字符串或true(RefreshPolicy.IMMEDIATE)
在操做(index,update,delete)发生以后，当即刷新相关的主分片与复制分片（不是刷新整个索引，只是刷新发生变化的文档）。目前从索引与查询的角度来看，他不会致使性能低下。
false(RefreshPolicy.NONE)
在操做(index,update,delete)执行完毕后，直接返回，而不执行刷新，而是依靠Elasticsearch的刷新机制。
wait_for(RefreshPolicy.WAIT_UNTIL)
操做发生后，并不当即强制刷新，而是等待刷新的发生，此时会阻塞等待直到超时或刷新事件到达。Elasticsearch会以一个固定的频率刷新那些发生了变化的索引分片，刷新周期默认为1s，经过参数index.refresh_interval配置。
按照官方的建议，该参数不建议对其进行修改，保持默认值便可。一般不建议使用wait_for，目前没有明显的证据显示RefreshPolicy.IMMEDIATE会带来明显的性能损耗。

5.11 超时
当执行索引操做时，主分片所在的节点可能不可用。形成这种状况的一些缘由多是，主分片目前正在从网关中恢复或正在进行从新安置。默认状况下，索引操做将在主上等待最多1分钟，而后失败并以错误响应。超时参数能够用来显式地指定它等待多长时间，可经过IndexRequest#timeout(timeout)方法设置，或经过?timeout=5m设置。

总结：本文首先罗列了Elasticsearch Index API，而后详细介绍了其API两个核心的对象(IndexRequest与RequestOptions)，接着经过示例演示了RestHighLevelClient index API的使用，最后深刻分析了Index API的一些内在处理机制。后续会更深一步从源码角度深度剖析其实现细节。

关注《中间件兴趣圈》公众号，查看更多关于中间件相关的文章：