一般实现分布式锁都有哪些方式？使用 Redis 如何设计分布式锁？使用 zk 来设计分布式锁可以吗？这两种分布式锁的实现方式哪种效率比较高？

1.Redis 分布式锁

（1）Redis 最普通的分布式锁

（2）RedLock 算法

2.zk 分布式锁

3.redis 分布式锁和zk分布式锁的对比

1.Redis 分布式锁

        官方叫做 RedLock 算法，是 Redis 官方支持的分布式锁算法。
这个分布式锁有3个重要的考量点：
        · 互斥(只能有一个客户端获取锁)
        · 不能死锁
· 容错(只要大部分 Redis 节点创建了这把锁就可以)

（1）Redis 最普通的分布式锁

第一个最普通的实现方式，就是在 Redis 里使用 SET key value [EX seconds][PX milliseconds] NX 创建一个 key，这样就算加锁。其中：
· NX ：表示只有 key 不存在的时候才会设置成功，如果此时 redis 中存在这个 key ，那么设置失败，返回 nil 。
· EX seconds ：设置 key 的过期时间，精确到秒级。意思是 seconds 秒后锁自动释放，别人创建的时候如果发现已经有了就不能加锁了。
· PX milliseconds ：同样是设置 key 的过期时间，精确到毫秒级。
比如执行以下命令：

SET resource_name my_random_value PX 30000 NX

释放锁就是删除 key ，但是一般可以用 lua 脚本删除，判断 value 一样才删除：

-- 删除锁的时候，找到 key 对应的 value，跟自己传过去的 value 做比较，如果是一样的才删除
if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])
elsereturn 0
end

为啥要用 random_value 随机值呢？因为如果某个客户端获取到了锁，但是阻塞了很长时间才执行完，比如说超过了 30s，此时可能已经自动释放锁了，此时可能别的客户端已经获取到了这个锁，要是你这个时候直接删除 key 的话会有问题，所以得用随机值加上面的 lua 脚本来释放锁。
但是这样是肯定不行的。因为如果是普通的 Redis 单实例，那就是单点故障。或者是 Redis 普通主从，那 Redis 主从异步复制，如果主节点挂了(key 就没有了)，key 还没同步到从节点，此时从节点切换为主节点，别人就可以 set key，从而拿到锁。

（2）RedLock 算法

        这个场景是假设有一个 Redis cluster，有5个 Redis master 实例。然后执行如下步骤获取一把锁：
        1.获取当前时间戳，单位是毫秒；
        2.跟上面类似，轮流尝试在每个 master 节点上创建锁，过期时间较短，一般就几十毫秒；

3.尝试在大多数节点上建立一个锁，比如5个节点就要求是3个节点 n/2+1；

4.客户端计算建立好锁的时间，如果建立锁的时间小于超时时间，就算建立成功了；

5.要是锁建立失败了，那么就依次之前建立过的锁删除；
6.只要别人建立了一把分布式锁，你就得不断轮询去尝试获取锁。

Redis 官方给出了以上两种基于 Redis 实现分布式锁的方法，详细说明可以查看：

https://redis.io/topics/distlock 。

2.zk 分布式锁

zk 分布式锁，其实可以做的比较简单，就是某个节点尝试创建临时 znode，此时创建成功了就获取了这个锁：这个时候别的客户端来创建锁会失败，只能注册个监听器监听这个锁。释放锁就是删除这个 znode，一旦释放掉就会通知客户端，然后有一个等待着的客户端就可以再次重新加锁。

/**
* ZooKeeperSession
*/
public class ZooKeeperSession {private static CountDownLatch connectedSemaphore = new CountDownLatch();private ZooKeeper zookeeper;private CountDownLatch latch;public ZooKeeperSession() {try {this.zookeeper = new ZooKeeper("192.168.31.187:2181,192.168.31try {connectedSemaphore.await();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("ZooKeeper session established......");} catch (Exception e) {e.printStackTrace();}
}/**
* 获取分布式锁
*
* @param productId
*/
public Boolean acquireDistributedLock(Long productId) {String path = "/product-lock-" + productId;try {zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, Crereturn true;} catch (Exception e) {while (true) {try {// 相当于是给node注册一个监听器，去看看这个监听器是否存在Stat stat = zk.exists(path, true);if (stat != null) {this.latch = new CountDownLatch(1);this.latch.await(waitTime, TimeUnit.MILLISECONDS);this.latch = null;}zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAreturn true;} catch (Exception ee) {continue;}}
}return true;}
}/**
* 释放掉一个分布式锁
*
* @param productId
*/
public void releaseDistributedLock(Long productId) {String path = "/product-lock-" + productId;try {zookeeper.delete(path, -1);System.out.println("release the lock for product[id=" + produc} catch (Exception e) {e.printStackTrace();}
}/**
* 建立 zk session 的 watcher
*/
private class ZooKeeperWatcher implements Watcher {public void process(WatchedEvent event) {System.out.println("Receive watched event: " + event.getState(if (KeeperState.SyncConnected == event.getState()) {connectedSemaphore.countDown();}if (this.latch != null) {this.latch.countDown();}}
}/**
* 封装单例的静态内部类
*/
private static class Singleton {private static ZooKeeperSession instance;static {instance = new ZooKeeperSession();}public static ZooKeeperSession getInstance() {return instance;}
}/**
* 获取单例
*
* @return
*/
public static ZooKeeperSession getInstance() {return Singleton.getInstance();
}/**
* 初始化单例的便捷方法
*/
public static void init() {getInstance();
}
}

也可以采用另一种方式，创建临时顺序节点：如果有一把锁，被多个人给竞争，此时多个人会排队，第一个拿到锁的人会执行，然后释放锁；后面的每个人都会去监听排在自己前面的那个人创建的 node ，一旦某个人释放了锁，排在自己后面的人就会被 ZooKeeper 给通知，一旦被通知了之后，就 ok了，自己就获取到了锁，就可以执行代码了。

public class ZooKeeperDistributedLock implements Watcher {private ZooKeeper zk;private String locksRoot = "/locks";private String productId;private String waitNode;private String lockNode;private CountDownLatch latch;private CountDownLatch connectedLatch = new CountDownLatch(1);private int sessionTimeout = 30000;public ZooKeeperDistributedLock(String productId) {this.productId = productId;try {String address = "192.168.31.187:2181,192.168.31.19:2181,192.1zk = new ZooKeeper(address, sessionTimeout, this);connectedLatch.await();} catch (IOException e) {throw new LockException(e);} catch (KeeperException e) {throw new LockException(e);} catch (InterruptedException e) {throw new LockException(e);}}public void process(WatchedEvent event) {if (event.getState() == KeeperState.SyncConnected) {connectedLatch.countDown();return;}if (this.latch != null) {this.latch.countDown();}}public void acquireDistributedLock() {try {if (this.tryLock()) {return;} else {waitForLock(waitNode, sessionTimeout);}} catch (KeeperException e) {throw new LockException(e);} catch (InterruptedException e) {throw new LockException(e);}}public boolean tryLock() {try {// 传入进去的locksRoot + “/” + productId// 假设productId代表了一个商品id，比如说1// locksRoot = locks// /locks/10000000000，/locks/10000000001，/locks/10000000002lockNode = zk.create(locksRoot + "/" + productId, new byte[0],// 看看刚创建的节点是不是最小的节点// locks：10000000000，10000000001，10000000002List<String> locks = zk.getChildren(locksRoot, false);Collections.sort(locks);if(lockNode.equals(locksRoot+"/"+ locks.get(0))){//如果是最小的节点,则表示取得锁return true;}//如果不是最小的节点，找到比自己小1的节点int previousLockIndex = -1;for(int i = 0; i < locks.size(); i++) {if(lockNode.equals(locksRoot + “/” + locks.get(i))) {previousLockIndex = i - 1;break;}}this.waitNode = locks.get(previousLockIndex);} catch (KeeperException e) {throw new LockException(e);} catch (InterruptedException e) {throw new LockException(e);}return false;}private boolean waitForLock(String waitNode, long waitTime) throws IntStat stat = zk.exists(locksRoot + "/" + waitNode, true);if (stat != null) {this.latch = new CountDownLatch(1);this.latch.await(waitTime, TimeUnit.MILLISECONDS);this.latch = null;}return true;}public void unlock() {try {// 删除/locks/10000000000节点// 删除/locks/10000000001节点System.out.println("unlock " + lockNode);zk.delete(lockNode, -1);lockNode = null;zk.close();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}public class LockException extends RuntimeException {private static final long serialVersionUID = 1L;public LockException(String e) {super(e);}public LockException(Exception e) {super(e);}}
}

3.redis 分布式锁和zk分布式锁的对比

        redis 分布式锁，需要自己不断去尝试获取锁，比较消耗性能。
        zk 分布式锁，获取不到锁，注册个监听器即可，不需要不断主动尝试获取锁，性能开销小。
        另外，如果是 Redis 获取锁的那个客户端出现 bug 挂了，那么只能等待超时时间才能释放锁；而zk 的话，因为创建的是临时 znode，只要客户端挂了，znode 就没了，此时自动释放锁。
        Redis 分布式锁大家没发现好麻烦吗？遍历上锁，计算时间等等；zk的分布式锁语义清晰实现简单。
        所以先不分析太多的东西，就说这两点，我个人实践认为 zk 的分布式锁比 Redis 的分布式锁牢靠、而且模型简单易用。

( ps：一个点赞一份爱，点个关注不迷路！)