Elasticsearch——併發衝突以及解決方案
在電商場景下，工作流程為：
- 讀取商品資訊，包括庫存數量
- 使用者下單購買
- 更新商品資訊，將庫存數減一
如果是多執行緒操作，就可能有多個執行緒併發的去執行上述的3步驟流程，假如此時有兩個人都來讀取商品資料，兩個執行緒併發的服務於兩個人，同時在進行商品庫存資料的修改。正確的情況：執行緒A將庫存-1，設定為99件，執行緒B接著讀取99件，再-1，變為98件。如果A,B執行緒都讀取的為100件，A處理完之後修改為99件，B處理完之後再次修改為99件，此時結果就出錯了。
2. 解決方案
2.1 悲觀鎖
在讀取商品資料時，同時對這一行資料加鎖，當此執行緒處理完資料之後，再解鎖，另一個執行緒開始處理。
悲觀鎖併發控制方案，就是在各種情況下，都上鎖。上鎖之後，就只有一個執行緒可以操作這一條資料，不同的場景之下，上的鎖不同，行級鎖，表級鎖，讀鎖，寫鎖。
2.2 樂觀鎖
樂觀鎖不加鎖，每個執行緒都可以任意操作。es的每條文件中有一個version欄位，新建文件後為1，修改一次累加，執行緒A,B同時讀取到資料，version=1，A處理完之後庫存為99，在寫入es的時候會跟es中的版本號比較，都是1，則寫入成功，version=2，B處理完之後也為99，存入es時與es中的資料的版本號version=2相比，明顯不同，此時不會用99去更新，而是重新讀取最新的資料，再減一，變為98，執行上述操作，寫入。
2.3 Elasticsearch的樂觀鎖
Elasticsearch的後臺都是多執行緒非同步的，多個請求之間是亂序的，可能後修改的先到，先修改的後到。
Elasticsearch的多執行緒非同步併發修改是基於自己的_version版本號進行樂觀鎖併發控制的。
在後修改的先到時，修改完畢後，當先修改的後到時，會比較一下_version版本號，如果不相等就直接扔掉，不需要了。這樣結果會就會儲存為一個正確狀態。
程式碼示例：
PUT /test_index/test_type/3
{
"test_field": "test test"
}
結果：
{
"_index": "test_index",
"_type": "test_type",
"_id": "3",
"_version": 1,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"created": true
}
修改
PUT /test_index/test_type/3
{
"test_field": "test1 test1"
}
結果
{
"_index": "test_index",
"_type": "test_type",
"_id": "3",
"_version": 2,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"created": true
}
刪除
DELETE /test_index/test_type/3
結果：
{
"_index": "test_index",
"_type": "test_type",
"_id": "3",
"_version": 3,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"created": true
}
重新建立
PUT /test_index/test_type/3
{
"test_field": "test1 test1"
}
結果
{
"_index": "test_index",
"_type": "test_type",
"_id": "4",
"_version": 1,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"created": true
}
複製程式碼
刪除操作也會對這條資料的版本號加1
在刪除一個document之後，可以從一個側面證明，它不是立即物理刪除掉的，因為它的一些版本號等資訊還是保留著的。先刪除一條document，再重新建立這條document，其實會在delete version基礎之上，再把version號加1
2.4 es的樂觀鎖併發控制示例
-
先新建一條資料
PUT /test_index/test_type/4 { "test_field": "test" } 複製程式碼 -
模擬兩個客戶端，都獲取到了同一條資料
GET /test_index/test_type/4 返回 { "_index": "test_index", "_type": "test_type", "_id": "4", "_version": 1, "found": true, "_source": { "test_field": "test" } } 複製程式碼 -
其中一個客戶端，先更新了一下這個資料， 同時帶上資料的版本號，確保說，es中的資料的版本號，跟客戶端中的資料的版本號是相同的，才能修改
PUT test_index/test_type/4?version=1 { "test_field": "client1 changed" } 返回結果 { "_index": "test_index", "_type": "test_type", "_id": "4", "_version": 2, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": false } 複製程式碼 -
另外一個客戶端，嘗試基於version=1的資料去進行修改，同樣帶上version版本號，進行樂觀鎖的併發控制
PUT test_index/test_type/4?version=1 { "test_field": "client2 changed" } 會出錯，返回 { "error": { "root_cause": [ { "type": "version_conflict_engine_exception", "reason": "[test_type][4]: version conflict, current version [2] is different than the one provided [1]", "index_uuid": "rsiZYqiwSCC2XdR8N2bJow", "shard": "2", "index": "test_index" } ], "type": "version_conflict_engine_exception", "reason": "[test_type][4]: version conflict, current version [2] is different than the one provided [1]", "index_uuid": "rsiZYqiwSCC2XdR8N2bJow", "shard": "2", "index": "test_index" }, "status": 409 } 複製程式碼樂觀鎖就成功阻止併發問題
-
在樂觀鎖成功阻止併發問題之後，嘗試正確的完成更新
重新進行GET請求，得到 version
GET /test_index/test_type/4 { "_index": "test_index", "_type": "test_type", "_id": "4", "_version": 2, "found": true, "_source": { "test_field": "client1 changed" } } 複製程式碼基於最新的資料和版本號，去進行修改，修改後，帶上最新的版本號，可能這個步驟會需要反覆執行好幾次，才能成功，特別是在多執行緒併發更新同一條資料很頻繁的情況下
PUT /test_index/test_type/4?version=2 { "test_field": "client2 changed" } 返回 { "_index": "test_index", "_type": "test_type", "_id": "4", "_version": 3, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": false } 複製程式碼
2.5 基於external version進行樂觀鎖併發控制
es提供了一個feature，就是說，你可以不用它提供的內部_version版本號來進行併發控制，可以基於你自己維護的一個版本號來進行併發控制。
?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
程式碼示例：
-
先建立一條資料
PUT test_index/test_type/5 { "test_field": "external test" } 返回 { "_index": "test_index", "_type": "test_type", "_id": "5", "_version": 1, "result": "created", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": true } 複製程式碼 -
模擬兩個客戶端同時查詢到這條資料
GET /test_index/test_type/5 返回 { "_index": "test_index", "_type": "test_type", "_id": "5", "_version": 1, "found": true, "_source": { "test_field": "external test" } } 複製程式碼 -
第一個客戶端先進行修改，此時客戶端程式是在自己的資料庫中獲取到了這條資料的最新版本號，比如說是2
PUT /test_index/test_type/5?version=2&version_type=external { "test_field": "external client1 changed" } 返回 { "_index": "test_index", "_type": "test_type", "_id": "5", "_version": 2, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": false } 複製程式碼 -
模擬第二個客戶端，同時拿到了自己資料庫中維護的那個版本號，也是2，同時基於version=2發起了修改
PUT /test_index/test_type/5?version=2&version_type=external { "test_field": "external client2 changed" } 會出錯，返回 { "error": { "root_cause": [ { "type": "version_conflict_engine_exception", "reason": "[test_type][5]: version conflict, current version [2] is higher or equal to the one provided [2]", "index_uuid": "rsiZYqiwSCC2XdR8N2bJow", "shard": "1", "index": "test_index" } ], "type": "version_conflict_engine_exception", "reason": "[test_type][5]: version conflict, current version [2] is higher or equal to the one provided [2]", "index_uuid": "rsiZYqiwSCC2XdR8N2bJow", "shard": "1", "index": "test_index" }, "status": 409 } 複製程式碼 -
在併發控制成功後，重新基於最新的版本號發起更新
GET /test_index/test_type/5 返回 { "_index": "test_index", "_type": "test_type", "_id": "5", "_version": 2, "found": true, "_source": { "test_field": "external client1 changed" } } PUT /test_index/test_type/5?version=3&version_type=external { "test_field": "external client2 changed" } 返回 { "_index": "test_index", "_type": "test_type", "_id": "5", "_version": 3, "result": "updated", "_shards": { "total": 2, "successful": 1, "failed": 0 }, "created": false }複製程式碼