| Elasticsearch | 您所在的位置：网站首页 › elasticsearch客户端异步 › Elasticsearch |
Elasticsearch
|
1. 详解并发冲突
在电商场景下，工作流程为： 读取商品信息，包括库存数量 用户下单购买 更新商品信息，将库存数减一如果是多线程操作，就可能有多个线程并发的去执行上述的3步骤流程，假如此时有两个人都来读取商品数据，两个线程并发的服务于两个人，同时在进行商品库存数据的修改。正确的情况：线程A将库存-1，设置为99件，线程B接着读取99件，再-1，变为98件。如果A,B线程都读取的为100件，A处理完之后修改为99件，B处理完之后再次修改为99件，此时结果就出错了。 在读取商品数据时，同时对这一行数据加锁，当此线程处理完数据之后，再解锁，另一个线程开始处理。 悲观锁并发控制方案，就是在各种情况下，都上锁。上锁之后，就只有一个线程可以操作这一条数据，不同的场景之下，上的锁不同，行级锁，表级锁，读锁，写锁。 2.2 乐观锁乐观锁不加锁，每个线程都可以任意操作。es的每条文档中有一个version字段，新建文档后为1，修改一次累加，线程A,B同时读取到数据，version=1，A处理完之后库存为99，在写入es的时候会跟es中的版本号比较，都是1，则写入成功，version=2，B处理完之后也为99，存入es时与es中的数据的版本号version=2相比，明显不同，此时不会用99去更新，而是重新读取最新的数据，再减一，变为98，执行上述操作，写入。 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 }复制代码 |
【本文地址】