0
清茶豆奶 · 2020年04月30日

Elasticsearch系列---实现分布式锁

Elasticsearch

概要

Elasticsearch在文档更新时默认使用的是乐观锁方案，而Elasticsearch利用文档的一些create限制条件，也能达到悲观锁的效果，我们一起来看一看。

乐观锁与悲观锁

乐观锁

ES默认实现乐观锁，所有的数据更新默认使用乐观锁机制。document更新时，必须要带上currenct version，更新时与document的version进行比较，如果相同进行更新操作，不相同表示已经被别的线程更新过了，此时更新失败，并且重新获取新的version再尝试更新。

悲观锁

我们举一个这样的例子：Elasticsearch存储文件系统的目录、文件名信息，有多个线程需要对/home/workspace/ReadMe.txt进行追加修改，而且是并发执行的，有先后顺序之分，跟之前的库存更新案例有点不一样，此时单纯使用乐观锁，可能会出现乱序的问题。

这种场景就需要使用悲观锁控制，保证线程的执行顺序，有一个线程在修改，其他的线程只能挂起等待。悲观锁通过/index/lock/实现，只有一个线程能做修改操作，其他线程block掉。

悲观锁有三种，分别对应三种粒度，由粗到细可为分：

  • 全局锁：最粗的锁，直接锁整个索引
  • document锁：指定id加锁，只锁一条数据，类似于数据库的行锁
  • 共享锁和排他锁：也叫读写锁，针对一条数据分读和写两种操作，一般共享锁允许多个线程对同一条数据进行加锁，排他锁只允许一个线程对数据加锁，并且排他锁和共享锁互斥。

锁的基本操作步骤

我们使用锁的基本步骤都是一样的，无论是关系型数据库、Redis/Memcache/Zookeeper分布式锁，还是今天介绍的Elasticsearch实现的锁机制，都有如下三步：

  • 上锁
  • 执行事务方法
  • 解锁

全局锁

假定有两个线程，线程1和线程2

  1. 线程1上锁命令：
PUT /files/file/global/_create
{}
  • files表示索引名称。
  • file为type，6.3.1一个索引只允许有一个type，选用file作用type名称。
  • global：即document的id，固定写为global表示全局锁，或者使用专门的索引进行加锁操作。
  • _create: 强制必须是创建，如果已经存在，那么创建失败，报错。
  1. 线程1执行事务方法：更新文件名
POST /files/file/global/_update
{
  "doc": {
    "name":"ReadMe.txt"
  }
}
  1. 线程2尝试加锁，失败，此时程序进行重试阶段，直到线程1释放锁
# 请求：
PUT /files/file/global/_create
{}

# 响应：
{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[file][global]: version conflict, document already exists (current version [1])",
        "index_uuid": "_6E1d7BLQmy9-7gJptVp7A",
        "shard": "2",
        "index": "files"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[file][global]: version conflict, document already exists (current version [1])",
    "index_uuid": "_6E1d7BLQmy9-7gJptVp7A",
    "shard": "2",
    "index": "files"
  },
  "status": 409
}
  1. 线程1释放锁

DELETE files/file/global

  1. 线程2加锁
PUT /files/file/global/_create
{}

响应

{
  "_index": "files",
  "_type": "file",
  "_id": "global",
  "_version": 3,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 2,
  "_primary_term": 1
}
  1. 加锁成功，然后执行事务方法。

优缺点

全局锁本质上是所有线程都用_create语法来创建id为global的文档，利用Elasticsearch对_create语法的校验来实现锁的目的。

  • 优点：操作简单，容易使用，成本低。
  • 缺点：直接锁住整个索引，除了加锁的那个线程，其他所有对此索引的线程都block住了，并发量较低。
  • 适用场景：读多写少的数据，并且加解锁的时间非常短，类似于数据库的表锁。

注意事项：加锁解锁的控制必须严格在程序里定义，因为单纯基于doc的锁控制，如果id固定使用global，在有锁的情况，任何线程执行delete操作都是可以成功的，因为大家都知道id。

document level级别的锁

document level级别的锁是更细粒度的锁，以文档为单位进行锁控制。

我们新建一个索引专门用于加锁操作：

PUT /files-lock/_mapping/lock
{
  "properties": {
    
  }
}

我们先创建一个script脚本，ES6.0以后默认使用painless脚本：

POST _scripts/document-lock
{
  "script": {
    "lang": "painless",
    "source": "if ( ctx._source.process_id != params.process_id ) { Debug.explain('already locked by other thread'); }  ctx.op = 'noop';"
  }
}

Debug.explain表示抛出一个异常，内容为already locked by other thread。
ctx.op = 'noop'表示不执行更新。

  1. 线程1增加行锁，此时传入的process_id为181ab3ee-28cc-4339-ba35-69802e06fe42
POST /files-lock/lock/1/_update
{
  "upsert": { "process_id": "181ab3ee-28cc-4339-ba35-69802e06fe42" },
  "script": {
    "id": "document-lock", 
    "params": {
      "process_id": "181ab3ee-28cc-4339-ba35-69802e06fe42"
    }
  }
}

响应结果：

{
  "_index": "files-lock",
  "_type": "lock",
  "_id": "1",
  "_version": 1,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "_seq_no": 0,
  "_primary_term": 1
}
  1. 线程1、线程2查询锁信息
{
  "_index": "files-lock",
  "_type": "lock",
  "_id": "1",
  "_version": 1,
  "found": true,
  "_source": {
    "process_id": "181ab3ee-28cc-4339-ba35-69802e06fe42"
  }
}
  1. 线程2传入的process_id为181ab3ee-28cc-4339-ba35-69802e06fe42，尝试加锁，失败，此时应该启动重试机制
POST /files-lock/lock/1/_update
{
  "upsert": { "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5" },
  "script": {
    "id": "document-lock", 
    "params": {
      "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5"
    }
  }
}

提示该文档已经被别的线程（线程1）锁住了，你不能更新了，响应报文如下：

{
  "error": {
    "root_cause": [
      {
        "type": "remote_transport_exception",
        "reason": "[node-1][192.168.17.137:9300][indices:data/write/update[s]]"
      }
    ],
    "type": "illegal_argument_exception",
    "reason": "failed to execute script",
    "caused_by": {
      "type": "script_exception",
      "reason": "runtime error",
      "painless_class": "java.lang.String",
      "to_string": "already locked by other thread",
      "java_class": "java.lang.String",
      "script_stack": [
        "Debug.explain('already locked by other thread'); }  ",
        "              ^---- HERE"
      ],
      "script": "judge-lock",
      "lang": "painless",
      "caused_by": {
        "type": "painless_explain_error",
        "reason": null
      }
    }
  },
  "status": 400
}
  1. 线程1执行事务方法
POST /files/file/1/_update
{
  "doc": {
    "name":"README1.txt"
  }
}
  1. 线程1的事务方法执行完成，并通过删除id为1的文档，相当于释放锁

DELETE /files-lock/lock/1

  1. 线程2在线程1执行事务的期间，一直在模拟挂起，重试的操作，直到线程1完成释放锁，然后线程2加锁成功
POST /files-lock/lock/1/_update
{
  "upsert": { "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5" },
  "script": {
    "id": "document-lock", 
    "params": {
      "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5"
    }
  }
}

结果：

{
  "_index": "files-lock",
  "_type": "lock",
  "_id": "1",
  "_version": 3,
  "found": true,
  "_source": {
    "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5"
  }
}

此时锁的process_id变成线程2传入的"a6d13529-86c0-4422-b95a-aa0a453625d5"

{
  "_index": "files-lock",
  "_type": "lock",
  "_id": "1",
  "_version": 3,
  "found": true,
  "_source": {
    "process_id": "a6d13529-86c0-4422-b95a-aa0a453625d5"
  }
}

这样基于ES的行锁操作控制过程就完成了。

脚本解释

update+upsert操作，如果该记录没加锁（此时document为空），执行upsert操作，设置process_id，如果已加锁，执行script
script内的逻辑是：判断传入参数与当前doc的process_id，如果不相等，说明有别的线程尝试对有锁的doc进行加锁操作，Debug.explain表示抛出一个异常。

process_id可以由Java应用系统里生成，如UUID。

如果两个process_id相同，说明当前执行的线程与加锁的线程是同一个，ctx.op = 'noop'表示什么都不做，返回成功的响应，Java客户端拿到成功响应的报文，就可以继续下一步的操作，一般这里的下一步就是执行事务方法。

点评

文档级别的锁颗粒度小，并发性高，吞吐量大，类似于数据库的行锁。

共享锁与排他锁

概念

共享锁：允许多个线程获取同一条数据的共享锁进行读操作
排他锁：同一条数据只能有一个线程获取排他锁，然后进行增删改操作

互斥性：共享锁与排他锁是互斥的，如果这条数据有共享锁存在，那么排他锁无法加上，必须得共享锁释放完了，排他锁才能加上。
反之也成立，如果这条数据当前被排他锁锁信，那么其他的排他锁不能加，共享锁也加不上。必须等这个排他锁释放完了，其他锁才加得上。
有人在改数据，就不允许别人来改，也不让别人来读。

读写锁的分离
如果只是读数据，每个线程都可以加一把共享锁，此时该数据的共享锁数量一直递增，如果这时有写数据的请求（写请求是排他锁），由于互斥性，必须等共享锁全部释放完，写锁才加得上。
有人在读数据，就不允许别人来改。

案例实验

我们先创建一个共享锁的脚本：

# 读操作加锁脚本
POST _scripts/rw-lock
{
    "script": {
        "lang": "painless",
        "source": "if (ctx._source.lock_type == 'exclusive') { Debug.explain('one thread is writing data, the lock is exclusive now'); } ctx._source.lock_count++"
    }
}

# 读操作完毕释放锁脚本
POST _scripts/rw-unlock
{
    "script": {
        "lang": "painless",
        "source": "if ( --ctx._source.lock_count == 0) { ctx.op = 'delete' }"
    }
}
  1. 每次有一个线程读数据时，执行一次加锁操作
POST /files-lock/lock/1/_update
{
  "upsert": { 
    "lock_type":  "shared",
    "lock_count": 1
  },
  "script": {
    "id": "rw-lock"
  }
}

在多个页面上尝试，可以看到lock_count在逐一递增，模拟多个线程同时读一个文档的操作。

  1. 在有线程读文档，还未释放的情况下，尝试对该文档加一个排他锁
PUT /files-lock/lock/1/_create
{ "lock_type": "exclusive" }

结果肯定会报错：

{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[lock][1]: version conflict, document already exists (current version [8])",
        "index_uuid": "XD7LFToWSKe_6f1EvLNoFw",
        "shard": "3",
        "index": "files-lock"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[lock][1]: version conflict, document already exists (current version [8])",
    "index_uuid": "XD7LFToWSKe_6f1EvLNoFw",
    "shard": "3",
    "index": "files-lock"
  },
  "status": 409
}
  1. 线程读数据完成后，对共享锁进行释放，执行释放锁的脚本
POST /files-lock/lock/1/_update
{
  "script": {
    "id": "rw-unlock"
  }
}

释放1次lock_count减1，减到0时，说明所有的共享锁已经释放完毕，就把这个doc删除掉

  1. 所有共享锁释放完毕，尝试加排他锁
PUT /files-lock/lock/1/_create
{ "lock_type": "exclusive" }

此时能够加锁成功，响应报文：

{
  "_index": "files-lock",
  "_type": "lock",
  "_id": "1",
  "_version": 1,
  "found": true,
  "_source": {
    "lock_type": "exclusive"
  }
}
  1. 有排他锁的情况，尝试加一个共享锁，失败信息如下：
{
  "error": {
    "root_cause": [
      {
        "type": "remote_transport_exception",
        "reason": "[node-1][192.168.17.137:9300][indices:data/write/update[s]]"
      }
    ],
    "type": "illegal_argument_exception",
    "reason": "failed to execute script",
    "caused_by": {
      "type": "script_exception",
      "reason": "runtime error",
      "painless_class": "java.lang.String",
      "to_string": "one thread is writing data, the lock is exclusive now",
      "java_class": "java.lang.String",
      "script_stack": [
        "Debug.explain('one thread is writing data, the lock is exclusive now'); } ",
        "              ^---- HERE"
      ],
      "script": "rw-lock",
      "lang": "painless",
      "caused_by": {
        "type": "painless_explain_error",
        "reason": null
      }
    }
  },
  "status": 400
}
  1. 排他锁事务执行完成时，删除文档即可对锁进行释放

DELETE /files-lock/lock/1

脚本解释

读锁的加锁脚本和释放锁脚本，成对出现，用来统计线程的数量。

写锁利用_create语法来实现，如果有线程对某一文档有读取操作，那么对这个文档执行_create操作肯定报错。

小结

利用Elasticsearch一些语法的特性，加上painless脚本的配合，也能完整的复现全局锁、行锁、读写锁的特性，实现的思路还是挺有意思的，跟使用redis、zookeeper实现分布式锁有异曲同工之处，只是生产案例上用redis实现分布式锁是比较成功的实践，Elasticsearch的对这种分布式锁的实现方式可能不是最佳实践，但也可以了解一下。

专注Java高并发、分布式架构，更多技术干货分享与心得，请关注公众号：Java架构社区
可以扫左边二维码添加好友，邀请你加入Java架构社区微信群共同探讨技术
![Java架构社区](https://img-blog.csdnimg.cn/2...

process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YW5neWluZzIxMjQ=,size_16,color_FFFFFF,t_70)

0 阅读 945
推荐阅读
Elasticsearch系列---初识Elasticsearch 「扫盲」 Elasticsearch 全文搜索引擎 Elasticsearch 入门：集群搭建 Elasticsearch系列---生产集群部署(下) Elasticsearch 6.4.0 + Logstash + Kibana 安装（全） ElasticSearch安装、插件介绍及Kibana的安装与使用详解
Elasticsearch系列
关注数
1
内容数
14
关注专栏 专栏主页
目录
极术微信服务号
关注极术微信号
实时接收点赞提醒和评论通知
安谋科技学堂公众号
关注安谋科技学堂
实时获取安谋科技及 Arm 教学资源
安谋科技招聘公众号
关注安谋科技招聘
实时获取安谋科技中国职位信息