ElasticSearch分布式文档系统动手实战

﹏ヽ暗。殇╰゛Y 2021-11-11 07:14 369阅读 0赞

1、深度图解剖析Elasticsearch并发冲突问题（类似并发操作）

![image.png][]

2、深度图解剖析悲观锁与乐观锁两种并发控制方案（类比之前的知识）

![image.png][image.png 1]

es是基于乐观锁的，基于版本号

3、图解Elasticsearch内部如何基于\_version进行乐观锁并发控制

![image.png][image.png 2]

上机动手实战演练基于external version进行乐观锁并发控制 external version

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/8
    
    {
    
     "test_field": "test"
    
    }

（2）模拟两个客户端同时查询到这条数据

GET /test_index/test_type/8

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

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

（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": "6m0G7yx7R1KECWWGnfH1sw",

"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": "6m0G7yx7R1KECWWGnfH1sw",

"shard": "1",

"index": "test\_index"

\},

"status": 409

（5）在并发控制成功后，重新基于最新的版本号发起更新

GET /test_index/test_type/8

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

4、图解partial update实现原理以及其优点

![image.png][image.png 3]

5、上机动手实战演练partial update

PUT /test_index/test_type/10
    
    {
    
     "test_field1": "test1",
    
     "test_field2": "test2"
    
    }

POST /test_index/test_type/10/_update
    
    {
    
     "doc": {
    
       "test_field2": "updated test2"
    
     }
    
    }

上机动手实战演练基于groovy脚本进行

es，其实是有个内置的脚本支持的，可以基于groovy脚本实现各种各样的复杂操作

基于groovy脚本，如何执行partial update

es scripting module，我们会在高手进阶篇去讲解，这里就只是初步讲解一下

PUT /test_index/test_type/11
    
    {
    
     "num": 0,
    
     "tags": []
    
    }

（1）内置脚本

POST /test_index/test_type/11/_update
    
    {
    
      "script" : "ctx._source.num+=1"
    
    }

（2）外部脚本

ctx.\_source.tags+=new\_tag （在elasticsearch.yml文件中添加）

POST /test_index/test_type/11/_update
    
    {
    
     "script": {
    
       "lang": "groovy",  
    
       "file": "test-add-tags",
    
       "params": {
    
         "new_tag": "tag1"
    
       }
    
     }
    
    }

（3）用脚本删除文档

ctx.op = ctx.\_source.num == count ? 'delete' : 'none'

POST /test_index/test_type/11/_update
    
    {
    
     "script": {
    
       "lang": "groovy",
    
       "file": "test-delete-document",
    
       "params": {
    
         "count": 1
    
       }
    
     }
    
    }

（4）upsert操作

POST /test_index/test_type/11/_update
    
    {
    
     "doc": {
    
       "num": 1
    
     }
    
    }

"error": \{

"root\_cause": \[

"type": "document\_missing\_exception",

"reason": "\[test\_type\]\[11\]: document missing",

"index\_uuid": "6m0G7yx7R1KECWWGnfH1sw",

"shard": "4",

"index": "test\_index"

\],

"type": "document\_missing\_exception",

"reason": "\[test\_type\]\[11\]: document missing",

"index\_uuid": "6m0G7yx7R1KECWWGnfH1sw",

"shard": "4",

"index": "test\_index"

\},

"status": 404

如果指定的document不存在，就执行upsert中的初始化操作；如果指定的document存在，就执行doc或者script指定的partial update操作

POST /test_index/test_type/11/_update
    
    {
    
      "script" : "ctx._source.num+=1",
    
      "upsert": {
    
          "num": 0,
    
          "tags": []
    
      }
    
    }

6、partial update内置乐观锁并发控制

![image.png][image.png 4]

7、批量查询的好处

就是一条一条的查询，比如说要查询100条数据，那么就要发送100次网络请求，这个开销还是很大的

如果进行批量查询的话，查询100条数据，就只要发送1次网络请求，网络请求的性能开销缩减100倍

a.mget的语法

（1）一条一条的查询

GET /test\_index/test\_type/1

GET /test\_index/test\_type/2

（2）mget批量查询

GET /_mget
    {
      "docs" : [
         {
            "_index" : "test_index",
            "_type" :  "test_type",
            "_id" :    1
         },
         {
            "_index" : "test_index",
            "_type" :  "test_type",
            "_id" :    2
         }
      ]
    }
    {
     "docs": [
       {
         "_index": "test_index",
         "_type": "test_type",
         "_id": "1",
         "_version": 2,
         "found": true,
         "_source": {
           "test_field1": "test field1",
           "test_field2": "test field2"
         }
       },
       {
         "_index": "test_index",
         "_type": "test_type",
         "_id": "2",
         "_version": 1,
         "found": true,
         "_source": {
           "test_content": "my test"
         }
       }
     ]
    }

（3）如果查询的document是一个index下的不同type种的话

GET /test_index/_mget
    
    {
    
      "docs" : [
    
         {
    
            "_type" :  "test_type",
    
            "_id" :    1
    
         },
    
         {
    
            "_type" :  "test_type",
    
            "_id" :    2
    
         }
    
      ]
    
    }

（4）如果查询的数据都在同一个index下的同一个type下，最简单了

GET /test_index/test_type/_mget
    
    {
    
      "ids": [1, 2]
    
    }

b.mget的重要性

可以说mget是很重要的，一般来说，在进行查询的时候，如果一次性要查询多条数据的话，那么一定要用batch批量操作的api

尽可能减少网络开销次数，可能可以将性能提升数倍，甚至数十倍，非常非常之重要

8、routing

(1）document路由到shard上是什么意思？

（2）路由算法：shard = hash(routing) % number\_of\_primary\_shards

举个例子，一个index有3个primary shard，P0，P1，P2

每次增删改查一个document的时候，都会带过来一个routing number，默认就是这个document的\_id（可能是手动指定，也可能是自动生成）

routing = \_id，假设\_id=1

会将这个routing值，传入一个hash函数中，产出一个routing值的hash值，hash(routing) = 21

然后将hash函数产出的值对这个index的primary shard的数量求余数，21 % 3 = 0

就决定了，这个document就放在P0上。

决定一个document在哪个shard上，最重要的一个值就是routing值，默认是\_id，也可以手动指定，相同的routing值，每次过来，从hash函数中，产出的hash值一定是相同的

无论hash值是几，无论是什么数字，对number\_of\_primary\_shards求余数，结果一定是在0~number\_of\_primary\_shards-1之间这个范围内的。0,1,2。

（3）\_id or custom routing value

默认的routing就是\_id

也可以在发送请求的时候，手动指定一个routing value，比如说put /index/type/id?routing=user\_id

手动指定routing value是很有用的，可以保证说，某一类document一定被路由到一个shard上去，那么在后续进行应用级别的负载均衡，以及提升批量读取的性能的时候，是很有帮助的

（4）primary shard数量不可变的谜底

![image.png][image.png 5]

9、document内部查询原理

(1)、客户端发送请求到任意一个node，成为coordinate node

(2)、coordinate node对document进行路由，将请求转发到对应的node，此时会使用round-robin随机轮询算法，在primary shard以及其所有replica中随机选择一个，让读请求负载均衡

(3)、接收请求的node返回document给coordinate node

(4)、coordinate node返回document给客户端

(5)、特殊情况：document如果还在建立索引过程中，可能只有primary shard有，任何一个replica shard都没有，此时可能会导致无法读取到document，但是document完成索引建立之后，primary shard和replica shard就都有了

![image.png][image.png 6]

[image.png]: /images/20211109/f858ef809c7d4a108e670d9ba891247f.png
[image.png 1]: /images/20211109/121640682dde419aa077673c02bdcf9d.png
[image.png 2]: /images/20211109/5a9d0346a6ab4185a960d150db3b8b0e.png
[image.png 3]: /images/20211109/5adaf7deecfd4a81886fd1d3fc80d5fd.png
[image.png 4]: /images/20211109/86e1278b353a495f835bdde06890b01c.png
[image.png 5]: /images/20211109/b7002081acd149ab8c8845b182167b48.png
[image.png 6]: /images/20211109/100f49943dbe48eab4f3e961995f9fac.png