1.概述

1.1 背景

分布式锁在很多场景中是非常有用的原语， 不同的进程必须以独占资源的方式实现资源共享就是一个典型的例子。

由于外围的实现存在着各种各样的问题, Redis 作者提出了一种 RedLock算法来约定分布式锁需要注意的事项。

当前java版的实现是 Redisson 框架。

1.2 Redis分布式锁的基本原则

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

1.3 单点问题 & Master-Slave问题

#基本实现
#加锁
实现Redis分布式锁的最简单的方法就是在Redis中创建一个key，
这个key有一个失效时间（TTL)，以保证锁最终会被自动释放掉（这个对应特性2）。
#解锁
当客户端释放资源(解锁）的时候，会删除掉这个key。

#单点问题 & Master-Slave问题
从表面上看，似乎效果还不错，但是这里有一个问题：
这个架构中存在一个严重的单点失败问题。如果Redis挂了怎么办？
你可能会说，可以通过增加一个slave节点解决这个问题。
但这通常是行不通的。这样做，我们不能实现资源的独享,因为Redis的主从同步通常是异步的。

#Master-Slave问题
在这种场景（主从结构）中存在明显的竞态:
>> 客户端A从master获取到锁
>> 在master将锁同步到slave之前，master宕掉了。
>> slave节点被晋级为master节点
>> 客户端B取得了同一个资源被客户端A已经获取到的另外一个锁。安全失效！

1.4 Redis单机版(version > 2.6)的正确实现方法

1.4.1 加锁

SET resource_name my_random_value NX PX 30000

这是一个原子命令(redis客户端已支持)。
需注意key对应的value是“my_random_value”(一个随机值），这个值在所有的客户端必须是唯一的。

1.4.2 解锁

value的值必须是随机数主要是为了更安全的释放锁，释放锁的时候使用脚本告诉Redis:
只有key存在并且存储的值和我指定的值一样才能告诉我删除成功。

#为保证两个操作的原子性, 这里需要使用 lua 脚本实现。
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

#解锁时, 校验 value 是否一致的原因
假设客户端A取得资源锁，但是紧接着被一个其他操作阻塞了，当客户端A运行完毕其他操作后要释放锁时，
原来的锁早已超时并且被Redis自动释放，并且在这期间资源锁又被客户端B再次获取到。
如果仅使用DEL命令将key删除，那么这种情况就会把客户端B的锁给删除掉。
使用Lua脚本就不会存在这种情况，因为脚本仅会删除value等于客户端A的value的key（value相当于客户端的一个签名）。

1.5 Redis 官网关于锁的探讨

1.5.1 加锁

1.5.1.1 Redlock算法

假设有5个Redis master(防止单点故障)。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。
在每个实例上使用与在Redis单实例下获取和释放锁获取和释放锁的方法。

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

1.5.1.2 系统时钟的影响 & 自动续约机制

#算法基于这样一个假设
虽然多个进程之间没有时钟同步，但每个进程都以相同的时钟频率前进，
时间差相对于失效时间来说几乎可以忽略不计。
每个计算机都有一个本地时钟，我们可以容忍多个计算机之间有较小的时钟漂移。

#注意点 (时钟漂移 & 闰秒现象 ---> 正确配置NTP):
只有在锁的有效时间（在步骤3计算的结果）范围内客户端能够做完它的工作，
锁的安全性才能得到保证（锁的实际有效时间通常要比设置的短，因为计算机之间有时钟漂移的现象）。

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

1.5.1.3 失败重试(注意脑裂现象)

当客户端无法取到锁时，应该在一个随机延迟后重试,
防止多个客户端在同时抢夺同一资源的锁（这样会导致脑裂，没有人会取到锁）。
同样，客户端取得大部分Redis实例锁所花费的时间越短，脑裂出现的概率就会越低（必要的重试），
所以，理想情况一下，客户端应该同时（并发地）向所有Redis发送SET命令。

需要强调，当客户端从大多数Redis实例获取锁失败时，应该尽快地释放（部分）已经成功取到的锁，
这样其他的客户端就不必非得等到锁过完“有效时间”才能取到。
然而，如果已经存在网络分裂，客户端已经无法和Redis实例通信，
此时就只能等待key的自动释放了，等于被惩罚了。

1.5.2 释放锁

#这个释放锁指的是已当前获取到锁的客户端向所有实例发送解锁命令
释放锁比较简单，向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁.

1.5.3 一些问题

#redis没设置 slave 节点
假设我们的redis没用使用备份。一个客户端获取到了3个实例的锁。
此时，其中一个已经被客户端取到锁的redis实例被重启，
在这个时间点，就可能出现3个节点没有设置锁，此时如果有另外一个客户端来设置锁，
锁就可能被再次获取到，这样锁的互相排斥的特性就被破坏掉了。

#如果我们启用了AOF持久化，情况会好很多。
我们可用使用SHUTDOWN命令关闭然后再次重启。
因为Redis到期是语义上实现的，所以当服务器关闭时，实际上还是经过了时间，
所有（保持锁）需要的条件都没有受到影响. 没有受到影响的前提是redis优雅的关闭。
停电了怎么办？
如果redis是每秒执行一次fsync，那么很有可能在redis重启之后，key已经丢弃。
理论上，如果我们想在Redis重启地任何情况下都保证锁的安全，我们必须开启fsync=always的配置。
这反过来将完全破坏与传统上用于以安全的方式实现分布式锁的同一级别的CP系统的性能.

然而情况总比一开始想象的好一些。
当一个redis节点重启后，只要它不参与到任意当前活动的锁，
没有被当做“当前存活”节点被客户端重新获取到,算法的安全性仍然是有保障的。

为了达到这种效果，我们只需要将新的redis实例，在一个TTL时间内，
对客户端不可用即可，在这个时间内，所有客户端锁将被失效或者自动释放.

使用"延迟重启"可以在不采用持久化策略的情况下达到同样的安全，
然而这样做有时会让系统转化为彻底不可用。
比如大部分的redis实例都崩溃了，系统在TTL时间内任何锁都将无法加锁成功。

Martin Kleppmann 与 antirez 关于 RedLock 算法的互怼
http://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
http://antirez.com/news/101

1.6 关于redis分布式锁的结论

#使用建议
1.分布式锁的redis采用单机部署，分布式锁专用
2.根据RedLock算法思想，意思是不能只在一个redis实例上创建锁，应该是在多个redis实例上创建锁，**n / 2 + 1**，
必须在大多数redis节点上都成功创建锁，才能算这个整体的RedLock加锁成功，避免说仅仅在一个redis实例上加锁而带来的问题。
要求: 搭建几台独立的redis机器, 互相之间不通信, 不构成主从/哨兵/集群关系.
3.如果对锁比较关注，一致性要求比较高，可以使用ZK实现的分布式锁

#其他方案
1.如果考虑各种网络、宕机等原因，很多问题需要考虑，问题会变的复杂，
其实分布式锁的应用场景不多，很多情况可以绕开分布式锁，使用其他方式解决，比如 队列，异步，响应式
2.分布式锁的场景，更多的应用是一个操作不能同时多处进行，不能短时间内重复执行，需要幂等操作等场景，
比如：防止快速的重复提交，mq与定时任务双线更改状态，防止消息重复消费 等等。
这些情况一般使用setNx即可解决。
3.减库存其实也用不到分布式锁, 可用redis+lua实现。

2.分布式锁的开源实现框架-Redisson

2.1 概述

redisson 是 redis 官方的分布式锁组件。

#Redisson的一些特点
1.redisson所有指令都通过lua脚本执行，redis支持lua脚本原子性执行
2.redisson设置一个key的默认过期时间为30s,如果某个客户端持有一个锁超过了30s怎么办？
>> redisson中有一个watchdog的概念，翻译过来就是看门狗，
它会在你获取锁之后，每隔10秒帮你把key的超时时间设为30s
>> 这样的话，就算一直持有锁也不会出现key过期了，其他线程获取到锁的问题了。
3.redisson的“看门狗”逻辑保证了没有死锁发生。
如果机器宕机了，看门狗也就没了。
此时就不会延长key的过期时间，到了30s之后就会自动过期了，其他线程可以获取到锁。

#ps
lua 脚本的执行是原子性的，再加上 Redis 执行命令是单线程的，
所以在 lua 脚本执行完之前，其他的命令都得等着。

Redisson中的watchdog.png

https://www.cnblogs.com/thisiswhy/p/12596069.html (这里有Redisson 实现分布式锁的分析, 挺好的, 本文不再分析)

2.2 基于lua脚本的无锁化 or 基于 Redisson 的分布式锁控制并发

package com.zy.redis5.single;

import org.assertj.core.util.Lists;
import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import org.redisson.client.codec.StringCodec;
import org.redisson.config.Config;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 此处 demo 以 扣减库存为例, 给出了两种分布式解决方案
 * 方案1:
 *  先将商品及库存数全量加载到 redis 中, 然后借助 lua 脚本实现原子性的扣减库存, 注意这里的原子性是从 redis 中扣减库存
 * 方案2:
 *  借助 redisson 的分布式锁框架, 获取全局资源操作权限, 然后操作 DB 库存, 由于首先于 DB 的 qps, 所以并发效果并不会很好
 *  Redis当做分布式锁服务器时，可使用获取锁和释放锁的响应时间，每秒钟可用执行多少次 acquire / release 操作作为性能指标。
 *  
 * 说明:
 *  可以自行写一个 controller, 启动一个项目, 借助 jmeter 等工具, 验证下并发情况
 */
public class RedisSingleAtomicLuaOrDistributedLock {

    private static RedissonClient client;
    private static Codec codec;
    private static final String KEY = "apple";
    private static final String LOCK_KEY = "lockKey";
    private static List<Object> keyList = Lists.newArrayList();
    private int count = 20;

    static {
        Config config = new Config();
        config.useSingleServer()
                .setDatabase(10)
                .setAddress("redis://192.168.0.156:6379");

        client = Redisson.create(config);
        // FIXME 这里定义了 StringCodec 类型的编解码器, 是因为其默认的编解码器是: MarshallingCodec
        // FIXME 而当使用 lua 脚本时, 要调用 lua 的 tonumber 函数 将库存(string类型) 转为 number 类型时,
        // FIXME 如果用默认的编解码器, 将会得到 nil 的结果, 会出错.
        // FIXME 故这里使用了 StringCodec 来解决, 也可以用 IntegerCodec 或 LongCodec.
        codec = StringCodec.INSTANCE;
        keyList.add(KEY);
    }

    /**************************** 方案1: 将数据全量加载至 redis 中, 在 redis 中扣减库存, 借助 lua 脚本控制并发 *******************************/
    @Test
    public void step01() {
        String luaScript = "return redis.call('set',KEYS[1],ARGV[1]);";
        Object result = client.getScript(codec).eval(RScript.Mode.READ_WRITE, luaScript, RScript.ReturnType.VALUE, keyList, 999);

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
        System.out.println(result);
        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
    }

    @Test
    public void step02() {
        String luaScript = "return redis.call('get', KEYS[1]);";
        Object result = client.getScript(codec).eval(RScript.Mode.READ_ONLY, luaScript, RScript.ReturnType.VALUE, keyList);

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
        System.out.println(result);
        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
    }

    @Test
    public void step03() {
        String luaScript =
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "return 0; " +
                        "end;" +
                        "local count = redis.call('get', KEYS[1]); " +
                        "local decrementCount = ARGV[1]; " +
                        "local a = tonumber(count); " +
                        "local b = tonumber(decrementCount); " +
                        "if (a < b) then " +
                        "return 0; " +
                        "end; " +
                        "redis.call('set', KEYS[1], (a - b)); " +
                        "return 1; ";
        Object result = client.getScript(codec).eval(RScript.Mode.READ_ONLY, luaScript, RScript.ReturnType.VALUE, keyList, 3);

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
        System.out.println(result);
        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>..");
    }

    /**************************** 方案2: 借助 redis 分布式锁 脚本控制并发 *******************************/
    @Test
    public void fn04() throws InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();
        int tobeDecreasedCount = 3;

        for (int i = 0; i < 10; i++) {
            executor.submit(() -> {
                RLock lock = client.getLock(LOCK_KEY);
                boolean b = lock.tryLock();
                if (b) {
                    try {
                        int count = getCount();
                        if (count > tobeDecreasedCount) {
                            decreaseCount(count, tobeDecreasedCount);
                        }
                    } finally {
                        lock.unlock();
                    }
                }
            });
        }

        TimeUnit.SECONDS.sleep(10L);
        System.out.println("剩余库存量是: " + getCount());
    }

    private int getCount() {
        return count;
    }

    private void decreaseCount(int count, int no) {
        this.count = count - no;
    }
}

参考资料
http://redis.cn/topics/distlock.html
https://redis.io/topics/distlock

分布式锁系列(2) 基于Redis & lua无锁化

分布式锁系列(2) 基于Redis & lua无锁化

1.概述

1.1 背景

1.2 Redis分布式锁的基本原则

1.3 单点问题 & Master-Slave问题

1.4 Redis单机版(version > 2.6)的正确实现方法

1.4.1 加锁

1.4.2 解锁

1.5 Redis 官网关于锁的探讨

1.5.1 加锁

1.5.1.1 Redlock算法

1.5.1.2 系统时钟的影响 & 自动续约机制

1.5.1.3 失败重试(注意脑裂现象)

1.5.2 释放锁

1.5.3 一些问题

1.6 关于redis分布式锁的结论

2.分布式锁的开源实现框架-Redisson

2.1 概述

2.2 基于lua脚本的无锁化 or 基于 Redisson 的分布式锁控制并发