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

- 日理万妓 2021-11-11 17:28 422阅读 0赞

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

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

## 1.1  了解\_version是什么 ##

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

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

返回结果：

{
      "_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

DELETE /test_index/test_type/6

结果：

{
      "_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

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

结果：

{
      "_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乐观锁实质 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70][]

假设无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数据

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

{
      "_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客户端，或打开两个命令行，模拟两个客户端，都获取到了同一条数据

GET test_index/test_type/8

{
      "_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中的数据的版本号，跟客户端中的数据的版本号是相同的，才能修改

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

{
      "_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版本号，进行乐观锁的并发控制

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

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

GET /test_index/test_type/8

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

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

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

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

{
      "_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来进行控制。

语法比较：

?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）先构造一条数据

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

{
      "_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）模拟两个客户端同时查询到这条数据

GET /test_index/test_type/9

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

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

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

{
      "_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发起了修改

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

{
      "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）在并发控制成功后，重新基于最新的版本号发起更新

GET /test_index/test_type/9

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

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

{
      "_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
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2l0Y2F0c19jbg_size_16_color_FFFFFF_t_70]: /images/20211109/c11dd78a532340fab2873a0e0d5f4e34.png