1、多个客户端竞争创建 lock 临时节点 2、其中某个客户端成功创建 lock 节点，其他客户端对 lock 节点设置 watcher 3、持有锁的客户端删除 lock 节点或该客户端崩溃，由 Zookeeper 删除 lock 节点 4、其他客户端获得 lock节点被删除的通知 5、重复上述4个步骤，直至无客户端在等待获取锁了
读写锁 一开始，所有的客户端都会创建自己的锁节点。客户端从 Zookeeper 端获取 /share_lock 下所有的子节点。 读锁节点（满足其中一个即可） 1、自己创建的节点序号排在所有其他子节点前面 2、自己创建的节点前面无写锁节点 写锁节点 自己创建的锁节点是否排在其他子节点前面1、所有客户端创建自己的锁节点 2、从 Zookeeper 端获取 /share_lock 下所有的子节点 3、判断自己创建的锁节点是否可以获取锁，如果可以，持有锁。否则对自己关心的锁节点设置 watcher 4、持有锁的客户端删除自己的锁节点，某个客户端收到该节点被删除的通知，并获取锁 5、重复步骤4，直至无客户端在等待获取锁了
CAP 理论告诉我们，一个分布式系统不可能同时满足以下三种
一致性（C:Consistency）可用性（A:Available）分区容错性（P:Partition Tolerance）zk本身更多是偏向于支持CP的模式,分布式锁需要对高可用有一定支持性。
当分布式锁加锁过程中，出现未知异常，需要主动释放锁，通常会采用try…finally的模版方式来做实现。 对于每次加锁都应该设置超时机制，否则会一直处于抢占资源的状态不做释放。尤其是高并发场景中，如果加锁过程中出现了某些不可预料的情况，导致锁没有正常释放，一直在redis中存储且没有设置过期时间，那么会一直占用redis的资源。 对于加锁过程中，业务调用接口长时间堵塞（但是并没有抛出异常，例如说查询数据的数据量非常大，处理时间很久），堵塞时间远远超过了加锁的时常，那么这个时候应该设计一种自动给锁进行延期的机制。 这里可以参考一下redisson框架的设计思路，内部采用了watch dog机制来做这块的优化。
setnx():该方法是原子的，如果key不存在，则设置当前key成功，返回1；如果当前key已经存在，则设置当前key失败，返回0。 getset()命令：该方法是原子的，对key设置newValue这个值，并且返回key原来的旧值。
双重防死锁，使用setNx + getSet两个原子性方法。
setnx(lockkey, 当前时间+过期超时时间) ，如果返回1，则获取锁成功；如果返回0则没有获取到锁，转向2。get(lockkey)获取值oldExpireTime ，并将这个value值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向3。计算newExpireTime=当前时间+过期超时时间，然后getset(lockkey, newExpireTime) 会返回当前lockkey的值currentExpireTime。判断currentExpireTime与oldExpireTime 是否相等，如果相等，说明当前getset设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行delete释放锁；如果大于锁设置的超时时间，则不需要再锁进行处理。 public void closeOrderTaskV3(){ log.info("关闭订单定时任务启动"); long lockTimeout = Long.parseLong(PropertiesUtil.getProperty("lock.timeout","5000")); Long setnxResult = RedisShardedPoolUtil.setnx(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK,String.valueOf(System.currentTimeMillis()+lockTimeout)); if(setnxResult != null && setnxResult.intValue() == 1){ closeOrder(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK); }else{ //未获取到锁，继续判断，判断时间戳，看是否可以重置并获取到锁 String lockValueStr = RedisShardedPoolUtil.get(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK); if(lockValueStr != null && System.currentTimeMillis() > Long.parseLong(lockValueStr)){ String getSetResult = RedisShardedPoolUtil.getSet(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK,String.valueOf(System.currentTimeMillis()+lockTimeout)); //再次用当前时间戳getset。 //返回给定的key的旧值，->旧值判断，是否可以获取锁 //当key没有旧值时，即key不存在时，返回nil ->获取锁 //这里我们set了一个新的value值，获取旧的值。 if(getSetResult == null || (getSetResult != null && StringUtils.equals(lockValueStr,getSetResult))){ //真正获取到锁 closeOrder(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK); }else{ log.info("没有获取到分布式锁:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK); } }else{ log.info("没有获取到分布式锁:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK); } } log.info("关闭订单定时任务结束"); }当然，我们可以最便捷的使用Redisson实现分布式锁。
1.网络抖动 进程A中的一个线程获取到了锁，然后执行finally中的释放锁的代码时，由程序到Redis的网络不好了，所以释放锁失败。 2.服务端宕机 进程A获取到了锁，Redis服务器宕机了，所以锁没有释放。
解锁的步骤有2步：需要先判断自己是否能够删除锁（是否是当前线程加的锁），然后再执行删除锁，这个过程由于是2个步骤，所以需要保证原子性才行，否则就会出问题。
RedLock 的思想是使用多台 Redis Master ，节点完全独立，节点间不需要进行数据同步，因为 Master-Slave 架构一旦 Master 发生故障时数据没有复制到 Slave，被选为 Master 的 Slave 就丢掉了锁，另一个客户端就可以再次拿到锁。锁通过 setNX（原子操作） 命令设置，在有效时间内当获得锁的数量大于 (n/2+1) 代表成功，失败后需要向所有节点发送释放锁的消息。
