ElasticSearch进阶篇(一)--版本控制

2023-12-09 16:26:39

一、前言

ElasticSearch（以下简称ES）的数据写入支持高并发，高并发就会带来很普遍的数据一致性问题。常见的解决方法就是加锁。同样，ES为了保证高并发写的数据一致性问题，加入了类似于锁的实现方法--版本控制。锁从其中的一个角度可分为乐观锁和悲观锁。

对于同一个数据的并发操作，悲观锁认为自己在使用数据的时候一定会有别的线程过来修改数据，因此在获取数据的时候会先加锁，确保数据不会被别的线程修改。而乐观锁则认为自己在使用数据时不会有别的线程来修改数据，所以不会添加锁，只是在更新或者提交数据的时候去判断之前有没有别的线程更新了这个数据。那么ES属于那种锁呢？下面大狮兄就和大家一起探讨官方的具体做法来回答这个问题。

二、版本控制实现及验证

1. ES6.7 Before

# 新建测试索引
PUT test
{
  "settings" : {
    "number_of_shards" : "3",
    "number_of_replicas" : "0"
  }
}

## 插入文档
PUT test/_doc/1 
{"user": "zhangsan", "age": 12}

## 响应结果
{
  "_index" : "test",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

更新文档（version版本大于已写入文档版本），更新年龄为10，版本号为200

## 更新文档
PUT test/_doc/1?version=200&version_type=external
{"user": "zhangsan", "age": 10}

## 返回结果
{
  "_index" : "test",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 200,
  "result" : "updated",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

## 查询文档
GET test/_doc/1
## 返回结果
{
  "_index" : "test",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 200,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "user" : "zhangsan",
    "age" : 10
  }
}

更新成功，年龄更新为10且版本号更新为200

更新文档（version版本小于或等于已写入文档版本），更新年龄为22，版本号为180

## 更新文档
PUT test/_doc/1?version=180&version_type=external
{"user": "zhangsan", "age": 22}

## 返回结果
{
  "error" : {
    "root_cause" : [
      {
        "type" : "version_conflict_engine_exception",
        "reason" : "[1]: version conflict, current version [200] is higher or equal to the one provided [180]",
        "index_uuid" : "fCv7Q1dkTl6e9E1Z0dNE1g",
        "shard" : "2",
        "index" : "test"
      }
    ],
    "type" : "version_conflict_engine_exception",
    "reason" : "[1]: version conflict, current version [200] is higher or equal to the one provided [180]",
    "index_uuid" : "fCv7Q1dkTl6e9E1Z0dNE1g",
    "shard" : "2",
    "index" : "test"
  },
  "status" : 409
}

## 查询文档
GET test/_doc/1

## 返回结果
{
  "_index" : "test",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 200,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "user" : "zhangsan",
    "age" : 10
  }
}

更新失败，数据没有变化，提示版本冲突，现有的版本号大于要插入的版本号。

  • vertion_type=external 或者 vertion_type=external_gt ：目标版本号大于已有的版本号才会更新成功。
  • vertion_type=external_gte ：目标版本号大于或等于已有的版本号才会更新成功。

2. ES6.7 OR Later

# 新建测试索引
PUT testccc
{
  "settings" : {
    "number_of_shards" : "1",
    "number_of_replicas" : "0"
  }
}


## 插入文档
PUT testccc/_doc/1 
{"user": "lisi", "age": 12}

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 0,
  "_primary_term" : 1
}

返回结果注意最后的两个字段，_seq_no表示序列号是自增的，_primary_term表是文档位于哪个shard。

更新数据（seq_no大于已写入文档序列号），更新年龄为10，序列号为20

## 更新文档
PUT testccc/_doc/1?if_seq_no=20&if_primary_term=1
{"user": "lisi", "age": 10}

## 返回结果
{
  "error" : {
    "root_cause" : [
      {
        "type" : "version_conflict_engine_exception",
        "reason" : "[1]: version conflict, required seqNo [20], primary term [1]. current document has seqNo [0] and primary term [1]",
        "index_uuid" : "N6LzBNj9S5yqVWFubt3x4Q",
        "shard" : "0",
        "index" : "testccc"
      }
    ],
    "type" : "version_conflict_engine_exception",
    "reason" : "[1]: version conflict, required seqNo [20], primary term [1]. current document has seqNo [0] and primary term [1]",
    "index_uuid" : "N6LzBNj9S5yqVWFubt3x4Q",
    "shard" : "0",
    "index" : "testccc"
  },
  "status" : 409
}


## 查询文档
GET testccc/_doc/1 

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "user" : "lisi",
    "age" : 12
  }
}

更新失败，数据无变化，提示版本冲突，最近文档的序列号为0，要更新的序列号为20。

更新数据（seq_no等于已写入文档序列号），更新年龄为10

## 更新文档
PUT testccc/_doc/1?if_seq_no=0&if_primary_term=1
{"user": "lisi", "age": 10}

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

## 查询文档
GET testccc/_doc/1 
## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 2,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "user" : "lisi",
    "age" : 10
  }
}

更新成功，且seq_no自增为1。

## 插入新文档
PUT testccc/_doc/2
{"user": "wangwu", "age": 40}

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "2",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 2,
  "_primary_term" : 1
}

## 更新原文档
PUT testccc/_doc/1?if_seq_no=1&if_primary_term=1
{"user": "lisi", "age": 50}

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 3,
  "result" : "updated",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 3,
  "_primary_term" : 1
}

## 更新新写入文档
PUT testccc/_doc/2?if_seq_no=2&if_primary_term=1
{"user": "wangwu", "age": 80}

## 返回结果
{
  "_index" : "testccc",
  "_type" : "_doc",
  "_id" : "2",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 4,
  "_primary_term" : 1
}

可以观察到对于不同的文档，seq_no总是自增1的。

三、总结

  1. ES版本控制类似于Java中的乐观锁，尤其对版本号字段的巧妙使用与解决乐观锁ABA问题的CAS算法有异曲同工之妙。
  2. ES6.7之后添加的if_seq_no与if_primary_term版本控制是针对于整个索引的，而_version和version_type版本控制是针对于单条记录（即单个文档）的，不同的应用场景可使用不同的版本控制策略。
  3. if_seq_no配置的值必须等于存在于现有文档中才能更新成功，而_version配置的值根据不同的version_type，必须大于或者大于等于文档最近更改过的_version值才能更新成功。
上一篇：序列化与反序列化


下一篇：[LeetCode] 457. Circular Array Loop