SETNX key value

起始版本：1.0.0
时间复杂度：O(1)
将key设置值为value，如果key不存在，这种情况下等同SET命令。 当key存在时，什么也不做。SETNX是”SET if Not eXists”的简写。

返回值

Integer reply, 特定值:

• 1 如果key被设置了
• 0 如果key没有被设置

  例子

  redis> SETNX mykey "Hello"
  (integer) 1
  redis> SETNX mykey "World"
  (integer) 0
  redis> GET mykey
  "Hello"
  redis>

  

  Design pattern: Locking with !SETNX

  设计模式：使用!SETNX加锁

  Please note that:
  请注意：

1. 不鼓励以下模式来实现the Redlock algorithm ，该算法实现起来有一些复杂，但是提供了更好的保证并且具有容错性。
   
2. 无论如何，我们保留旧的模式，因为肯定存在一些已实现的方法链接到该页面作为引用。而且，这是一个有趣的例子说明Redis命令能够被用来作为编程原语的。
   
3. 无论如何，即使假设一个单例的加锁原语，但是从 2.6.12 开始，可以创建一个更加简单的加锁原语，相当于使用SET命令来获取锁，并且用一个简单的 Lua 脚本来释放锁。该模式被记录在SET命令的页面中。
   

也就是说，SETNX能够被使用并且以前也在被使用去作为一个加锁原语。例如，获取键为foo的锁，客户端可以尝试一下操作：

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

如果客户端获得锁，SETNX返回1，那么将lock.foo键的Unix时间设置为不在被认为有效的时间。客户端随后会使用DEL lock.foo去释放该锁。
如果SETNX返回0，那么该键已经被其他的客户端锁定。如果这是一个非阻塞的锁，才能立刻返回给调用者，或者尝试重新获取该锁，直到成功或者过期超时。

处理死锁

以上加锁算法存在一个问题：如果客户端出现故障，崩溃或者其他情况无法释放该锁会发生什么情况？这是能够检测到这种情况，因为该锁包含一个Unix时间戳，如果这样一个时间戳等于当前的Unix时间，该锁将不再有效。
当以下这种情况发生时，我们不能调用DEL来删除该锁，并且尝试执行一个SETNX，因为这里存在一个竞态条件，当多个客户端察觉到一个过期的锁并且都尝试去释放它。

• C1 和 C2 读lock.foo检查时间戳，因为他们执行完SETNX后都被返回了0，因为锁仍然被 C3 所持有，并且 C3 已经崩溃。
• C1 发送DEL lock.foo
• C1 发送SETNX lock.foo命令并且成功返回
• C2 发送DEL lock.foo
• C2 发送SETNX lock.foo命令并且成功返回
• 错误：由于竞态条件导致 C1 和 C2 都获取到了锁

幸运的是，可以使用以下的算法来避免这种情况，请看 C4 客户端所使用的好的算法：

• C4 发送SETNX lock.foo为了获得该锁
• 已经崩溃的客户端 C3 仍然持有该锁，所以Redis将会返回0给 C4
• C4 发送GET lock.foo检查该锁是否已经过期。如果没有过期，C4 客户端将会睡眠一会，并且从一开始进行重试操作

• 另一种情况，如果因为 lock.foo键的Unix时间小于当前的Unix时间而导致该锁已经过期，C4 会尝试执行以下的操作：

  GETSET lock.foo <current Unix timestamp + lock timeout + 1>
  

• 由于GETSET 的语意，C4会检查已经过期的旧值是否仍然存储在lock.foo中。如果是的话，C4 会获得锁

• 如果另一个客户端，假如为 C5 ，比 C4 更快的通过GETSET操作获取到锁，那么 C4 执行GETSET操作会被返回一个不过期的时间戳。C4 将会从第一个步骤重新开始。请注意：即使 C4 在将来几秒设置该键，这也不是问题。

为了使这种加锁算法更加的健壮，持有锁的客户端应该总是要检查是否超时，保证使用DEL释放锁之前不会过期，因为客户端故障的情况可能是复杂的，不止是崩溃，还会阻塞一段时间，阻止一些操作的执行，并且在阻塞恢复后尝试执行DEL（此时，该LOCK已经被其他客户端所持有）

Redis分布式锁

分布式锁在很多场景中是非常有用的原语， 不同的进程必须以独占资源的方式实现资源共享就是一个典型的例子。
有很多分布式锁的库和描述怎么实现分布式锁管理器（DLM)的博客,但是每个库的实现方式都不太一样，很多库的实现方式为了简单降低了可靠性，而有的使用了稍微复杂的设计。
这个页面试图提供一个使用Redis实现分布式锁的规范算法。我们提出一种算法，叫Redlock,我们认为这种实现比普通的单实例实现更安全,我们希望redis社区能帮助分析一下这种实现方法，并给我们提供反馈。

实现细节

在我们开始描述算法之前，我们已经有一些可供参考的实现库.

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

安全和活性失效保障

按照我们的思路和设计方案，算法只需具备3个特性就可以实现一个最低保障的分布式锁。

1. 安全属性（Safety property）: 独享（相互排斥）。在任意一个时刻，只有一个客户端持有锁。
2. 活性A(Liveness property A): 无死锁。即便持有锁的客户端崩溃（crashed)或者网络被分裂（gets partitioned)，锁仍然可以被获取。
3. 活性B(Liveness property B): 容错。 只要大部分Redis节点都活着，客户端就可以获取和释放锁.

为什么基于故障转移的实现还不够

为了更好的理解我们想要改进的方面，我们先分析一下当前大多数基于Redis的分布式锁现状和实现方法.
实现Redis分布式锁的最简单的方法就是在Redis中创建一个key，这个key有一个失效时间（TTL)，以保证锁最终会被自动释放掉（这个对应特性2）。当客户端释放资源(解锁）的时候，会删除掉这个key。
从表面上看，似乎效果还不错，但是这里有一个问题：这个架构中存在一个严重的单点失败问题。如果Redis挂了怎么办？你可能会说，可以通过增加一个slave节点解决这个问题。但这通常是行不通的。这样做，我们不能实现资源的独享,因为Redis的主从同步通常是异步的。
在这种场景（主从结构）中存在明显的竞态:

1. 客户端A从master获取到锁
2. 在master将锁同步到slave之前，master宕掉了。
3. slave节点被晋级为master节点
4. 客户端B取得了同一个资源被客户端A已经获取到的另外一个锁。安全失效！

有时候程序就是这么巧，比如说正好一个节点挂掉的时候，多个客户端同时取到了锁。如果你可以接受这种小概率错误，那用这个基于复制的方案就完全没有问题。否则的话，我们建议你实现下面描述的解决方案。

单Redis实例实现分布式锁的正确方法

在尝试克服上述单实例设置的限制之前，让我们先讨论一下在这种简单情况下实现分布式锁的正确做法，实际上这是一种可行的方案，尽管存在竞态，结果仍然是可接受的，另外，这里讨论的单实例加锁方法也是分布式加锁算法的基础。
获取锁使用命令:

SET resource_name my_random_value NX PX 30000

这个命令仅在不存在key的时候才能被执行成功（NX选项），并且这个key有一个30秒的自动失效时间（PX属性）。这个key的值是“my_random_value”(一个随机值），这个值在所有的客户端必须是唯一的，所有同一key的获取者（竞争者）这个值都不能一样。
value的值必须是随机数主要是为了更安全的释放锁，释放锁的时候使用脚本告诉Redis:只有key存在并且存储的值和我指定的值一样才能告诉我删除成功。可以通过以下Lua脚本实现：

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

使用这种方式释放锁可以避免删除别的客户端获取成功的锁。举个例子：客户端A取得资源锁，但是紧接着被一个其他操作阻塞了，当客户端A运行完毕其他操作后要释放锁时，原来的锁早已超时并且被Redis自动释放，并且在这期间资源锁又被客户端B再次获取到。如果仅使用DEL命令将key删除，那么这种情况就会把客户端B的锁给删除掉。使用Lua脚本就不会存在这种情况，因为脚本仅会删除value等于客户端A的value的key（value相当于客户端的一个签名）。
这个随机字符串应该怎么设置？我认为它应该是从/dev/urandom产生的一个20字节随机数，但是我想你可以找到比这种方法代价更小的方法，只要这个数在你的任务中是唯一的就行。例如一种安全可行的方法是使用/dev/urandom作为RC4的种子和源产生一个伪随机流;一种更简单的方法是把以毫秒为单位的unix时间和客户端ID拼接起来，理论上不是完全安全，但是在多数情况下可以满足需求.
key的失效时间，被称作“锁定有效期”。它不仅是key自动失效时间，而且还是一个客户端持有锁多长时间后可以被另外一个客户端重新获得。
截至到目前，我们已经有较好的方法获取锁和释放锁。基于Redis单实例，假设这个单实例总是可用，这种方法已经足够安全。现在让我们扩展一下，假设Redis没有总是可用的保障。

Redlock算法

在Redis的分布式环境中，我们假设有N个Redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。之前我们已经描述了在Redis单实例下怎么安全地获取和释放锁。我们确保将在每（N)个实例上使用此方法获取和释放锁。在这个样例中，我们假设有5个Redis master节点，这是一个比较合理的设置，所以我们需要在5台机器上面或者5台虚拟机上面运行这些实例，这样保证他们不会同时都宕掉。
为了取到锁，客户端应该执行以下操作:

1. 获取当前Unix时间，以毫秒为单位。
2. 依次尝试从N个实例，使用相同的key和随机值获取锁。在步骤2，当向Redis设置锁时,客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试另外一个Redis实例。
3. 客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
4. 如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。
5. 如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功）。

这个算法是异步的么?

算法基于这样一个假设：虽然多个进程之间没有时钟同步，但每个进程都以相同的时钟频率前进，时间差相对于失效时间来说几乎可以忽略不计。这种假设和我们的真实世界非常接近：每个计算机都有一个本地时钟，我们可以容忍多个计算机之间有较小的时钟漂移。
从这点来说，我们必须再次强调我们的互相排斥规则：只有在锁的有效时间（在步骤3计算的结果）范围内客户端能够做完它的工作，锁的安全性才能得到保证（锁的实际有效时间通常要比设置的短，因为计算机之间有时钟漂移的现象）。.
想要了解更多关于需要时钟漂移间隙的相似系统, 这里有一个非常有趣的参考: Leases: an efficient fault-tolerant mechanism for distributed file cache consistency.

失败时重试

当客户端无法取到锁时，应该在一个随机延迟后重试,防止多个客户端在同时抢夺同一资源的锁（这样会导致脑裂，没有人会取到锁）。同样，客户端取得大部分Redis实例锁所花费的时间越短，脑裂出现的概率就会越低（必要的重试），所以，理想情况一下，客户端应该同时（并发地）向所有Redis发送SET命令。
需要强调，当客户端从大多数Redis实例获取锁失败时，应该尽快地释放（部分）已经成功取到的锁，这样其他的客户端就不必非得等到锁过完“有效时间”才能取到（然而，如果已经存在网络分裂，客户端已经无法和Redis实例通信，此时就只能等待key的自动释放了，等于被惩罚了）。

释放锁

释放锁比较简单，向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁.

安全争议

这个算法安全么？我们可以从不同的场景讨论一下。
让我们假设客户端从大多数Redis实例取到了锁。所有的实例都包含同样的key，并且key的有效时间也一样。然而，key肯定是在不同的时间被设置上的，所以key的失效时间也不是精确的相同。我们假设第一个设置的key时间是T1(开始向第一个server发送命令前时间），最后一个设置的key时间是T2(得到最后一台server的答复后的时间），我们可以确认，第一个server的key至少会存活 MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT。所有其他的key的存活时间，都会比这个key时间晚，所以可以肯定，所有key的失效时间至少是MIN_VALIDITY。
当大部分实例的key被设置后，其他的客户端将不能再取到锁，因为至少N/2+1个实例已经存在key。所以，如果一个锁被（客户端）获取后，客户端自己也不能再次申请到锁(违反互相排斥属性）。
然而我们也想确保，当多个客户端同时抢夺一个锁时不能两个都成功。
如果客户端在获取到大多数redis实例锁，使用的时间接近或者已经大于失效时间，客户端将认为锁是失效的锁，并且将释放掉已经获取到的锁，所以我们只需要在有效时间范围内获取到大部分锁这种情况。在上面已经讨论过有争议的地方，在MIN_VALIDITY时间内，将没有客户端再次取得锁。所以只有一种情况，多个客户端会在相同时间取得N/2+1实例的锁，那就是取得锁的时间大于失效时间（TTL time)，这样取到的锁也是无效的.
如果你能提供关于现有的类似算法的一个正式证明（指出正确性），或者是发现这个算法的bug？ 我们将非常感激.

活性争议

系统的活性安全基于三个主要特性:

1. 锁的自动释放（因为key失效了）：最终锁可以再次被使用.
2. 客户端通常会将没有获取到的锁删除，或者锁被取到后，使用完后，客户端会主动（提前）释放锁，而不是等到锁失效另外的客户端才能取到锁。.
3. 当客户端重试获取锁时，需要等待一段时间，这个时间必须大于从大多数Redis实例成功获取锁使用的时间，以最大限度地避免脑裂。.

然而，当网络出现问题时系统在失效时间(TTL)内就无法服务，这种情况下我们的程序就会为此付出代价。如果网络持续的有问题，可能就会出现死循环了。 这种情况发生在当客户端刚取到一个锁还没有来得及释放锁就被网络隔离.
如果网络一直没有恢复，这个算法会导致系统不可用.

性能，崩溃恢复和Redis同步

很多用户把Redis当做分布式锁服务器，使用获取锁和释放锁的响应时间，每秒钟可用执行多少次 acquire / release 操作作为性能指标。为了达到这一要求，增加Redis实例当然可用降低响应延迟（没有钱买硬件的”穷人”,也可以在网络方面做优化，使用非阻塞模型，一次发送所有的命令，然后异步的读取响应结果，假设客户端和redis服务器之间的RTT都差不多。
然而，如果我们想使用可以从备份中恢复的redis模式，有另外一种持久化情况你需要考虑，.
我们考虑这样一种场景，假设我们的redis没用使用备份。一个客户端获取到了3个实例的锁。此时，其中一个已经被客户端取到锁的redis实例被重启，在这个时间点，就可能出现3个节点没有设置锁，此时如果有另外一个客户端来设置锁，锁就可能被再次获取到，这样锁的互相排斥的特性就被破坏掉了。
如果我们启用了AOF持久化，情况会好很多。我们可用使用SHUTDOWN命令关闭然后再次重启。因为Redis到期是语义上实现的，所以当服务器关闭时，实际上还是经过了时间，所有（保持锁）需要的条件都没有受到影响. 没有受到影响的前提是redis优雅的关闭。停电了怎么办？如果redis是每秒执行一次fsync，那么很有可能在redis重启之后，key已经丢弃。理论上，如果我们想在Redis重启地任何情况下都保证锁的安全，我们必须开启fsync=always的配置。这反过来将完全破坏与传统上用于以安全的方式实现分布式锁的同一级别的CP系统的性能.
然而情况总比一开始想象的好一些。当一个redis节点重启后，只要它不参与到任意当前活动的锁，没有被当做“当前存活”节点被客户端重新获取到,算法的安全性仍然是有保障的。
为了达到这种效果，我们只需要将新的redis实例，在一个TTL时间内，对客户端不可用即可，在这个时间内，所有客户端锁将被失效或者自动释放.
使用延迟重启可以在不采用持久化策略的情况下达到同样的安全，然而这样做有时会让系统转化为彻底不可用。比如大部分的redis实例都崩溃了，系统在TTL时间内任何锁都将无法加锁成功。

使算法更加可靠：锁的扩展

如果你的工作可以拆分为许多小步骤，可以将有效时间设置的小一些，使用锁的一些扩展机制。在工作进行的过程中，当发现锁剩下的有效时间很短时，可以再次向redis的所有实例发送一个Lua脚本，让key的有效时间延长一点（前提还是key存在并且value是之前设置的value)。
客户端扩展TTL时必须像首次取得锁一样在大多数实例上扩展成功才算再次取到锁，并且是在有效时间内再次取到锁（算法和获取锁是非常相似的）。
这样做从技术上将并不会改变算法的正确性，所以扩展锁的过程中仍然需要达到获取到N/2+1个实例这个要求，否则活性特性之一就会失效。

想要得到帮助?

如果你正在做分布式系统，你的意见和分析非常重要。其他语言实现分布式锁的算法同样非常宝贵。
提前感谢各位!

Redlock 分析

1. Martin Kleppmann 在这儿分析了Redlock. 我不赞同他的说法，并且对他做出了回复 我的回复在这儿.

http://redis.cn/topics/distlock.html

LockTake

/// <summary>
/// The lazy connection.
/// </summary>
private Lazy<ConnectionMultiplexer> lazyConnection = new Lazy<ConnectionMultiplexer>(() =>
{
    ConfigurationOptions configuration = new ConfigurationOptions
    {
        AbortOnConnectFail = false,
        ConnectTimeout = 5000,
        Password= "123456",
        DefaultDatabase=1
    };

    configuration.EndPoints.Add("127.0.0.1", 6379);

    return ConnectionMultiplexer.Connect(configuration.ToString());
});

/// <summary>
/// Gets the connection.
/// </summary>
/// <value>The connection.</value>
public ConnectionMultiplexer Connection => lazyConnection.Value;

/// <summary>
/// Locks the take test.
/// </summary>
[TestCase]
public void LockTakeTest()
{
    var db = Connection.GetDatabase();
    string lockTakeKey = "LockTakeTest";
    Parallel.For(0, 10, (i) =>
    {
        string lockTakeValue = "LockTakeValue";
        //string lockTakeValue = $"LockTakeTest_{i}";
        var lockTakeResult = db.LockTake(lockTakeKey, lockTakeValue, TimeSpan.FromSeconds(10));
        //System.Threading.Thread.Sleep(1000);
        var lockReleaseResult = db.LockRelease(lockTakeKey, lockTakeValue);
        Console.WriteLine($"LockTakeTest_{i},lockTakeResult:{lockTakeResult},lockReleaseResult:{lockReleaseResult}");
    });
}

StackExchange.Redis\src\StackExchange.Redis\RedisDatabase.cs

private Message GetStringSetMessage(RedisKey key, RedisValue value, TimeSpan? expiry = null, When when = When.Always, CommandFlags flags = CommandFlags.None)
{
    WhenAlwaysOrExistsOrNotExists(when);
    if (value.IsNull) return Message.Create(Database, flags, RedisCommand.DEL, key);

    if (expiry == null || expiry.Value == TimeSpan.MaxValue)
    { // no expiry
        switch (when)
        {
            case When.Always: return Message.Create(Database, flags, RedisCommand.SET, key, value);
            case When.NotExists: return Message.Create(Database, flags, RedisCommand.SETNX, key, value);
            case When.Exists: return Message.Create(Database, flags, RedisCommand.SET, key, value, RedisLiterals.XX);
        }
    }
    long milliseconds = expiry.Value.Ticks / TimeSpan.TicksPerMillisecond;

    if ((milliseconds % 1000) == 0)
    {
        // a nice round number of seconds
        long seconds = milliseconds / 1000;
        switch (when)
        {
            case When.Always: return Message.Create(Database, flags, RedisCommand.SETEX, key, seconds, value);
            case When.Exists: return Message.Create(Database, flags, RedisCommand.SET, key, value, RedisLiterals.EX, seconds, RedisLiterals.XX);
            case When.NotExists: return Message.Create(Database, flags, RedisCommand.SET, key, value, RedisLiterals.EX, seconds, RedisLiterals.NX);
        }
    }

    switch (when)
    {
        case When.Always: return Message.Create(Database, flags, RedisCommand.PSETEX, key, milliseconds, value);
        case When.Exists: return Message.Create(Database, flags, RedisCommand.SET, key, value, RedisLiterals.PX, milliseconds, RedisLiterals.XX);
        case When.NotExists: return Message.Create(Database, flags, RedisCommand.SET, key, value, RedisLiterals.PX, milliseconds, RedisLiterals.NX);
    }
    throw new NotSupportedException();
}

LockRelease

StackExchange.Redis\src\StackExchange.Redis\RedisTransaction.cs

private ITransaction GetLockReleaseTransaction(RedisKey key, RedisValue value)
{
    var tran = CreateTransactionIfAvailable(asyncState);
    if (tran != null)
    {
        tran.AddCondition(Condition.StringEqual(key, value));
        tran.KeyDeleteAsync(key, CommandFlags.FireAndForget);
    }
    return tran;
}

public ConditionResult AddCondition(Condition condition)
{
    if (condition == null) throw new ArgumentNullException(nameof(condition));

    var commandMap = multiplexer.CommandMap;
    lock (SyncLock)
    {
        if (_conditions == null)
        {
            // we don't demand these unless the user is requesting conditions, but we need both...
            commandMap.AssertAvailable(RedisCommand.WATCH);
            commandMap.AssertAvailable(RedisCommand.UNWATCH);
            _conditions = new List<ConditionResult>();
        }
        condition.CheckCommands(commandMap);
        var result = new ConditionResult(condition);
        _conditions.Add(result);
        return result;
    }
}

/// <summary>
/// Releases the lock.
/// </summary>
/// <param name="key">Key.</param>
/// <param name="value">Value.</param>
/// <returns><c>true</c>, if lock was released, <c>false</c> otherwise.</returns>
private bool ReleaseLock(string key, string value)
{
    string lua_script = @"  
            if (redis.call('GET', KEYS[1]) == ARGV[1]) then  
            redis.call('DEL', KEYS[1])  
            return true  
            else  
            return false  
            end  
            ";

    try
    {
        var res = Connection.GetDatabase().ScriptEvaluate(lua_script,
                                                   new RedisKey[] { key },
                                                   new RedisValue[] { value });
        return (bool)res;
    }
    catch (Exception ex)
    {
        Console.WriteLine($"ReleaseLock lock fail...{ex.Message}");
        return false;
    }
}

https://redis.io/commands/setnx