三种方式实现分布式锁_如何实现分布式锁

单机锁

当Java应用程序被部署到一台服务器时，我们只需要使用单机锁就可以保证一段代码在同一时刻只被一个线程访问。我们常用的单机锁如Synchronized、可重入锁等都能保证在同一个JVM进程内的多个线程同步访问一段程序来保证数据一致性。

Synchronized等锁能否在分布式环境中保证不同节点的线程同步？答案是不能，以Synchronized关键字为例，Synchronized关键字无论是在偏向锁、轻量级锁还是重量级锁状态都不能实现这点，如重量级锁，重量级锁是靠系统底层的互斥量Mutex实现的，也就是说每个节点(服务器)所使用的互斥量是分开的，节点A的互斥量是无法锁住节点B的线程访问临界区，因此Synchronized关键字只能保证单服务器内的JVM进程的不同线程同步，是不能用做分布式环境中来保证线程同步。

分布式锁

在分布式的集群环境中，同样的Java程序会被部署到多台机器中用来处理同样的请求，此时有多个不同JVM进程的线程在处理请求。比如高并发抢购某个商品，该应用程序部署在分布式环境中，若该商品总量为10000，此时若有5个用户发起请求进行购买，负载均衡服务器进行请求处理，将请求转发到5个不同服务器的应用程序中进行处理，若此时无分布式锁，5台不同的机器中的JVM进程的线程同时调用同样的代码块，读取商品总量10000，然后同时将商品减1，然后将商品余量更新回数据库，商品本应该只剩9995，但因为没有使用分布式锁，使得商品数量出错，剩余量变成9999。单机锁只能约束同一JVM进程内的线程使其同步，而分布式锁却能约束系统内所有节点的不同JVM进程，使得不同进程内的线程同步。

分布式锁的作用

分布式锁可以保证数据在分布式环境中的正确性，不但如此，分布式锁的另外一个作用是可以避免了不同节点重复相同的工作。因此分布式的作用主要是以下两点:
(1)避免不同节点重复相同的工作：比如用户执行了某个操作有可能不同节点会发送多封邮件；
(2)保证数据的正确性：如果两个节点在同一条数据上同时进行操作，可能会造成数据错误或不一致的情况出现

分布式锁应有的特性

(1)锁具有排他性，只能被一台主机中的一个线程获取。
(2)具有高可靠性、高可用性，有一定的容错能力且对锁失效有应对机制。
(3)锁是可重入的，不能存在自己把自己锁死的情况。
(4)加锁解锁应该具有高性能，且加锁和解锁的客户应该是同一个;且具有安全性，自己的锁不能被别人的钥匙捅开。
(5)不存在死锁，不存在多个线程相互锁死的情况。

分布式锁的实现方式

(1)基于数据库的实现方式
(2)基于缓存的实现方式(利用Redis、Memcached的原子性操作)
(3)基于Zookeeper的实现方式

基于数据库的实现方式

基于数据库的实现方式有基于数据库的表以及基于数据库的排他锁两种。

这部分会涉及一些当前读、行级锁、表级锁、排他锁的概念，若对这些内容不是很了解可以翻阅《解析数据库锁协议和InnoDB锁机制(全面解析行级锁、表级锁、排他锁、共享锁、悲观锁、乐观锁等常用锁)》和《MVCC实现原理》这两篇博客。

基于数据库的表

创建一个专门用做分布式锁的表，且该表上有一个method_name字段，在method_name字段上创建唯一索引，以避免方法名相同的记录出现。当多个节点的线程同时想要获得同一个共享资源或者执行同一个方法时，谁先成功往分布式锁表中插入方法名字段为该资源名或者方法名的记录谁就获得分布式锁，就可以获得该共享资源或执行该方法。释放分布式锁的方法就是在获得资源或者执行该方法之后在分布式锁表中删除该行记录。

CREATE TABLE `lock_table` (
  `id` int(11) NOT NULL,
  `method_name` varchar(100) COLLATE utf8_unicode_ci NOT NULL COMMENT '唯一',
  `insert_time` datetime DEFAULT NULL COMMENT '插入时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `method_unique` (`method_name`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;
1
2
3
4
5
6
7

这种方式不被推荐采用，因为存在以下缺点:
(1)存在单机问题，若分布式锁表存在的数据库崩溃则无法插入记录，所有需要用到锁的操作都无法进行。虽然可以通过双机部署、数据部署、主备切换来解决，但数据库的可用性和性能将直接影响分布式锁的可用性及性能。
(2)锁不具备可重入性，但可通过在分布式锁表中加入机器主机信息和线程ID字段的方式来赋予可重入性，线程再次需要获得锁时，进行线程ID和机器主机信息查询匹配，若相同则赋予锁。
(3)没有失效时间，若某个线程获得锁之后没有释放锁就挂了，然后该记录一直存在数据库中，则会导致系统内其它线程无法获得该分布式。虽然可以通过定期清除数据库内的记录来解决这个问题，但损耗性能。
(4)该锁是不阻塞的，所以需要通过不停的进行循环插入来获得锁，会消耗CPU资源。

基于数据库表的排他锁

同样是创建一个专门用做分布式锁的表，且该表上有一个method_name字段，在method_name字段上创建唯一索引，以避免出现方法名相同的记录。不同的是线程获得锁的方式是进行当前读，通过使用当前读(for update)的方式读取分布式锁表上的记录来获取锁。通过提交事务来释放排他锁。需要注意的是MySQL默认使用的存储引擎是支持行级排他锁，但前提是通过索引来进行查询时才会使用行级排他锁，也就是说如下的查询命令where后的查询条件的method_name字段上面一定得建立索引，因为之前在该字段上建立了唯一索引，所以这个条件满足。

select method_name from lock_table where method_name='test1' for update;
1

这种方法解决了锁没有失效时间问题(机器宕机后，数据库会自动释放锁)和不是阻塞的问题，但同样存在单机问题、不可重入问题，而且当分布式锁表的记录少时，MySQL的优化器会判定不走索引而是全表扫描会效率更好，因为行级锁是建立在索引之上，因此此时使用的就不再是行级排他锁而是表级排他锁，整个分布式系统都竞争同一个表的一个锁，可想而知这样的后果是啥。

小结

不难发现用数据库来实现分布式锁，虽然一定程度上能满足分布式锁应有的特性，但锁依赖于数据库导致了其性能低、不可靠性高且实现起来复杂，因为需要双机部署等以防止出错，因此这里不推荐使用这种方案。

基于缓存的实现方式

基于缓存的实现方式是利用了非关系型数据库的原子性操作来实现分布式锁。比如比较常用的Redis和Memcached，利用Memcached的add()方法的只有当key不存在时才能加锁的特性以及该方法操作的原子性来进行加锁或者是Redis中的setnx()方法的只有当key不存在时才能加锁的特性以及该方法操作的原子性来进行加锁。

这里主要详细分析Redis实现分布式锁的原理和方法。这里先分析下Redis实现分布式的几个要点:
(1)加锁:Redis实现分布式锁可以通过Redis的SETNX [key ，value]命令实现，当Redis中key值不存在时才可以成功赋值，保证了只有一个线程能够赋值，因此只有一个线程能获得锁；(看似可行其实不可行，详情下解)
(2)解锁:线程得到锁之后完成相应的操作之后需要释放锁，可以通过Redis的DEL命令去删除该key和value，且线程A得到锁应该只有线程A释放，而不能被其它线程释放，因此加锁时设置的value不应该是特定值值而应该是一个唯一的随机数，所有同一个key的竞争者的value值都不能一样。在释放锁之前先进行value值判断，只有自己预期的值与Redis中存储的value值相同才能释放锁，但是判断和del操作是两个操作不具有原子性，因此会出现问题，解决方案下面详解。
(3）锁超时问题:若某个线程获得锁之后该线程挂掉，此时锁未释放，其它线程都无法获得锁，进入阻塞状态。因此在获取锁时需要进行过期时间设置，但SETNX命令不支持过期时间设置，因此需要使用EXPIRE命令进行过期时间设置。但SETNX操作和EXPIRE操作是两个分开的操作，不具有原子性，因此会出现问题，解决方法下面详述。
(4)若业务过长，获得锁的线程还没来得及访问完临界区，锁时间就到期，其它线程获得锁之后也进入临界区，导致多线程访问临界区，出现不可预料的错误，这个问题该如何解决?
(5）Redis若是部署在单机上面，若该机器宕机则所有的锁无法获取，导致无法进行相应的操作。即使使用了主从模式，有备用节点，但若master节点中的锁备份到salve节点之前master节点就宕机，就可能会出现多个用户同时获得锁。因此单机这个问题也需要解决。

加锁以及锁失效时间设置

从上文可知分布式锁通过Redis的SETNX指令来实现加锁，当Redis中不存在某个key变量时，即表明当前没有任何线程获得该锁，某线程执行SETNX指令，往Redis中插入一个key-value对，此时该线程获得锁。其它线程来执行SETNX来插入相同的key值，操作无效，无法获得锁，因此也无法执行方法或者获得资源。而且SETNX这个命令加锁过程具有原子性，无法被其它线程打断，保证了一旦第一个执行则一定获得锁，且锁具有排他性，有且仅有一个线程获得锁。

SETNX [key ，value] （SET if Not eXists）
当且仅当 key 不存在时,将 key 的值设为 value 。若给定的 key 已经存在，则SETNX不做任何动作。
返回值：设置成功，返回 1 。设置失败，返回 0 。

若只加锁，不设置锁过期时间，当某个获得锁的线程在释放锁之前就挂掉了，则该key值会一直存在Redis中，锁永远无法得到释放，其它线程对该key进行SETNX操作都无效，因此其它需要该锁开宝箱(执行方法或获取资源)的线程也只能苦苦等，因此需要设置过期时间。但SETNX指令并没有提供过期时间的功能，因此可以通过EXPIRE指令来对key变量进行过期时间设定。

EXPIRE [key ，t]
命令用于设置 key 的过期时间。key 过期后将不再可用。
返回值:设置成功返回 1 。 当 key 不存在或者不能为 key 设置过期时间时返回 0 。

新的问题就来了，SETNX指令和EXPIRE指令单独执行是具有原子性，但把他们放在一起是不具有原子性的，若某个线程在成功执行SETNX指令获得锁之后，还未来得及执行EXPIRE指令就挂掉了，该锁还是不能被其它线程获得，因此我们需要使得这两个指令一起执行且具有原子性。Redis 2.6.12以上版本为SET指令增加了NX可选参数，使得SET这个指令既可以设置过期时间又可以设置为只有key不存在时，才能进行key设置，且这个指令具有原子性。

SET key value [EX seconds] [PX milliseconds] [NX|XX]
(1) EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
(2)PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecondvalue 。
(3)NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
(4)XX ：只在键已经存在时，才对键进行设置操作。
(5)返回值: SET 在设置操作成功完成时，才返回 OK ;设置了 NX 或者 XX ，但因为条件没达到而造成设置操作未执行，那么命令返回空批量回复（NULL Bulk Reply）

Java中加锁的实现代码:

private static final String LOCK_SUCCESS = "OK";
private static final Long RELEASE_SUCCESS = 1L;
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "PX";

@Override
public String acquire() {
    try {
        // 获取锁的超时时间，超过这个时间则放弃获取锁
        long end = System.currentTimeMillis() + acquireTimeout;
        // 随机生成一个全局唯一的Value，防止自己的锁被别人打开
        String requireToken = UUID.randomUUID().toString();
        while (System.currentTimeMillis() < end) {
            String result = jedis
                .set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
            if (LOCK_SUCCESS.equals(result)) {
                return requireToken;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    } catch (Exception e) {
        log.error("acquire lock due to error", e);
    }

    return null;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

续命(非必须)

锁设置了过期时间带来的一个负面影响就是当锁过期之后，锁就会被释放，而若此时临界区还没访问完，则需要继续访问下去，但此时会有其它线程获得锁而进入临界区，于是形成了多个线程并行访问同一临界区。这就相当于还没让男友变前男友，就跟新男友你侬我侬，于是旧男友和新男友铁定扭打在一起，发生不可追悔的错误。

因此我们需要给该渣女一点时间来分手，获得锁的线程在获得锁时开启一个守护线程(与男友的兄弟打成一片)，当线程的锁时间快到期时(渣女感觉不爱了)，让守护线程执行EXPIRE指令给锁再续上一段时间(让男友的兄弟去跟男友说分手，确实是渣女本渣了)，线程在执行完毕之前显示关闭守护线程(拉黑男友及其兄弟)。

并不是一定需要续命，若自己设置的过期时间合适，且所作的业务花费时间不长根本不需要续命，此时续命不划算，因为续命也是要代价的。

解锁

客户端线程在执行完代码之后需要释放锁，也就是说需要把自己存入Redis中的键值对给删除掉，可以通过DEL指令来删除该键值对。前文我们也提到了无论是什么锁，最基本的要求就是自己的锁只能被自己的钥匙打开，不能说线程A加的锁，被线程B给释放掉了。因此我们在加锁时需要给锁添上自己的唯一的密码，key值是所有线程公用的，因此可以在value值上做文章，将value值设置为一个唯一的随机值，用作释放锁时的密码，我们需要在释放锁之前先通过判定自己手里拿着的密码(value值)是否和在Redis里面存储的锁的value值相同，若相同才能释放，不相同就不能释放。

DEL [key]
命令用于删除已存在的键。不存在的 key 会被忽略。
返回值:被删除 key 的数量

因此释放锁的过程是先判断是否是自己的锁，然后再释放锁，但这其实是两个操作，不具有原子性，在高并发情况下可能会出现问题。若线程A准备释放锁时，已经判断是自己的锁了，在准备执行释放锁指令时线程A所在的进程因为GC造成STW，在这时间段内，线程A获得锁的时间已过期，线程B获得锁，线程A获得CPU之后继续执行，释放掉的就是线程B的锁。因此我们需要使用Lua脚本将判断和释放锁的过程一起实现，使得这两个操作变成一个操作且具有原子性。

Java中解锁的实现代码:

@Override
public boolean release(String identify) {
    if (identify == null) {
        return false;
    }
     //lua脚本
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    Object result = new Object();
    try {
        result = jedis.eval(script, Collections.singletonList(lockKey),
            Collections.singletonList(identify));
        if (RELEASE_SUCCESS.equals(result)) {
            log.info("release lock success, requestToken:{}", identify);
            return true;
        }
    } catch (Exception e) {
        log.error("release lock due to error", e);
    } finally {
        if (jedis != null) {
            jedis.close();
        }
    }

    log.info("release lock failed, requestToken:{}, result:{}", identify, result);
    return false;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

单点Redis实现分布式锁示例(代码)

以秒杀库存数量为场景，使用分布式锁完成购买

public interface DistributedLock {
    /**
     * 获取锁
     * @return 锁标识
     */
    String acquire();

    /**
     * 释放锁
     * @param indentifier
     * @return
     */
    boolean release(String indentifier);
}


@Slf4j
public class RedisDistributedLock implements DistributedLock{

    private static final String LOCK_SUCCESS = "OK";
    private static final Long RELEASE_SUCCESS = 1L;
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";

    /**
     * redis 客户端
     */
    private Jedis jedis;

    /**
     * 分布式锁的键值
     */
    private String lockKey;

    /**
     * 锁的超时时间 10s
     */
    int expireTime = 10 * 1000;

    /**
     * 锁等待，防止线程饥饿
     */
    int acquireTimeout  = 1 * 1000;

    /**
     * 获取指定键值的锁
     * @param jedis jedis Redis客户端
     * @param lockKey 锁的键值
     */
    public RedisDistributedLock(Jedis jedis, String lockKey) {
        this.jedis = jedis;
        this.lockKey = lockKey;
    }

    /**
     * 获取指定键值的锁,同时设置获取锁超时时间
     * @param jedis jedis Redis客户端
     * @param lockKey 锁的键值
     * @param acquireTimeout 获取锁超时时间
     */
    public RedisDistributedLock(Jedis jedis,String lockKey, int acquireTimeout) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
    }

    /**
     * 获取指定键值的锁,同时设置获取锁超时时间和锁过期时间
     * @param jedis jedis Redis客户端
     * @param lockKey 锁的键值
     * @param acquireTimeout 获取锁超时时间
     * @param expireTime 锁失效时间
     */
    public RedisDistributedLock(Jedis jedis, String lockKey, int acquireTimeout, int expireTime) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.acquireTimeout = acquireTimeout;
        this.expireTime = expireTime;
    }

    @Override
    public String acquire() {
        try {
            // 获取锁的超时时间，超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout;
            // 随机生成一个value
            String requireToken = UUID.randomUUID().toString();
            while (System.currentTimeMillis() < end) {
                String result = jedis.set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
                if (LOCK_SUCCESS.equals(result)) {
                    return requireToken;
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (Exception e) {
            log.error("acquire lock due to error", e);
        }

        return null;
    }

    @Override
    public boolean release(String identify) {
　　　 if(identify == null){
            return false;
        }

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = new Object();
        try {
            result = jedis.eval(script, Collections.singletonList(lockKey),
                Collections.singletonList(identify));
        if (RELEASE_SUCCESS.equals(result)) {
            log.info("release lock success, requestToken:{}", identify);
            return true;
        }}catch (Exception e){
            log.error("release lock due to error",e);
        }finally {
            if(jedis != null){
                jedis.close();
            }
        }

        log.info("release lock failed, requestToken:{}, result:{}", identify, result);
        return false;
    }
}
　　下面就以秒杀库存数量为场景，测试下上面实现的分布式锁的效果。具体测试代码如下：

public class RedisDistributedLockTest {
    static int n = 500;
    public static void secskill() {
        System.out.println(--n);
    }

    public static void main(String[] args) {
        Runnable runnable = () -> {
            RedisDistributedLock lock = null;
            String unLockIdentify = null;
            try {
                Jedis conn = new Jedis("127.0.0.1",6379);
                lock = new RedisDistributedLock(conn, "test1");
                unLockIdentify = lock.acquire();
                System.out.println(Thread.currentThread().getName() + "正在运行");
                secskill();
            } finally {
                if (lock != null) {
                    lock.release(unLockIdentify);
                }
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

小结

1、使用set指令来实现加锁而不是setnx指令，网上写的那些setnx指令实现加锁的是错误的，因为它没有考虑过期时间的设置，以及两个操作在一起不具有原子性的问题。
2、解锁之前先判断是不是自己的锁，而且判断的操作与解锁这个操作应该一起执行，且应具有原子性，通过Lua脚本实现这一点。
3、给锁续命这个过程不是一定需要，可以通过合适的过期时间的设置和将大业务分割成小业务来替代。
4、上述的所有过程的实现前提其实是Redis只存在一个节点中，因此存在单机故障问题，即使是进行主从备份，有备用节点，但若master节点中的锁备份到salve节点之前master节点就宕机，就可能会出现多个用户同时获得锁，因此单机故障问题不能通过备份解决。若你只追求性能且不担心单机故障，下面这部分Redis官方发布的基于多节点Redis 实现分布式锁的方式-RedLock可以选择跳过。

RedLock

ReadLock算法是Redis官方提出的一种分布式算法，它与上述SET和DEL指令实现的分布式锁的最大区别就在于ReadLock算法用于Redis存在于多个节点中，可以有效的防止单机故障问题。它实现了分布式锁应有的要求:
(1)安全特性：互斥访问，即永远只有一个 client 能拿到锁
(2)避免死锁：最终 客户端都可能拿到锁，不会出现死锁的情况，即使原本锁住某资源的客户端崩溃，其它客户端仍然能获得锁。
(3)容错性：只要大部分 Redis 节点存活就可以正常提供服务。

以客户端通过N个Redis主服务器来实现加锁和解锁的过程来说明RedLock算法的原理

加锁

假设N=5，也就是说分布式系统内分配了5台独立的Redis主服务器，当客户端能向至少(N/2)+1个Redis申请到锁，客户端才可以真正的获得锁，不然就获取锁失败，释放已有的锁。
(1)获取当前时间戳，微秒单位。
(2)客户端尝试使用相同的key,value依次获取所有redis服务的锁，在获取锁的过程中花费的获取时间应该比锁过期时间短很多，这是为了不要过长时间等待已经关闭的redis服务，而应该试着获取下一个redis实例。比如说锁的有效时间(过期时间)是10s,则获取锁最大允许时间(timeout)应该是5~50ms，如果在timeout期限之内无法获取该redis的锁，则放弃该锁，继续获取下一个redis的锁。
(3)最终当客户端在大于等于 3 个redis 上成功申请到锁的时候，且它申请所有锁总共花费的时间小于锁的有效时间时，客户端才真正获得锁。申请锁花费总时间为申请完毕的当下时间减去步骤(1)的时间。此时锁真正剩余的有效时间为锁原始有效时间(ttl)减去申请锁花费的总时间。比如锁原始有效时间(ttl)为5s，申请锁的过程花费3s，此时锁真正的有效时间为2s。
(4)如果 client 申请锁失败了，那么它就会在少部分申请成功锁的redis节点上执行释放锁的操作。

解锁

解锁过程很简单，向所有redis节点发送解锁命令即可，redis会自行判断客户端的value是否与redis中key对应的value相同，相同就成功释放。

失败重试

当client不能获取锁时，应该在随机时间后重试获取锁；并且最好在同一时刻并发的把set命令发送给所有redis实例；而且对于已经获取锁的client在完成任务后要及时释放锁，这是为了节省时间。

崩溃恢复策略

1、如果我们的节点没有持久化机制，client 从 5 个 master 中的 3 个处获得了锁，然后其中一个重启了，整个环境中又出现了 3 个 master 可供另一个 client 申请同一把锁！ 违反了互斥性。
2、如果我们开启了 AOF 持久化那么情况会稍微好转一些，因为 Redis 的过期机制是语义层面实现的，所以在Redis挂了的时候时间依旧在流逝，重启之后锁状态不会受到污染。但是考虑断电之后呢，AOF部分命令没来得及刷回磁盘直接丢失了，除非我们配置刷回策略为 fsnyc = always，但这会损伤性能。
3、解决这个问题的方法是redis同步到磁盘方式保持默认的每秒，当一个节点停掉之后，我们规定它要等待TTL时间后再重启，即延时重启，缺陷就是这这段时间内该节点不能做任何事情。

小结

RedLock算法是Redis官方提出的用于解决Redis做分布式锁时存在的单点故障问题，通过客户端只要取得Redis节点的大部分锁就可以成功获得锁的机制使得若小部分Redis实例所在的机器故障也不会影响分布式锁的使用。当然RedLock也存在一定的缺陷，也就是可靠性不如即将登场的Zookeeper实现的分布式锁。对于RedLock存在的问题， Martin Kleppmann 的文章 How to do distributed locking 进行了详述，感兴趣者可以自行观看。

关于RedLock的Java实现-Redisson的代码可以观看源码

基于Zookeeper的实现方式

下面会涉及Zookeeper基础知识，若是对Zookeeper不是很了解的可以翻看这篇博文----《Zookeeper原理以及要点知识》

Zookeeper实现分布式锁是基于Znode中的临时节点或者临时顺序节点，分别可以用来实现非公平锁和公平锁。原理就是把节点当作锁，创建节点成功(非公平锁)或者多个节点创建成功让节点当前存在且创建时间最小的线程获得锁(公平锁)。临时节点或者临时顺序节点这两种节点都有一个共同特性，一旦创建该节点的客户端断开与Zookeeper的会话，则该节点就会自动删除，也就是说除了自己主动删除节点以释放锁，当线程挂了或者机器宕机时会话会自动断开，临时节点也会被删除，锁也被释放了，因此该锁具有高可靠性。

非公平锁

通过在指定节点(持久节点)下创建同一名称的临时节点，来实现非公平锁，谁先创建成功则谁获得锁，因此锁具有排他性。例如我们在/Lock节点下创建methodLock节点，也就是/Lock/methodLock节点，谁创建成功谁就获得分布式锁，其它未获得锁的线程在该指定节点(例子中的/Lock）注册监听器(Watcher)，当其子节点被删除时会得到通知，此时线程会被唤醒去抢夺锁。通过对节点设置删除权限使得只有创建者才能主动删除以此使得其它线程不能删除自己的锁。可以通过在客户端实现相应的逻辑使得其具有可重入性。以此看来，该非公平锁具有排他性、高可靠性、可重入性、安全性(自创自删)，并且从获得锁的方式来看就知道它不会产生死锁。但这种实现方式具有两个缺点，一就是不公平，二就是过于重量，会出现惊群问题。每次锁释放之后，所有订阅Watcher的线程都会被唤醒，产生不必要的线程调度，产生系统开销。

公平锁

公平锁不会出现惊群问题，因为每次只有一个线程会被通知去获得锁，类似于等待队列。它的实现依赖于Znode中的临时有序节点，临时有序节点具有与临时节点相同的特性即客户端创建的节点会因为断开会话而被删除，不同于临时节点的是它具有根据创建时间而被赋予编号的特性，而这个特性正好用来实现公平锁。

在Zookeeper当中创建一个持久节点Lock。当客户端想要获得锁时，需在Lock这个节点下面创建一个临时顺序节点。然后查找Lock下面所有的临时顺序节点并排序，判断自己所创建的节点是不是编号最小的节点，若是则获得锁，若不是则未能获得锁，并在排序在它前面的那个节点上注册监听器(watcher)，当它前面那个节点删除时会通知该节点，该节点再进行判断自己是不是Lock节点下编号最小的子节点，若是则获得锁，若不是则继续在Lock目录下当前排序在它前面的那个节点上注册监听器。通过以上过程你可以发现锁的获取是按照创建时间来的，谁先来争取锁谁就先获得锁，因此它实现的是公平锁。而且对比于非公平锁实现方式，这种实现方式不会出现惊群现象，因为它们的注册器不是注册在同一个父节点上，而是编号排在它们前面的那个节点，因此每次有锁被释放，只会有一个线程被唤醒。

下面以示例来进行说明(示例来自于Zookeeper实现分布式锁)
首先，在Zookeeper当中创建一个持久节点ParentLock。当第一个客户端想要获得锁时，需要在ParentLock这个节点下面创建一个临时顺序节点 Lock1。

之后，Client1查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock1是不是顺序最靠前的一个。如果是第一个节点，则成功获得锁。

这时候，如果再有一个客户端 Client2 前来获取锁，则在ParentLock下载再创建一个临时顺序节点Lock2。

Client2查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock2是不是顺序最靠前的一个，结果发现节点Lock2并不是最小的。

于是，Client2向排序仅比它靠前的节点Lock1注册Watcher，用于监听Lock1节点是否存在。这意味着Client2抢锁失败，进入了等待状态。

这时候，如果又有一个客户端Client3前来获取锁，则在ParentLock下载再创建一个临时顺序节点Lock3。

Client3查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock3是不是顺序最靠前的一个，结果同样发现节点Lock3并不是最小的。于是，Client3向排序仅比它靠前的节点Lock2注册Watcher，用于监听Lock2节点是否存在。这意味着Client3同样抢锁失败，进入了等待状态。

这样一来，Client1得到了锁，Client2监听了Lock1，Client3监听了Lock2。这恰恰形成了一个等待队列。

若想使用Zookeeper的方式实现分布式锁，可以使用Curator，它封装了Zookeeper的API，且提供了很多常用的功能的实现，包括分布式锁，感兴趣的可以自行搜索。

三种方式比较

从理解的难易程度角度（从低到高）:数据库 < 缓存 < Zookeeper

从实现的复杂性角度（从低到高）：Zookeeper <=缓存 < 数据库

从性能角度（从高到低）:缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）:Zookeeper > 缓存 > 数据库

基于数据库的方式不被推荐，因为实现复杂、性能低、可靠性低,用来理解分布式锁即可。Zookeeper虽然实现方式优雅，但它删除添加节点(锁)的性能低(删除添加节点要在Zookeeper集群内部进行同步)，而且需要维护一个Zookeeper集群，因此很少会把它单独作为锁使用；而Redis虽然也常是集群部署，但它常用来做缓存，使用的比Zookeeper多，而且Redis删除添加锁效率高，不过它的可靠性不如Zookeeper。分布式锁的实现方式可以根据你所需要应用的场景，综合性能、可靠性、实现复杂程度等多个方面来选择。