Elasticsearch-87-基于共享锁和排它锁实现悲观锁并发控制

发表于 | 分类于

共享锁和排它锁

共享锁: 这份数据是共享的,然后多个线程过来,都可以获取同一个数据的共享锁,然后对这个数据执行读操作.
排它锁: 是排他的操作,只能一个线程获取排他锁,然后执行增删改操作

读写锁分离

如果是要读取数据的话,那么任意多少个线程都可以进来然后读取数据,每个线程都可以上一个共享锁,但是这个时候如果有线程过来修改数据,会尝试上排他锁排它锁会跟共享锁互斥,也就是说,如果有人已经上了共享锁了,那么排它锁就不能上,就得等.

就是说如果有人在读数据,就不允许别人来修改数据,反之也是一样的.

如果有人在修改数据,就是加了排他锁,这时候其他线程过来也要修改数据,也会尝试加排它锁,此时就会失败,锁冲突,就必须等待重试,同时只能有一个线程修改数据,如果有线程过来读取数据,就是尝试加共享锁,此时也会失败,因为共享锁和排它锁是互斥的

如果有线程再修改数据,就不许别的线程再来修改,也不许别的线程读取数据

案例

多个线程上共享锁

先试一下共享锁,多个线程同时读取数据. 上共享锁也需要一个groovy脚本,和之前一样,脚本内容是

1
if (ctx._source.lock_type == 'exclusive') { assert false }; ctx._source.lock_count++

就是如果锁已经存在了,判断这个锁的类型是排它锁的话,直接报错,是共享锁的话,就把锁的数量加1

假设,先有一个线程过来上了一个共享锁

1
2
3
4
5
6
7
8
9
10
11
POST /fs/lock/1/_update
{
  "upsert": {
    "lock_type":"shared",
    "lock_count": 1
  },
  "script": {
    "lang": "groovy",
    "file": "judge-lock-2"
  }
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
{
  "_index": "fs",
  "_type": "lock",
  "_id": "1",
  "_version": 6,
  "found": true,
  "_source": {
    "lock_type": "shared",
    "lock_count": 1
  }
}

添加的数据,一个是锁的类型,一个是锁的数量,已经成功上了一个共享锁了

这时候又有一个线程过来上共享锁

1
2
3
4
5
6
7
8
9
10
11
POST /fs/lock/1/_update
{
  "upsert": {
    "lock_type":"shared",
    "lock_count": 1
  },
  "script": {
    "lang": "groovy",
    "file": "judge-lock-2"
  }
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
12
{
  "_index": "fs",
  "_type": "lock",
  "_id": "1",
  "_version": 7,
  "result": "updated",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  }
}

执行还是成功的,然后查询一下这个锁

1
GET /fs/lock/1

返回值:

1
2
3
4
5
6
7
8
9
10
11
{
  "_index": "fs",
  "_type": "lock",
  "_id": "1",
  "_version": 7,
  "found": true,
  "_source": {
    "lock_type": "shared",
    "lock_count": 2
  }
}

lock_count是2,就是加了两个共享锁. 就是有线程上了共享锁以后,其他线程还要再上,直接上就可以了,只是lock_count + 1

共享锁没有释放的情况下 上排他锁

上排他锁的请求

1
2
3
4
PUT /fs/lock/1/_create
{
  "lock_type":"exclusive"
}

这里用的是_create请求,就是强制创建,要求lock必须不能存在,但是之前已经有了共享锁了,/fs/lock/1是存在的 所以这里会创建失败,报错

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[lock][1]: version conflict, document already exists (current version [7])",
        "index_uuid": "RWijfql1Qlqa_OPoor0Ubw",
        "shard": "3",
        "index": "fs"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[lock][1]: version conflict, document already exists (current version [7])",
    "index_uuid": "RWijfql1Qlqa_OPoor0Ubw",
    "shard": "3",
    "index": "fs"
  },
  "status": 409
}

释放共享锁

这里也是需要一个释放共享锁的groovy脚本

1
if(--ctx._source.lock_count == 0){ctx.op = 'delete'}

就是执行脚本的时候,先把lock_count-1 然后判断是不是0,是的话直接删除掉.

1
2
3
4
5
6
7
POST /fs/lock/1/_update
{
  "script": {
    "lang": "groovy",
    "file": "unlock-shared"
  }
}

之前上了两个锁,执行两次,lock_count就是0了,然后就被删除了,所有的共享锁就都被释放了

再上排它锁
1
2
3
4
PUT /fs/lock/1/_create
{
  "lock_type":"exclusive"
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
12
13
{
  "_index": "fs",
  "_type": "lock",
  "_id": "1",
  "_version": 10,
  "result": "created",
  "_shards": {
    "total": 2,
    "successful": 1,
    "failed": 0
  },
  "created": true
}

创建成功了,这时候如果再有其他线程过来申请排它锁

1
2
3
4
PUT /fs/lock/1/_create
{
  "lock_type":"exclusive"
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
  "error": {
    "root_cause": [
      {
        "type": "version_conflict_engine_exception",
        "reason": "[lock][1]: version conflict, document already exists (current version [10])",
        "index_uuid": "RWijfql1Qlqa_OPoor0Ubw",
        "shard": "3",
        "index": "fs"
      }
    ],
    "type": "version_conflict_engine_exception",
    "reason": "[lock][1]: version conflict, document already exists (current version [10])",
    "index_uuid": "RWijfql1Qlqa_OPoor0Ubw",
    "shard": "3",
    "index": "fs"
  },
  "status": 409
}

还是报错的,上锁失败,如果再上一个共享锁呢

1
2
3
4
5
6
7
8
9
10
11
POST /fs/lock/1/_update
{
  "upsert": {
    "lock_type":"shared",
    "lock_count": 1
  },
  "script": {
    "lang": "groovy",
    "file": "judge-lock-2"
  }
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
{
  "error": {
    "root_cause": [
      {
        "type": "remote_transport_exception",
        "reason": "[f57uV91][127.0.0.1:9300][indices:data/write/update[s]]"
      }
    ],
    "type": "illegal_argument_exception",
    "reason": "failed to execute script",
    "caused_by": {
      "type": "script_exception",
      "reason": "error evaluating judge-lock-2",
      "caused_by": {
        "type": "power_assertion_error",
        "reason": "assert false\n"
      },
      "script_stack": [],
      "script": "",
      "lang": "groovy"
    }
  },
  "status": 400
}

也是报错,加不了的

释放排他锁

直接把这个删除掉就好了

1
DELETE /fs/lock/1