[大数据]基于Mysql、Redis和zookeeper实现分布式锁，解决超卖问题

分布式锁

Mysql中的超卖现象

假设一个商品库存为5000，使用jmeter一秒钟请求5000次进行测试，

1. 在非集群部署下，无锁，不使用数据库的情况下，使用吞吐量可以达到5400+，但是此时存在超卖的问题。
2. 将商品库存的数量存在数据库中，进行测试，吞吐量可以达到2000+，但是此时超卖问题更加严重。
3. 将商品库存的数量存在数据库中，并使用ReentrantLock对方法加锁，进行测试，吞吐量可以达到550+，此时没有超卖问题。
4. 将商品库存的数量存在数据库中，并使用synchronized对方法加锁，进行测试，吞吐量可以达到520+，略低于ReentrantLock，此时没有超卖问题。

导致JVM本地锁失效的三种情况

• 多例模式，吞吐量在1800+
• 数据库事务，吞吐量在750+（如果将事务等级改为read_uncommitted，可以解决超卖问题，但不会那么做，吞吐量在750+）
• 集群部署，nginx部署两台，吞吐量在610+，有超卖问题。

使用一个更新sql完成判断及更新

可以解决三个锁失效的三种情况。

吞吐量在1950+，没有超卖问题。

但是存在问题：

1. 锁范围问题（表级锁还是行级锁）。
2. 同一个商品有多个记录。
3. 无法记录数据库变化前后的状态。

悲观锁解决超卖问题

select … for update.

是表级锁，会锁住整个表.

如何在Mysql悲观锁中，使用行级锁？

1. 锁的查询条件或者更新条件必须是索引字段
2. 查询或者更新条件必须是具体值（如like或!=之类的不会走索引）

索引失效原因

1. 非最左前缀法则
2. 索引列函数运算
3. 模糊查询头部模糊
4. 使用了or连接符

吞吐量600+，性能比JVM本地锁略低。

Mysql悲观锁问题

1. 性能问题。
2. 死锁问题：对多条数据加锁时，加锁顺序要一致。
3. 库存操作要统一（select…for update 和 普通select）

使用Mysql乐观锁解决超卖问题

使用时间戳或者版本号利用CAS机制来实现。

不需要加手动事务注解防止阻塞导致超时，递归调用时需要间隔避免栈内存溢出。

吞吐量：230+，没有出现超卖问题。

Mysql乐观锁问题

1. 高并发情况下，性能极低
2. 由于CAS导致的ABA问题
3. 读写分离（主从复制）的情况下会导致乐观锁不可靠

Mysql锁总结

性能：一个sql > 悲观锁 > JVM锁 > 乐观锁

如果追求极致性能、业务场景简单且不需要记录数据前后变化的情况下，优先选择一个sql。

如果写并发量较低，争抢不是很激烈的情况下，优先选择乐观锁。

如果写并发量高，经常发生冲突的话，选择乐观锁会导致业务代码不间断的重试，应该优先选择mysql悲观锁。

不推荐jvm本地锁。

Redis中的超卖现象,并实现分布式锁

从redis中取，进行判断，如果>0，则从redis中扣库存。

第一次运行吞吐量2000+，系统预热后吞吐量达到3200+。但是出现了超卖问题。

JVM本地锁

与mysql类似。

redis乐观锁

watch：可以监控一个或者多个key的值，如果在事务（exec）执行之前，key的值发生变化，则取消事务执行。

multi：开启事务

exec：执行事务

使用redisTemple.execute(SessionCallback<Object>) 方法在Java中使用redis事务。

吞吐量370+

redis乐观锁的问题

1. 性能较低
2. 电脑性能低会导致连接不够用以至乐观锁失效。

分布式锁机制

跨进程，跨服务，跨服务器

适合场景：

1. 超卖现象（NoSQL）
2. 缓存击穿（一个热点key过期，导致mysql宕机）

实现方式：

1. 基于redis实现
2. 基于zookeeper/etcd实现
3. 基于mysql实现

特征：

1. 独占排他使用：setnx

2. 死锁

   1. 防死锁的发生（redis客户端从redis服务中获取到锁后宕机，导致无法释放）：添加过期时间expire
   2. 不可重入导致死锁：可重入性

3. 原子性：

   1. 获取锁和设置过期时间的原子性操作，setnx key value nx ex time
   2. 判断和释放锁之间：lua脚本

4. 防止误删：（解铃还须系铃人）先判断再删除（UUID）

5. 可重入性：hash数据模型+lua脚本

6. 自动续期：Timer定时器+lua脚本

7. 在集群情况下，导致锁机制失效

   1. 客户端1，在主中获取锁
   2. 还没来得及同步数据，主挂了
   3. 于是从升级为主
   4. 客户端程序2就从新主中获取到锁，导致锁机制失效

操作：

1. 加锁：setnx
2. 解锁：del
3. 重试：递归或循环

使用redis的setnx+del实现的分布式锁，吞吐量550左右（递归）

使用redis的setnx+del实现的分布式锁，吞吐量600左右（循环，可以避免栈溢出）

lua脚本

一次性发送多个指令给redis。redis单线程 执行指令遵守one-by-one规则。

EVAL script numkeys key [key …] arg [arg …] (控制台输出的是return结果）

script：lua脚本字符串

numkeys：key列表的元素数量

key列表：以空格分割。KEYS[index 从1开始]

arg列表：以空格分割。ARGV[index 从1开始]

变量**(在redis中只允许声明局部变量，不允许声明全局变量)**：

? 全局变量：a = 5

? 局部变量：local a = 5

分支控制：

if 条件
then
    代码块
elseif 条件
then
    代码块
else
    代码块
end

可以在在脚本中使用redis.call()来调用redis方法。

防止误删lua代码：

if redis.call('get','lock') == uuid
then
    return redis.call('del','lock')
else
    return 0
end

使用lua脚本防止误删，第一次运行吞吐量550，系统预热后吞吐量达到620。

可重入锁加锁流程

ReentrantLock.lock() --> NonfairSync.lock() --> AQS.acquire() --> NonfairSync.tryAcquire() --> Sync.nonfairTryAcquire()

1. CAS获取锁，如果没有线程占用锁（state==0），加锁成功并记录当前线程是有锁线程(两次)
2. 如果state的值不为0，说明锁已经被占用。则判断当前线程是否是有锁线程，如果是则重入（state+1）
3. 否则加锁失败，入队等待

可重入锁解锁流程

ReentrantLock.unlock() --> AQS.release() --> Sync.tryRelease()

1. 判断当前线程是否是有锁线程，不是则抛出异常
2. 对state的值减1之后，判断state的值，是否为0，为0则解锁成功，返回true
3. 如果减1后值不为0，则返回false

参照ReentrantLock中的非公平可重入锁，实现分布式可重入锁

hash + lua脚本

加锁：

1. 判断锁是否存在（exists），则直接获取锁 hset key field value
2. 如果所存在，则判断是否是自己的锁（hexists），如果是自己的锁则重入（hincrby key field increment）
3. 否则重试：递归 循环

if redis.call('exist',KEYS[1]) == 0 or redis.call('hexist',KEYS[1],ARGV[1]) == 1
then
    redis.call('hincrby',KEYS[1],ARGV[1],1)
    redis.call('expire',KEYS[1],ARGV[2])
    return 1
else
    return 0
end 

key: lock
arg: uuid 30

解锁：

1. 判断自己的锁是否存在（hexists），不存在则返回nil
2. 如果自己的锁存在，则减1（hincrby -1），判断减1后的值 是否为0，为0则释放锁（del）返回1
3. 不为0，则返回0

if redis.call('hexist',KEYS[1],ARGV[1]) == 0
then
	return nil
elseif redis.call('hincrby',KEYS[1],ARGV[1],-1) == 0
then
    return redis.call('del',KEYS[1])
else
    return 0
end

key: lock
arg: uuid

吞吐量可达到600左右。

自动续期

定时任务（时间驱动，Timer定时器） + lua脚本

? 判断自己的锁是否存在（hexists）,如果存在则重置过期时间

if redis.call('hexists',KEYS[1],ARGV[1]) == 1
then
    return redis.call('expire',KEYS[1],ARGV[2])
else
    return 0
end

key: lock
arg: uuid 30

红锁算法Redlock

1. 应用程序获取系统当前时间
2. 应用程序使用相同的kv值依次从多个redis实例中获取锁。如果一个节点超过一定时间依然没有获取到锁，则直接放弃，尽快尝试下一个健康的redis节点获取锁，以避免被一个宕机了的节点阻塞住。
3. 计算获取锁的消耗时间 = 客户端程序的系统当前时间 - step1中的时间。获取锁的消耗时间小于总的锁定时间（30s）并且半数以上节点获取锁成功，认为获取成功，否则失败。
4. 计算剩余锁定时间 = 总锁定时间-setp3中的时间。
5. 如果获取锁失败了，对所有的redis节点释放锁

Redisson

它是一个redis的java客户端，包含分布式锁的封装。

使用redisson进行加锁解锁操作，吞吐量870左右。

实现了分布式的可重入锁、公平锁、读写锁、信号量、倒计数器、联锁、红锁等。

基于zookeeper实现分布式锁

znode节点类型

1. 永久节点：客户端与zookeeper断开连接后，该节点依旧存在
2. 临时节点：客户端与zookeeper断开连接后，该节点被删除
3. 永久序列化节点：客户端与zookeeper断开连接后，该节点依旧存在，只是zookeeper给该节点名称进行顺序编号
4. 临时序列化节点：客户端与zookeeper断开连接后，该节点被删除，只是zookeeper给该节点名称进行顺序编号

节点的事件监听：一次性

1. 节点创建：NodeCreated

   stat -w /xx

2. 节点删除：NodeDeleted

   stat -w /xx

3. 节点数据变化：NodeDataChanged

   get -w /xx

4. 子节点变化:：NodeChildrenChanged

   ls -w /xx

Java客户端

1. 官方提供
2. ZkClient
3. Curator

分布式锁

1. 独占排他：znode节点不可重复 自旋锁
2. 阻塞锁：临时序列化节点
   1. 所有请求要求获取锁时，给每一个求情创建临时序列化节点
   2. 获取当前节点的前置节点，如果前置节点为空，则获取锁成功，否则，监听前置节点
   3. 获取锁成功后，执行业务操作，然后释放当前节点的锁
3. 可重入：同一线程已经获取过该锁的情况下，可重入
   1. 在节点的内容中记录服务器、线程、重入信息
   2. ThreadLocal：线程的局部变量，线程私有
4. 公平锁：
   1. 有序列

zookeeper实现分布式锁的特征

1. 独占排他互斥使用：节点不重复
2. 防死锁：
   1. 客户端程序获取到锁后服务器宕机：临时节点，一旦客户端服务器宕机，链接就会关闭，此时zk心跳检测不到客户端程序，删除对应的临时节点。
   2. 不可重入：ThreadLocal
3. 防误删：给每一个请求线程创建一个唯一的序列化节点。
4. 原子性：
   1. 创建节点、删除节点、查询及监听：具备原子性
5. 可重入：ThreadLocal、节点数据、ConcurrentHashMap
6. 自动续期：临时节点，没有过期时间，不需要续期
7. 单点故障：zk一般都是集群部署
8. zk集群：偏向于一致性的集群（优于Redis集群）

? 缺点：单个线程OOm后，连接没有断开，zk不会删除锁。

Curator

Curator-framework：zk底层进行封装

Curator-recipes：典型的应用场景进行封装，分布式锁

InterProcessMutex：类似于ReentrantLock可重入锁 分布式版本。加锁：acquire()，解锁release()

InterProcessSemaphoreMutex：不可重入锁

InterProcessRead WriteMutex：可重入读写锁

InterProcessMultiLock：联锁

InterProcessSemephoreV2：信号量，限流

共享计数器：

1. ShareCount
2. DistributedAtomicNumber：DistributedAtomicLong和DistributedAtomicInteger

InterProcessMutex

basePath：初始化锁时指定多个节点路径

internals：LockInternals对象，加锁 解锁

ConcurrentMap<Thread,LockData> threadData：记录了重入信息

1. owningThread：记录线程对象
1. lockPath：锁路径，哪个锁的重入信息
1. lockCount：重入次数

LockInternals

maxLeases：租约，值为1

basePath：初始化锁时指定多个节点路径

path：basePath + “/lock-”

加锁

InterProcessMutex.acquire() --> InterProcessMutex.internalLock() --> LockInternals.attemptLock()

Mysql实现分布式锁：唯一键索引

redis：基于Key的唯一性

zk：基于znode节点唯一性

思路：

1. 加锁：insert into tb_lock(lock_name) values ('lock') 执行成功代表获取锁成功。
1. 获取锁成功的请求执行业务操作，执行完成之后通过delete删除对应记录。
1. 重试：递归

1. 独占排他互斥使用：唯一键索引
2. 防死锁：
   1. 客户端程序获取到所之后，客户端程序的服务器宕机：给锁记录添加一个获取锁时间列。额外的定时器检查获取锁的系统时间和当前系统时间的差值是否超过某个阈值。
   2. 不可重入：记录服务信息，线程信息和重入次数
3. 防误删：借助于id的唯一性防止误删
4. 原子性：一个写操作 还可以借助于mysql的悲观锁
5. 可重入：记录服务信息，线程信息和重入次数
6. 自动续期：服务器内的定时器重置获取锁的系统时间
7. 单机故障，搭建mysql主备
8. 集群情况下锁机制失效问题。
9. 阻塞锁：难实现。

总结

1. 建议程序：mysql分布式锁 > redis(lua脚本) > zk
2. 性能：redis > zk > mysql
3. 可靠性：zk > redis = mysql

最求极致性能：redis

追求可靠性：zk

简单玩一下，实现独占排他，对性能对可靠性要求都不高：mysql