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

你们春节在家抢红包玩的不亦乐乎，抢红包服务看起来很是简单，实际上要作好这个服务，特别是money相关服务是不容许出错的，想一想看每一个红包的数字都是真金白银，要求服务的鲁棒性很是高，背后包含着不少后台服务技术细节。

文章每周持续更新，各位的「三连」是对我最大的确定。能够微信搜索公众号「 后端技术学堂 」第一时间阅读（通常比博客早更新一到两篇）

抛砖引玉，今天就来讲说其中一个技术细节，也是在我另外一篇文章**Linux后台开发C++学习路线技能加点中提到但没展开讲的，高并发服务编程中的redis分布式锁**。

这里罗列出3种redis实现的分布式锁，并分别对比说明各自特色。

Redis单实例分布式锁

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

setnx用法参考redis官方文档

语法

SETNX key value

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

返回值：

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

加锁步骤

1. SETNX lock.foo <current Unix time + lock timeout + 1>

若是客户端得到锁，SETNX返回1，加锁成功。

若是SETNX返回0，那么该键已经被其余的客户端锁定。

1. 接上一步，SETNX返回0加锁失败，此时，调用GET lock.foo获取时间戳检查该锁是否已通过期：

2. 若是没有过时，则休眠一会重试。

3. 若是已通过期，则能够获取该锁。具体的：调用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命令增长了一系列选项:

• EXseconds – Set the specified expire time, in seconds.
• PXmilliseconds – 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.
• EXseconds – 设置键key的过时时间，单位时秒
• PXmilliseconds – 设置键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. 客户端1从Master获取了锁。
2. Master宕机了，存储锁的key尚未来得及同步到Slave上。
3. Slave升级为Master。
4. 客户端2重新的Master获取到了对应同一个资源的锁。

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

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

加锁步骤

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

具体的：

1. 获取当前时间（毫秒数）。
2. 按顺序依次向N个Redis节点执行获取锁的操做。这个获取操做跟前面基于单Redis节点的获取锁的过程相同，包含随机字符串my_random_value，也包含过时时间(好比PX 30000，即锁的有效时间)。为了保证在某个Redis节点不可用的时候算法可以继续运行，这个获取锁的操做还有一个超时时间(time out)，它要远小于锁的有效时间（几十毫秒量级）。客户端在向某个Redis节点获取锁失败之后，应该当即尝试下一个Redis节点。这里的失败，应该包含任何类型的失败，好比该Redis节点不可用，或者该Redis节点上的锁已经被其它客户端持有（注：Redlock原文中这里只提到了Redis节点不可用的状况，但也应该包含其它的失败状况）。
3. 计算整个获取锁的过程总共消耗了多长时间，计算方法是用当前时间减去第1步记录的时间。若是客户端从大多数Redis节点（>= N/2+1）成功获取到了锁，而且获取锁总共消耗的时间没有超过锁的有效时间(lock validity time)，那么这时客户端才认为最终获取锁成功；不然，认为最终获取锁失败。
4. 若是最终获取锁成功了，那么这个锁的有效时间应该从新计算，它等于最初的锁的有效时间减去第3步计算出来的获取锁消耗的时间。
5. 若是最终获取锁失败了（可能因为获取到锁的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++实现

源码在这

建立分布式锁管理类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实例场景推荐用实现三最安全，不过实现三也不是完美无瑕，也有针对这种算法缺陷的讨论（节点宕机同步时延、时间同步假设），你们还须要根据自身业务场景灵活选择或定制本身的分布式锁。

参考

Distributed locks with Redis

How to do distributed locking

基于Redis的分布式锁到底安全吗

---

最后，感谢各位的阅读，文章的目的是分享对知识的理解，若文中出现明显纰漏也欢迎指出，咱们一块儿在探讨中学习。

原创不易，看到这里动动手指，各位的「三连」是对我持续创做的最大支持。

能够微信搜索公众号「 后端技术学堂 」回复「资料」有我给你准备的各类编程学习资料。文章每周持续更新，咱们下期见！