Elasticsearch系統學習（七）-ES並發控制

一、ES並發控制原理

1.1、ES並發沖突問題

1.2、悲觀鎖與樂觀鎖並發控制圖解

ES內部基於_version進行樂觀鎖並發控制：

二、並發控制實踐

2.1、基於_version進行樂觀鎖並發控制

1）構建數據

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

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

2）模擬兩個客戶端，都獲取了同一條數據

GET /test_index/test_type/7

{
  "_index": "test_index",
  "_type": "test_type",
  "_id": "7",
  "_version": 1,   #此時版本號是1
  "found": true,
  "_source": {
    "test_field": "test test"
  }
}

3）模擬一個客戶端，先更新了這個數據

同時帶上數據的版本號，確保說，es中的數據的版本號，跟客戶端中的數據的版本號是相同的，才能修改

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

{
  "_index": "test_index",
  "_type": "test_type",
  "_id": "7",
  "_version": 2,   #此時實際版本號已經變成2了
  "result": "updated",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "created": false
}

4）另外一個客戶端，嘗試基於version=1的數據去進行修改，同樣帶上version版本號，進行樂觀鎖的並發控制

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

{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[test_type][7]: version conflict, current version [2] is different than the one provided [1]",   #版本沖突
        "index_uuid": "6t1NqhChSpyuYY7m-Eq2jA",
        "shard": "3",
        "index": "test_index"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[test_type][7]: version conflict, current version [2] is different than the one provided [1]",
    "index_uuid": "6t1NqhChSpyuYY7m-Eq2jA",
    "shard": "3",
    "index": "test_index"
  },
  "status": 409
}

5）在樂觀鎖成功阻止並發問題之后，嘗試正確的完成更新

首先獲取正確的版本號

GET /test_index/test_type/7

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

基於最新的數據和版本號，去進行修改，帶上最新的版本號，可能這個步驟會需要反復執行好幾次，才能成功，特別是在多線程並發更新同一條數據很頻繁的情況下

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

{
  "_index": "test_index",
  "_type": "test_type",
  "_id": "7",
  "_version": 3,   #修改成功，版本號再次更改
  "result": "updated",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "created": false
}

2.2、基於external version進行樂觀鎖並發控制

es提供了一個feature，就是說，你可以不用它提供的內部_version版本號來進行並發控制，可以基於你自己維護的一個版本號來進行並發控制。舉個列子，加入你的數據在mysql里也有一份，然后你的應用系統本身就維護了一個版本號，無論是什么自己生成的，程序控制的。這個時候，你進行樂觀鎖並發控制的時候，可能並不是想要用es內部的_version來進行控制，而是用你自己維護的那個version來進行控制。

_version與external version區別：

1）格式區別

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

2）修改區別

_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

模擬external version樂觀鎖並發控制：

1）構建數據

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

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

2）模擬兩個客戶端同時查詢到這條數據

GET /test_index/test_type/8

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

3）第一個客戶端先進行修改，此時客戶端程序是在自己的數據庫中獲取到了這條數據的最新版本號，比如說是2

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

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

4）模擬第二個客戶端，同時拿到了自己數據庫中維護的那個版本號，也是2，同時基於version=2發起了修改

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

{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[test_type][8]: version conflict, current version [2] is higher or equal to the one provided [2]",  #報錯
        "index_uuid": "6t1NqhChSpyuYY7m-Eq2jA",
        "shard": "1",
        "index": "test_index"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[test_type][8]: version conflict, current version [2] is higher or equal to the one provided [2]",
    "index_uuid": "6t1NqhChSpyuYY7m-Eq2jA",
    "shard": "1",
    "index": "test_index"
  },
  "status": 409
}

5）在並發控制成功后，重新基於最新的版本號發起更新

獲取最新的版本號：

GET /test_index/test_type/8

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

重新發起put請求，version加1

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

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