ElasticSearch基于version进行乐观锁并发控制实战解析

一、ES解决并发问题的实质

ES在多线程异步修改数据时，是根据_version(版本号)来进行乐观锁控制的。

1.1  了解_version是什么

1、在第一次创建document的时候，该document的_version版本号为1，每次对document进行修改、删除操作，document的_version版本号加1

代码语言：javascript复制

PUT /test_index/test_type/6
{
  "test_field": "test test"
}

返回结果：

代码语言：javascript复制

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

2、删除这条document

代码语言：javascript复制

DELETE /test_index/test_type/6

结果：

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "6",
  "_version" : 2,
  "result" : "deleted",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

3、重新创建这条document

代码语言：javascript复制

PUT /test_index/test_type/6
{
  "test_field": "test test"
}

结果：

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "6",
  "_version" : 3,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 2,
  "_primary_term" : 1
}

结论：

在删除命令后，该document并不是立即物理删除掉的，它的一些版本号等信息还是保留着的。先删除一条document，再重新创建这条document，其实会在delete version基础之上，再把version号加1。

1.2  图解ES乐观锁实质

假设无ES乐观锁控制的场景

①创建一条document，假设field1 = text1  

②线程1修改数据field1 = text2   (先修改)

③线程2修改数据field1 = text3  (后修改)

④ES内部Primary Shard的数据会同步到Replica Shard中，类似于Replica Shard的同步请求，都是多线程异步的，也就是说多个修改请求之间是乱序的，有可能存在后修改的被先修改，先修改的被后修改等问题。

如果field1 = text3先修改，不进行_version的版本控制，field1 = text2会直接覆盖，此时数据出现错误。正确的逻辑应该是先修改的先修改，后修改的后修改，此时就引出了ES乐观锁控制并发的问题。

ES乐观锁控制并发的场景

①创建一条document，假设field1 = text1    _version = 1

②线程1修改数据field1 = text2   (先修改)     获取到修改前的_version=1

③线程2修改数据field1 = text3  (后修改)     获取到修改前的_version=2

④假设后修改的数据先到，field1 = text3 ，此时_version=2。

    此时先修改的数据后到，field1 = text2， 获取到修改前的_version=2，自己的_version=1，不相同，将低版本的数据忽略，保持field1 = text3，通过版本控制不会让旧数据覆盖新数据。

二、实战模拟ES基于_version并发修改的原理

1、创建一条document数据

代码语言：javascript复制

PUT /test_index/test_type/8
{
  "test_field": "test test"
}

代码语言：javascript复制

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

2、打开两个Kibana客户端，或打开两个命令行，模拟两个客户端，都获取到了同一条数据

代码语言：javascript复制

GET test_index/test_type/8

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "8",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "test_field" : "test test"
  }
}

3、其中一个客户端，更新该document，同时带上数据的版本号，确保es中的数据的版本号，跟客户端中的数据的版本号是相同的，才能修改

代码语言：javascript复制

PUT /test_index/test_type/8?version=1
{
  "test_field": "test client 1"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "8",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

4、另外一个客户端，尝试基于version=1的数据去进行修改，同样带上version版本号，进行乐观锁的并发控制

代码语言：javascript复制

PUT /test_index/test_type/7?version=1 
{
  "test_field": "test client 2"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "8",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 1,
  "_primary_term" : 1
}

5、在乐观锁成功阻止并发问题之后，尝试正确的完成更新

代码语言：javascript复制

GET /test_index/test_type/8

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "8",
  "_version" : 2,
  "_seq_no" : 1,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "test_field" : "test client 1"
  }
}

基于最新的数据和版本号进行修改，修改后，带上最新的版本号，可能这个步骤会需要反复执行好几次，才能成功，特别是在多线程并发更新同一条数据很频繁的情况下。

尝试再次修改，直到修改成功：

代码语言：javascript复制

PUT /test_index/test_type/8?version=2
{
  "test_field": "test client 2"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "8",
  "_version" : 3,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 2,
  "_primary_term" : 1
}

三、实战模拟ES基于version external并发修改的原理

es提供了一个feature，可以不用它提供的内部_version版本号来进行并发控制，可以基于自己维护的一个版本号来进行并发控制。举个列子，假设数据在mysql里也有一份，然后应用系统本身就维护了一个版本号，无论是什么自己生成的，程序控制的。这个时候，在进行乐观锁并发控制的时候，可能并不是想要用es内部的_version来进行控制，而是用你自己维护的那个version来进行控制。

语法比较：

代码语言：javascript复制

?version=1
?version=1&version_type=external

version_type=external，唯一的区别在于，_version，只有当提供的version与es中的_version一模一样的时候，才可以进行修改，只要不一样，就报错；当version_type=external的时候，只有当提供的version比es中的_version大的时候，才能完成修改。

es，_version=1，?version=1，才能更新成功 es，_version=1，?version>1&version_type=external，才能成功，比如说version=2&version_type=external

（1）先构造一条数据

代码语言：javascript复制

PUT /test_index/test_type/9
{
  "test_field": "test"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "9",
  "_version" : 1,
  "result" : "created",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 3,
  "_primary_term" : 1
}

（2）模拟两个客户端同时查询到这条数据

代码语言：javascript复制

GET /test_index/test_type/9

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "9",
  "_version" : 1,
  "_seq_no" : 3,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "test_field" : "test"
  }
}

（3）第一个客户端先进行修改，此时客户端程序是在自己的数据库中获取到了这条数据的最新版本号，比如说是2

代码语言：javascript复制

PUT /test_index/test_type/9?version=2&version_type=external
{
  "test_field": "test client 1"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "9",
  "_version" : 2,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 4,
  "_primary_term" : 1
}

（4）模拟第二个客户端，同时拿到了自己数据库中维护的那个版本号，也是2，同时基于version=2发起了修改

代码语言：javascript复制

PUT /test_index/test_type/9?version=2&version_type=external
{
  "test_field": "test client 2"
}

代码语言：javascript复制

{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[test_type][9]: version conflict, current version [2] is higher or equal to the one provided [2]",
        "index_uuid": "h99RQHS7Si6VxZXsQnzTOA",
        "shard": "1",
        "index": "test_index"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[test_type][9]: version conflict, current version [2] is higher or equal to the one provided [2]",
    "index_uuid": "h99RQHS7Si6VxZXsQnzTOA",
    "shard": "1",
    "index": "test_index"
  },
  "status": 409
}

（5）在并发控制成功后，重新基于最新的版本号发起更新

代码语言：javascript复制

GET /test_index/test_type/9

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "9",
  "_version" : 2,
  "_seq_no" : 4,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "test_field" : "test client 1"
  }
}

代码语言：javascript复制

PUT /test_index/test_type/9?version=3&version_type=external
{
  "test_field": "test client 2"
}

代码语言：javascript复制

{
  "_index" : "test_index",
  "_type" : "test_type",
  "_id" : "9",
  "_version" : 3,
  "result" : "updated",
  "_shards" : {
    "total" : 2,
    "successful" : 1,
    "failed" : 0
  },
  "_seq_no" : 5,
  "_primary_term" : 1
}

es 数据库 sql 编程算法

0 人点赞