elasticsearch学习笔记（十三）——Elasticsearch乐观锁并发控制实战

1、elasticsearch基于_version新版本更新为if_seq_no和if_primary_term进行乐观锁的并发控制1先构造一条数据PUT /test_index/_doc/1 { test_field:test test }2模拟两个客户端都获取到同一条数据GET /test_index/_doc/13其中一个客户端先更新了一下这个数据PUT /test_index/_doc/1?version1 { test_field: test client 1 }4另外一个客户端尝试基于version1的数据去修改同样带上version版本号进行乐观锁的并发控制PUT /test_index/_doc/1?version1 { test_field: test client 2 }此时就会报错version conflict5在乐观锁成功阻止并发问题之后尝试正确的完成更新GET /test_index/_doc/1对于更新失败的客户端查询出最新的版本号之后基于最新的数据和版本号去再次进行修改修改时带上最新的版本号可能这个步骤会需要反复执行好几次才能成功特别是在多线程并发更新同一条数据很频繁的情况下注意特别说明要完成上面的实验需要es版本低于6.7。version Deprecated in 6.7.0. Please use if_seq_no if_primary_term instead. See Optimistic concurrency control for more details.这段说得很清楚高版本是使用if_seg_no和if_primary_term这两个参数实现的。下面把上面的实现用高版本做一下。我用的版本是7.0的1先构造一条数据PUT /test_index/_doc/1 { test_field:test test } { _index : test_index, _type : _doc, _id : 1, _version : 3, result : created, _shards : { total : 2, successful : 1, failed : 0 }, _seq_no : 2, _primary_term : 1 }2模拟两个客户端都获取到了同一条数据GET /test_index/_doc/1 { _index : test_index, _type : _doc, _id : 1, _version : 3, _seq_no : 2, _primary_term : 1, found : true, _source : { test_field : test test } }3其中一个客户端先更新了一下这个数据PUT /test_index/_doc/1?if_seq_no2if_primary_term1 { test_field:test client 1 } { _index : test_index, _type : _doc, _id : 1, _version : 4, result : updated, _shards : { total : 2, successful : 1, failed : 0 }, _seq_no : 3, _primary_term : 1 }4另外一个客户端尝试基于if_seq_no和if_primary_term的数据进行修改进行乐观锁的并发控制PUT /test_index/_doc/1?if_seq_no2if_primary_term1 { test_field:test client 2 } { error: { root_cause: [ { type: version_conflict_engine_exception, reason: [1]: version conflict, required seqNo [2], primary term [1]. current document has seqNo [3] and primary term [1], index_uuid: 0jAS2GP1TPG5J8PlqGdYIQ, shard: 4, index: test_index } ], type: version_conflict_engine_exception, reason: [1]: version conflict, required seqNo [2], primary term [1]. current document has seqNo [3] and primary term [1], index_uuid: 0jAS2GP1TPG5J8PlqGdYIQ, shard: 4, index: test_index }, status: 409 }5在乐观锁成功阻止并发问题之后尝试正确的完成更新GET /test_index/_doc/1 { _index : test_index, _type : _doc, _id : 1, _version : 4, _seq_no : 3, _primary_term : 1, found : true, _source : { test_field : test client 1 } }基于最新的数据和if_seq_no,if_primary_term进行修改可能这个过程会需要反复执行好几次才能成功特别是在多线程并发更新同一条数据很频繁的情况下PUT /test_index/_doc/1?if_seq_no3if_primary_term1 { test_field:test client 2 } { _index : test_index, _type : _doc, _id : 1, _version : 5, result : updated, _shards : { total : 2, successful : 1, failed : 0 }, _seq_no : 4, _primary_term : 1 } GET /test_index/_doc/1 { _index : test_index, _type : _doc, _id : 1, _version : 5, _seq_no : 4, _primary_term : 1, found : true, _source : { test_field : test client 2 } }对于if_seq_no和if_primary_term官方文档已经有比较详细的叙述https://www.elastic.co/guide/... 。这里我说下简单的理解方式对于if_primary_term记录的就是具体的哪个主分片而if_seq_no这个参数起的作用和旧版本中的_version是一样的之所以加上if_primary_term这个参数主要是提高并发的性能以及更自然因为每个document都只会在某一个主分片中所以由所在主分片分配序列号比由之前通过一个参数_version相当于由整个ES集群分配版本号要来的更好。To ensure an older version of a document doesn’t overwrite a newer version, every operation performed to a document is assigned a sequence number by the primary shard that coordinates that change. The sequence number is increased with each operation and thus newer operations are guaranteed to have a higher sequence number than older operations. Elasticsearch can then use the sequence number of operations to make sure a newer document version is never overridden by a change that has a smaller sequence number assigned to it.简单翻译就是为确保较旧版本的文档不会覆盖较新版本对文档执行的每个操作都会由协调该更改的主分片分配序列号。每次操作都会增加序列号因此保证较新的操作具有比旧操作更高的序列号。然后Elasticsearch可以使用序列号操作来确保更新的文档版本永远不会被分配给它的序列号更小的更改覆盖。2、基于external version进行乐观锁的并发控制6.7版本已移除具体的实战就不做了本质思想也很简单就是版本号是存储在自己的数据库中的可以由开发人员自己控制。但是在6.7版本之后就移除这个功能主要是因为The update API does not support versioning other than internalExternal (version types external and external_gte) or forced (version type force) versioning is not supported by the update API as it would result in Elasticsearch version numbers being out of sync with the external system.更新API不支持外部版本类型external和external_gte或强制版本类型force版本控制因为它会导致Elasticsearch版本号与外部系统不同步