解决在使用 Elasticsearch(ES)多线程批量操作时导致并发一致性的问题！！

朱雀 2024-03-30 09:45 13阅读 0赞

先说一下什么是数据库数据库中**并发一致性**问题！

#### 1、在并发环境下，事务的隔离性很难保证，因此会出现很多并发一致性问题。 ####

*  **数据丢失**  
    T1 和 T2 两个事务都对一个数据进行修改，T1 先修改，T2 随后修改，T2 的修改覆盖了 T1 的修改。  
    ![在这里插入图片描述][37be3f91fbdf448380824e3bc1b266b6.png]
 *  **读脏数据**  
    T1 修改一个数据，T2 随后读取这个数据。如果 T1 撤销了这次修改，那么 T2 读取的数据是脏数据。  
    ![在这里插入图片描述][c99365cd4c7c4a749caecc5a04eb7e4d.png]
 *  **不可重复读**  
    T2 读取一个数据，T1 对该数据做了修改。如果 T2 再次读取这个数据，此时读取的结果和第一次读取的结果不同。  
    ![在这里插入图片描述][aabc36c7d21347109c7732516034b754.png]
 *  **幻影读**  
    T1 读取某个范围的数据，T2 在这个范围内插入新的数据，T1 再次读取这个范围的数据，此时读取的结果和和第一次读取的结果不同。

#### 当然上面只是提一下什么是一致性的问题。 ####

现在就用一个在开发中的实际例子来说一下ES中多线程的情况下如何避免不一致性的问题。  
这里我使用了批量更新数据的一个例子；  
**业务:** 前端发送任务后，将ES中所有符合条件的数据的某一个字段全部更新一遍。由于数据量太大，所以准备使用多线程的方式去执行。例如前端发送了两个任务，这就代表我需要去更新两次。但是如果使用多线程就会导致最后一次更新完的数据可能将第一次更新的部分数据覆盖。

在多线程情况下，更新 Elasticsearch 数据库可能会导致并发一致性问题。这意味着，如果多个线程同时访问数据库并进行更新操作，可能会导致冲突，从而使数据库的状态混乱不堪。

为了解决这个问题，可以使用 Elasticsearch 的冲突解决机制。这包括使用版本号来跟踪文档的更新，并在更新操作时使用比较-替换模式（compare-and-swap）。  
例如，假设你有一个文档，其中包含一个版本号字段：

{
        
    	  "title": "My Document",
    	  "description": "This is my document",
    	  "version": 1
    	}

当你想要更新这个文档时，可以使用 version 字段来进行并发控制。例如，假设你想要更新文档的 description 字段：

POST my_index/my_type/123/_update
    {
        
      "doc": {
        
        "description": "This is an updated description"
      },
      "version": 1
    }

这将会检查文档的当前版本是否为 1，并只有在版本匹配的情况下才会执行更新操作。如果文档的版本已被更新，则会返回冲突错误。

在 Java 中使用 Elasticsearch 的 HighLevelRestClient 实现版本号控制时，可以使用 UpdateRequest 类来创建更新请求。该请求可以使用 setIfSeqNo(long) 和 setIfPrimaryTerm(long) 方法来设置版本号。  
例如，假设你想要更新文档的 description 字段，并且想要使用版本号进行并发控制：

RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200, "http")));
    
    UpdateRequest request = new UpdateRequest("my_index", "my_type", "123")
      .doc("description", "This is an updated description")
      .setIfSeqNo(1)
      .setIfPrimaryTerm(1);
    
    client.update(request);

这将会检查文档的当前版本是否为 1，并只有在版本匹配的情况下才会执行更新操作。如果文档的版本已被更新，则会返回冲突错误。

#### 如果多次请求更新我怎么知道版本号具体设置多少? ####

在获取文档时获取版本号。可以使用 GetRequest.fetchSourceContext(FetchSourceContext) 方法来获取文档的版本号。例如：

GetRequest request = new GetRequest("my_index", "my_type", "123")
      .fetchSourceContext(new FetchSourceContext(true, new String[]{
        "_seq_no", "_primary_term"}, null));
    
    GetResponse response = client.get(request);
    long seqNo = response.getSeqNo();
    long primaryTerm = response.getPrimaryTerm();

在执行更新操作时使用 Get 操作获取版本号。在执行更新操作之前，可以使用 Get 操作获取文档的版本号，然后将版本号设置到更新请求中。例如：

GetRequest getRequest = new GetRequest("my_index", "my_type", "123")
      .fetchSourceContext(new FetchSourceContext(true, new String[]{
        "_seq_no", "_primary_term"}, null));
    //在执行前获取版本号
    GetResponse getResponse = client.get(getRequest);
    long seqNo = getResponse.getSeqNo();
    long primaryTerm = getResponse.getPrimaryTerm();
    //设置版本号
    UpdateRequest updateRequest = new UpdateRequest("my_index", "my_type", "123")
      .doc("description", "This is an updated description")
      .setIfSeqNo(seqNo)
      .setIfPrimaryTerm(primaryTerm);
    
    client.update(updateRequest);

[37be3f91fbdf448380824e3bc1b266b6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/4dd643f236e04e8189f7326a46519c81.png
[c99365cd4c7c4a749caecc5a04eb7e4d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/555d3c2fe17b4fd9bcd1984f32ef46e4.png
[aabc36c7d21347109c7732516034b754.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/21eeaf941b8f47118df688a653bf727d.png