面试官问我redis锁怎么实现?我一口气和他说了3种方法!

原创文章首发微信公众号「后端技术学堂」转载请先与我联系，点文末链接「了解更多」

今天就来说说高并发编程中redis分布式锁实现，这里罗列出3种redis实现的分布式锁，并分别对比说明各自特点。

Redis单实例分布式锁

实现一：SETNX实现的分布式锁

setnx用法参考redis官方文档

语法

SETNX key value

将key设置值为value，如果key不存在，这种情况下等同SET命令。当key存在时，什么也不做。SETNX是”SET if Not eXists”的简写。

返回值：

• 1 设置key成功
• 0 设置key失败

加锁步骤

0. SETNX lock.foo <current Unix time + lock timeout + 1>如果客户端获得锁，SETNX返回1，加锁成功。如果SETNX返回0，那么该键已经被其他的客户端锁定。
1. 接上一步，SETNX返回0加锁失败，此时，调用GET lock.foo获取时间戳检查该锁是否已经过期：
2. 若旧的时间戳已过期，则表示加锁成功。
3. 若旧的时间戳还未过期（说明被其他客户端抢去并设置了时间戳），代表加锁失败，需要等待重试。
4. 如果没有过期，则休眠一会重试。
5. 如果已经过期，则可以获取该锁。具体的：调用GETSET lock.foo <current Unix timestamp + lock timeout + 1>基于当前时间设置新的过期时间。注意: 这里设置的时候因为在SETNX与GETSET之间有个窗口期，在这期间锁可能已被其他客户端抢去，所以这里需要判断GETSET的返回值，它的返回值是SET之前旧的时间戳：

解锁步骤

解锁相对简单，只需GET lock.foo时间戳，判断是否过期，过期就调用删除DEL lock.foo

实现二：SET实现的分布式锁

set用法参考官方文档

语法

SET key value [EX seconds|PX milliseconds] [NX|XX]

将键key设定为指定的“字符串”值。如果 key 已经保存了一个值，那么这个操作会直接覆盖原来的值，并且忽略原始类型。当set命令执行成功之后，之前设置的过期时间都将失效。

从2.6.12版本开始，redis为SET命令增加了一系列选项:

• EX seconds – Set the specified expire time, in seconds.
• PX milliseconds – Set the specified expire time, in milliseconds.
• NX – Only set the key if it does not already exist.
• XX – Only set the key if it already exist.
• EX seconds – 设置键key的过期时间，单位时秒
• PX milliseconds – 设置键key的过期时间，单位是毫秒
• NX – 只有键key不存在的时候才会设置key的值
• XX – 只有键key存在的时候才会设置key的值

版本>= 6.0

• KEEPTTL -- 保持 key 之前的有效时间TTL

加锁步骤

一条命令即可加锁: SET resource_name my_random_value NX PX 30000

The command will set the key only if it does not already exist (NX option), with an expire of 30000 milliseconds (PX option). The key is set to a value “myrandomvalue”. This value must be unique across all clients and all lock requests.

这个命令只有当key 对应的键不存在resource_name时（NX选项的作用）才生效，同时设置30000毫秒的超时，成功设置其值为my_random_value，这是个在所有redis客户端加锁请求中全局唯一的随机值。

解锁步骤

解锁时需要确保my_random_value和加锁的时候一致。下面的Lua脚本可以完成

if redis.call("get",KEYS[1]) == ARGV[1] then
  return redis.call("del",KEYS[1])
else
  return 0
end

这段Lua脚本在执行的时候要把前面的my_random_value作为ARGV[1]的值传进去，把resource_name作为KEYS[1]的值传进去。释放锁其实包含三步操作：’GET’、判断和’DEL’，用Lua脚本来实现能保证这三步的原子性。

Redis集群分布式锁

实现三：Redlock

前面两种分布式锁的实现都是针对单redis master实例，既不是有互为备份的slave节点也不是多master集群，如果是redis集群，每个redis master节点都是独立存储，这种场景用前面两种加锁策略有锁的安全性问题。

比如下面这种场景：

客户端1从Master获取了锁。

Master宕机了，存储锁的key还没有来得及同步到Slave上。

Slave升级为Master。

客户端2从新的Master获取到了对应同一个资源的锁。

于是，客户端1和客户端2同时持有了同一个资源的锁。锁的安全性被打破。

针对这种多redis服务实例的场景，redis作者antirez设计了Redlock （Distributed locks with Redis）算法，就是我们接下来介绍的。

加锁步骤

集群加锁的总体思想是尝试锁住所有节点，当有一半以上节点被锁住就代表加锁成功。集群部署你的数据可能保存在任何一个redis服务节点上，一旦加锁必须确保集群内任意节点被锁住，否则也就失去了加锁的意义。

具体的：

0. 获取当前时间（毫秒数）。
1. 按顺序依次向N个Redis节点执行获取锁的操作。这个获取操作跟前面基于单Redis节点的获取锁的过程相同，包含随机字符串my_random_value，也包含过期时间(比如PX 30000，即锁的有效时间)。为了保证在某个Redis节点不可用的时候算法能够继续运行，这个获取锁的操作还有一个超时时间(time out)，它要远小于锁的有效时间（几十毫秒量级）。客户端在向某个Redis节点获取锁失败以后，应该立即尝试下一个Redis节点。这里的失败，应该包含任何类型的失败，比如该Redis节点不可用，或者该Redis节点上的锁已经被其它客户端持有（注：Redlock原文中这里只提到了Redis节点不可用的情况，但也应该包含其它的失败情况）。
2. 计算整个获取锁的过程总共消耗了多长时间，计算方法是用当前时间减去第1步记录的时间。如果客户端从大多数Redis节点（>= N/2+1）成功获取到了锁，并且获取锁总共消耗的时间没有超过锁的有效时间(lock validity time)，那么这时客户端才认为最终获取锁成功；否则，认为最终获取锁失败。
3. 如果最终获取锁成功了，那么这个锁的有效时间应该重新计算，它等于最初的锁的有效时间减去第3步计算出来的获取锁消耗的时间。
4. 如果最终获取锁失败了（可能由于获取到锁的Redis节点个数少于N/2+1，或者整个获取锁的过程消耗的时间超过了锁的最初有效时间），那么客户端应该立即向所有Redis节点发起释放锁的操作（即前面介绍的Redis Lua脚本）。

我们接着讲

解锁步骤

客户端向所有Redis节点发起释放锁的操作，不管这些节点当时在获取锁的时候成功与否。

算法实现

上面描述的算法已经有现成的实现，各种语言版本。

• Redlock-rb (Ruby implementation). There is also a fork of Redlock-rb that adds a gem for easy distribution and perhaps more.
• Redlock-py (Python implementation).
• Aioredlock (Asyncio Python implementation).
• Redlock-php (PHP implementation).
• PHPRedisMutex (further PHP implementation)
• cheprasov/php-redis-lock (PHP library for locks)
• Redsync (Go implementation).
• Redisson (Java implementation).
• Redis::DistLock (Perl implementation).
• Redlock-cpp (C++ implementation).
• Redlock-cs (C#/.NET implementation).
• RedLock.net (C#/.NET implementation). Includes async and lock extension support.
• ScarletLock (C# .NET implementation with configurable datastore)
• Redlock4Net (C# .NET implementation)
• node-redlock (NodeJS implementation). Includes support for lock extension.

比如我用的C++实现

源码在这 https://github.com/jacket-code/redlock-cpp

创建分布式锁管理类CRedLock

CRedLock * dlm = new CRedLock();
dlm->AddServerUrl("127.0.0.1", 5005);
dlm->AddServerUrl("127.0.0.1", 5006);
dlm->AddServerUrl("127.0.0.1", 5007);

加锁并设置超时时间

CLock my_lock;
bool flag = dlm->Lock("my_resource_name", 1000, my_lock);

加锁并保持直到释放

CLock my_lock;
bool flag = dlm->ContinueLock("my_resource_name", 1000, my_lock);

my_resource_name是加锁标识；1000是锁的有效期，单位毫秒。

加锁失败返回false， 加锁成功返回Lock结构如下

class CLock {
public:
   int m_validityTime; => 9897.3020019531 // 当前锁可以存活的时间, 毫秒
   sds m_resource; => my_resource_name // 要锁住的资源名称
   sds m_val; => 53771bfa1e775 // 锁住资源的进程随机名字
};

解锁

dlm->Unlock(my_lock);

总结

综上所述，三种实现方式。

• 单redis实例场景，分布式锁实现一和实现二都可以，实现二更简洁推荐用实现二，用实现三也可以，但是实现三对单实例场景有点复杂略显笨重。
• 多redis实例场景推荐用实现三最安全，不过实现三也不是完美无瑕，也有针对这种算法缺陷的讨论（节点宕机同步时延、时间同步假设），大家还需要根据自身业务场景灵活选择或定制实现自己的分布式锁。

参考

https://redis.io/topics/distlock

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html

http://zhangtielei.com/posts/blog-redlock-reasoning.html

创作不易，点赞关注支持一下吧

我会持续分享软件编程和程序员那些事，欢迎关注。若你对编程感兴趣，我整理了这些年学习编程的各种资源，关注公众号「后端技术学堂」发送「资源」分享给你，点下方「了解更多」链接。