一起学Elasticsearch系列-并发控制-Toy模板网

这篇具有很好参考价值的文章主要介绍了一起学Elasticsearch系列-并发控制。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

本文已收录至Github，推荐阅读 👉 Java随想录
微信公众号：Java随想录

ES的并发控制是一种机制，用于处理多个同时对同一份数据进行读写操作的情况，以确保数据的一致性和正确性。

实现并发控制的方法主要有两种：悲观锁和乐观锁。

悲观锁

悲观锁是一种并发控制机制，它基于一种假设：在任何时候都会发生并发冲突。因此，在进行读写操作之前，悲观锁会将数据标记为“被锁定”，以阻止其他操作对其进行修改。

对于一个共享数据，某个线程访问到这个数据的时候，会认为这个数据随时有可能会被其他线程访问而造成数据不安全的情况，因此线程在每次访问的时候都会对数据加一把锁。这样其他线程如果在加锁期间想访问当前数据就只能等待，也就是阻塞线程了。

一个现实中的悲观锁例子是银行柜台排队取款。

假设某个银行只有一个柜台提供服务，多个客户需要办理业务。当第一个客户进入柜台并开始办理业务时，其他客户会悲观地认为自己无法立即获得服务，因此必须在柜台前排队等待。这种情况下，每个客户都悲观地预期自己必须等待一段时间才能办理业务，直到轮到自己。柜台就像是被锁住的资源，只允许一个客户同时使用，其他客户需等待释放才能进行操作。

乐观锁

乐观锁基于假设多个事务之间很少发生冲突的思想。在使用乐观锁的情况下，系统默认认为并发操作不会产生冲突，因此不会立即阻塞其他事务的执行。

具体实现乐观锁的方式是，每个事务在读取数据时会获取一个版本号（或时间戳），在提交更新时会检查该版本号是否被其他事务修改过。如果版本号未被修改，意味着没有冲突发生，可以继续提交更新；但如果版本号已经被修改，说明有其他事务已经修改了数据，当前事务则需要重新读取最新数据并重新执行。

乐观锁主要依赖于数据的版本控制来实现并发控制，可以降低锁粒度，提高并发性能。然而，在高并发环境下，如果冲突频繁发生，乐观锁可能会导致大量的回滚和重试操作，影响系统的性能。因此，在选择乐观锁时需要仔细评估并发冲突的概率和代价。

一个现实中的乐观锁例子是电影院的选座系统。当多个用户同时访问选座系统时，系统采用乐观锁机制来处理并发操作。

假设用户A和用户B同时进入选座系统，并选择了相同的座位。系统会首先记录用户A和用户B选择该座位的时间戳或版本号。当用户A提交座位选择后，系统会检查座位的时间戳或版本号是否被修改。如果未被修改，则说明没有冲突发生，系统会将座位分配给用户A。但如果时间戳或版本号已经被修改，说明用户B已经在此期间选择了相同的座位，系统需要重新读取最新数据并通知用户A重新选择座位。

在这个例子中，选座系统默认认为用户之间很少选择相同座位，因此不立即阻塞其他用户的操作。通过乐观锁机制，系统可以减少并发冲突的发生，并提高用户的选择效率和系统的并发性能。

如何选择

首先，悲观锁和乐观锁没有孰优孰劣，它们各自有各自的适用场景

选择乐观锁还是悲观锁取决于具体的应用场景和需求。下面是一些考虑因素：

并发程度：如果系统中并发冲突较为频繁，多个事务之间经常需要争抢同一个资源，那么悲观锁可能更适合。悲观锁可以确保资源的互斥访问，但会导致其他事务等待锁释放，可能影响系统的性能。
冲突概率：如果系统中并发冲突较为罕见，多个事务之间很少竞争同一个资源，那么乐观锁可能更适合。乐观锁假设并发操作不会产生冲突，可以提高系统的并发性能。但如果冲突发生频率较高，乐观锁可能会导致大量的回滚和重试，降低系统性能。
锁粒度：悲观锁通常会对整个资源或数据进行加锁，阻塞其他事务的访问。如果需要细粒度的并发控制，或者希望允许多个事务同时读取数据，那么乐观锁可能更适合。乐观锁可以降低锁粒度，提高并发性能。
实现复杂度：乐观锁相对而言实现起来更简单，只需要添加版本号或时间戳等机制即可。而悲观锁的实现可能需要借助底层的锁机制，如数据库的行级锁或使用并发控制工具。因此，在实现复杂度方面，乐观锁更容易实现和维护。

总而言之，选择乐观锁还是悲观锁应该根据具体场景和需求进行评估。如果并发冲突较为频繁且需要确保互斥访问，可以选择悲观锁；如果并发冲突较为罕见且需要提高并发性能，可以选择乐观锁。

ES的并发控制

ES的并发控制是通过乐观锁机制来实现的

Elasticsearch 是分布式的。创建、更新或删除文档时，必须将文档的新版本复制到集群中的其他节点。ES 也是异步并行的，所以这些复制请求是并行发送的，并且可能不按顺序执行到每个节点。ES需要一种并发策略来保证数据的安全性，而这种策略就是乐观锁并发控制策略。

为了保证旧文档不会被新文档覆盖，对文档执行的每个操作都由协调该更改的主分片分配一个序列号（_seq_no）。每个操作都会操作序列号递增，因此可以保证较新的操作具有更高的序列号。然后，ES 可以使用操作序列号来确保更新的文档版本永远不会被分配了较小序列号的版本覆盖。

版本号：_version

基本原理

每个索引文档都有一个版本号。默认情况下，使用从 1 开始的内部版本控制，每次更新都会增加，包括删除。

版本号可以设置为外部值（例如，如果在数据库中维护）。要启用此功能，version_type 应设置为 external。提供的值必须是大于或等于 0 且小于 9.2e+18 左右的数字长整型值。

使用外部版本类型时，系统会检查传递给索引请求的版本号是否大于当前存储文档的版本。如果为真，文档将被索引并使用新的版本号。如果提供的值小于或等于存储文档的版本号，则会发生版本冲突，索引操作将失败。

作用范围

_version 的有效范围为当前文档。

版本类型

version_type字段有以下几种取值：

internal（默认值）：使用内部版本号（_version）来检测文档版本冲突。如果两个操作同时修改了相同的文档，后面执行的操作将失败并返回版本冲突的错误。
external：使用外部版本号来检测文档版本冲突。当执行操作时，必须提供文档的当前版本号，如果提供的版本号与实际版本号不匹配，则操作将失败。
external_gte：同样使用外部版本号，但提供的版本号大于等于实际版本号时才执行操作。如果提供的版本号小于实际版本号，则操作将失败，external_gte 需要谨慎使用，否则可能会丢失数据。

通过指定适当的version_type，可以根据业务需求选择如何处理文档版本冲突。在某些场景下，外部版本号可能更适合，因为它允许应用程序明确控制版本冲突的处理方式。而在其他情况下，可以使用内部版本号来简化版本管理，并自动处理版本冲突。

_seq_no & _primary_term

_seq_no 和 _primary_term 是用来并发控制，和 _version不同，_version属于当前文档，而 _seq_no属于整个index。

_seq_no & _primary_term

_seq_no：索引级别的版本号，索引中所有文档共享一个 _seq_no 。
_primary_term：primary_term是一个整数，每当Primary Shard发生重新分配时，比如节点重启，Primary选举或重新分配等primary_term会递增1。主要作用是用来恢复数据时处理当多个文档的_seq_no 一样时的冲突，避免 Primary Shard 上的数据写入被覆盖。

if_seq_no & if_primary_term

在Elasticsearch中，if_seq_no 和 if_primary_term 是用于乐观锁并发控制的参数，用于确保对文档的操作不会与其他操作产生冲突。

if_seq_no 参数用于指定期望的文档序列号（seq_no），而 if_primary_term 参数用于指定期望的 primary term。这两个参数一起作为条件，如果提供的条件与实际存储的文档序列号和主要项匹配，则操作成功执行；否则，操作将失败并返回版本冲突的错误。

假设我们有一个名为 my_index 的索引，其中包含 _id 为 1 的文档。当前文档的 seq_no 是 10，primary_term 是 1。

示例 1：更新文档

PUT my_index/_doc/1?if_seq_no=10&if_primary_term=1
{
  "foo": "bar"
}

输出：

{
  "_index": "my_index",
  "_type": "_doc",
  "_id": "1",
  "_version": 11,
  "result": "updated",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  }
}

在这个示例中，通过提供正确的 if_seq_no 和 if_primary_term 条件，操作成功地更新了文档，并返回了更新后的版本号 _version。

示例 2：更新文档，但条件不匹配

PUT my_index/_doc/1?if_seq_no=11&if_primary_term=1
{
  "foo": "bar"
}

输出：

{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[1]: version conflict, current version [11], provided version [11]",
        "index_uuid": "xxxxxxxxxxxxx",
        "shard": "0",
        "index": "my_index"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[1]: version conflict, current version [11], provided version [11]",
    "index_uuid": "xxxxxxxxxxxxx",
    "shard": "0",
    "index": "my_index"
  },
  "status": 409
}