细说分布式锁实现

分手后的思念是犯贱 2021-11-16 21:32 220阅读 0赞

## I. 分布式锁 ##

分布式锁的主要目的是：在分布式系统多进程下，保证只有一个进程能够执行。

参考单机的锁特点来说，我们实现的分布式锁需要拥有以下特性：

*  互斥性
 *  不发生死锁
 *  正确释放锁
 *  高性能
 *  高可用

最好还能支持：

*  可重入
 *  阻塞获取/超时获取/中断获取
 *  公平/非公平

## II. Redis分布式锁 ##

Redis获取分布式锁主要是靠 `setnx` 命令来实现的，该命令表示当 key 存在时，set 失败，返回为0，key 不存在 set 成功返回 1。

### 单机Redis分布式锁实现 ###

分布式锁的主要关注点在于获取锁、释放锁以及避免死锁，在实现一个锁工具支持的功能时可以考虑的文章开始的8个点。Redis分布式锁的实现，其中获取锁主要依靠 `setnx` 命令，释放锁删除对应的 key 即可，避免死锁则依靠 key 的自动失效来保证。

目前看到其他的文章主要的实现方式以下几种：

> [分布式锁看这篇就够了][Link 1]  
>    
> **获取锁**：当一个进程前来请求分布式锁，首先通过 `setnx` 来尝试获取锁，如果获取锁成功，OK，设置好过期时间，可以返回去干自己事情了。如果 `setnx` 没获取成功，事情麻烦了，说明肯定有人拿到锁了。这个时候就要看看拿锁的进程是不是该让让位了——比较当前时间和锁里设置的过期时间（通过 `get` 查看），如果没到过期时间，好吧，自己就只好放弃或者继续等待了。如果发现当前时间已经超过过期时间了，不好意思，自己获取锁咯，利用 `getset` 设置一个新的过期时间，拿出旧的过期时间。看看旧的过期时间是不是和上一次看到的一样，卧槽，竟然不一样说明被哪个进程先下手给改了新的过期时间，那自己又没获取锁成功。  
> **释放锁**：直接删除key

这种实现思路乍一看没问题，其实问题很多，首先没有正确的释放锁。比如：当A进程获取到锁，由于某些原因执行时间较久，到了锁过期时间自动释放锁。这时，B进程过来获取锁，也执行一段代码，正在执行着，A进程释放锁（直接删除key），结果把B的锁反而给释放了。这出大事情了！！C进程可以获取到锁了…

这种方式的改进在于释放锁时，不要直接释放，而是判断只有 key 未过期才进行释放。这样A就不会删除B设置的 key 了。也就是说，只有未超时就结束业务的进程才能释放锁，锁的互斥性简介保证了正确的释放锁——解铃还须系铃人。但还是没有完全解决问题，因为可能刚判断完key未过期，然后当前线程停顿了一会，在这过程中发生了锁到了失效时间被其他进程获取，然后原来线程释放锁，GG，还是出现之前的莫名其妙释放锁问题。

其次，该方案利用时间戳，决定了该方法强依赖分布式系统的时间同步。如果不同步，必然会给算法带来影响。

同时，`getset` 是一个原子操作，但是每一个尝试获取锁的进程都会进行 `set`，也就是说这一步其实是会发生覆盖操作的。比如AB两进程同时获取锁，A `getset` 成功获取锁，B也进行`getset` 获取，B会把新的过期时间设置进去，A的过期时间返回出来。但B获取锁是失败的，却莫名其妙更新的A的过期时间，这是不合理的。

另一种极为常见的实现方案：

> `setnx` 和 `expire` 两步操作非原子、释放锁检查和释放动作两步操作也非原子  
>    
> **首先**我们分析 `setnx` 和 `expire` 两步操作非原子带来的影响，如果先 `setnx` 成功，如果此时因为某种原因，当前进程挂了，则无法 `expire`，最终导致死锁。这其实问题还是不可忽略的。  
> **其次**释放锁时讲究解铃还须系铃人，那么就需要进行锁的检查，是否是自己的。当锁检查完毕时自己加上的，那么便可以释放锁。这两步动作如果不是原子，出现了胡乱释放锁的问题前文已经讨论过。

那么考虑到以上细节，这里实现一个单机版的Redis锁实现。

@Slf4j
    public class RedisLock { 
    
        private RedisTemplate<String, Serializable> redisTemplate;
        private Integer sleepInterval;
    
        public RedisLock(RedisTemplate<String, Serializable> redisTemplate) { 
            this(redisTemplate, 5);
        }
    
        public RedisLock(RedisTemplate<String, Serializable> redisTemplate, Integer sleepInterval) { 
            this.redisTemplate = redisTemplate;
            this.sleepInterval = sleepInterval;
        }
    
        /** * 阻塞获取锁 */
        public void lock(String key, String lockId, long expireTime, TimeUnit timeUnit) { 
            for (;;) { 
                if (tryLock(key, lockId, expireTime, timeUnit)) { 
                    return;
                }
                try { 
                    TimeUnit.MILLISECONDS.sleep(sleepInterval);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        /** * 快速尝试获取锁 * set nx ex 整体原子性 * @param key 锁key * @param lockId 锁value（区分锁的id号） * @param expireTime 锁的过期时间 * @param timeUnit 时间单位 * @return */
        public boolean tryLock(String key, String lockId, long expireTime, TimeUnit timeUnit) { 
            Boolean res = redisTemplate.execute((RedisCallback<Boolean>) connection -> connection.set(key.getBytes(), lockId.getBytes(), Expiration.from(expireTime, timeUnit), RedisStringCommands.SetOption.SET_IF_ABSENT));
            if (Boolean.TRUE.equals(res)) { 
                log.info("Thread {} get lock success, key:{} lockId:{} expire:{} timeUnit:{}.", Thread.currentThread(), key, lockId, expireTime, timeUnit);
                return true;
            }
            return false;
        }
    
        /** * 超时获取锁 * @param key 锁key * @param lockId 锁value（区分锁的id号） * @param expireTime 锁的过期时间 * @param time 获取锁超时时间 * @param unit 时间单位 * @return * @throws InterruptedException */
        public boolean tryLock(String key, String lockId, long expireTime, long time, TimeUnit unit) { 
            long timeout = System.currentTimeMillis() + unit.toMillis(time);
            while (System.currentTimeMillis() <= timeout) { 
                if (tryLock(key, lockId, expireTime, unit)) { 
                    return true;
                }
                try { 
                    TimeUnit.MILLISECONDS.sleep(sleepInterval);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            return false;
        }
    
        /** * 释放锁 * 如果锁还未释放，则释放自己的锁（原子操作，且保证了解铃还须系铃人） * @param key 锁的key * @param lockId 锁id * @return */
        public boolean unlock(String key, String lockId) { 
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Long res = redisTemplate.execute((RedisCallback<Long>) connection -> connection.eval(script.getBytes(), ReturnType.INTEGER, 1, key.getBytes(), lockId.getBytes()));
            if (Long.valueOf(1).equals(res)) { 
                log.info("Thread {} release lock success, key:{} lockId:{}.", Thread.currentThread(), key, lockId);
                return true;
            }
            return false;
        }
    
        /** * 生成锁的id * @return */
        public static String generateLockId() { 
            return UUID.randomUUID().toString() + "-" + Thread.currentThread().getId();
        }
    }

测试实现的Redis分布式锁组件：

@Controller
    public class HelloController { 
        @Autowired
        private RedisTemplate<String, Serializable> redisTemplate;
    
        private int i = 0;
    
        @RequestMapping("/hello")
        @ResponseBody
        public String hello() { 
            RedisLock lock = new RedisLock(redisTemplate);
            String key = "hello";
            String lockId = RedisLock.generateLockId();
            try { 
                lock.lock(key, lockId, 2, TimeUnit.SECONDS);
                i++;
                System.out.println("do something......" + i);
            } finally { 
                lock.unlock(key, lockId);
            }
            return "hello world!";
        }
    
        @RequestMapping("/hello2")
        @ResponseBody
        public String hello2() { 
            i++;
            System.out.println("do something......" + i);
            return "hello world!";
        }
    }

其中 */hello* 为加锁同步方法，*/hello2* 不加锁。加锁同步的 `i` 是有序的递增的，而未加锁是乱序的，读者可以自行尝试。

### RedLock与Redisson ###

上面实现的单机Redis分布式锁，基本满足了互斥性、不发生死锁、正确释放锁。高性能还算勉强，至于高可用，如果单机Redis挂了，那么分布式锁服务就直接没用了。

现有保证Redis分布式锁的高可用方案，有RedLock算法，其Java实现是Redisson里的分布式锁。

关于RedLock的正确与否，Redis作者和一位分布式系统专家有过相关的讨论：

*  [How to do distributed locking][]
 *  [Is Redlock safe?][Is Redlock safe]

里面比较关注的一个细节可能你早就注意到了，为了防止死锁的发生，我们给锁设置了自动释放时间。那么就会担心，如果A进程在拿到锁之后执行了很长一段时间的业务代码，超过了租期，锁自动释放了，那么其他进程就可以拿到锁继续执行，这就会引来并发问题。没错，这个问题是无法解决的。

了解RedLock，推荐官方文档 [Distributed locks with Redis][]

关于Redisson的分布式锁，参考[Distributed-locks-and-synchronizers][]

## III. ZooKeeper分布式锁 ##

ZooKeeper实现分布式锁主要依靠ZK上创建临时唯一节点实现，多个进程竞争在ZK上创建一个临时节点，ZK保证只有一个节点创建成功，创建成功的进程相当于成功获取锁，其他未获取锁的进程则监听该临时节点的删除事件（`Watcher`），重新获取锁。

目前关于ZK实现分布式锁原理的文章大多相似，目前主要实现思路如下：

> [七张图彻底讲清楚ZooKeeper分布式锁的实现原理][ZooKeeper]  
>    
> 文中的方案即是避免“惊群效应”的ZK分布式锁方案，但最终实现的效果是一个公平锁，大家排队获取锁，取锁顺序和临时节点号顺序有关。  
> 其他一些文章自己动手实现的分布式锁，要么质量无法保证，没有生产环境的验证，或者只是个demo；又或者功能不够完善，没有达到本文开头所说的那几个要求。

本文着重主动分析已经实现好基于ZK的分布式锁Curator框架源码，来仔细看看其算法实现。

public class InterProcessMutex implements InterProcessLock, Revocable<InterProcessMutex> { 
        private final LockInternals internals;
        private final String basePath;
        private final ConcurrentMap<Thread, InterProcessMutex.LockData> threadData;
        private static final String LOCK_NAME = "lock-";
    
        public InterProcessMutex(CuratorFramework client, String path) { 
            this(client, path, new StandardLockInternalsDriver());
        }
    
        public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver) { 
            this(client, path, "lock-", 1, driver);
        }
    
    	InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) { 
            this.threadData = Maps.newConcurrentMap();
            this.basePath = PathUtils.validatePath(path);
            this.internals = new LockInternals(client, driver, path, lockName, maxLeases);
        }
    
    	/** * 阻塞获取锁，获取成功/失去连接 才会退出方法 */ 
        public void acquire() throws Exception { 
            if (!this.internalLock(-1L, (TimeUnit)null)) { 
                throw new IOException("Lost connection while trying to acquire lock: " + this.basePath);
            }
        }
    
    	/** * 超时获取锁，获取成功/超时 才会退出方法 */
        public boolean acquire(long time, TimeUnit unit) throws Exception { 
            return this.internalLock(time, unit);
        }
    
        public boolean isAcquiredInThisProcess() { 
            return this.threadData.size() > 0;
        }
    
    	/** * 释放锁 */	
        public void release() throws Exception { 
            Thread currentThread = Thread.currentThread();
            InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);
            if (lockData == null) { 
                throw new IllegalMonitorStateException("You do not own the lock: " + this.basePath);
            } else { 
                int newLockCount = lockData.lockCount.decrementAndGet();
                if (newLockCount <= 0) { 
                    if (newLockCount < 0) { 
                        throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + this.basePath);
                    } else { 
                    	// 真正释放锁
                        try { 
                            this.internals.releaseLock(lockData.lockPath);
                        } finally { 
                            this.threadData.remove(currentThread);
                        }
    
                    }
                }
            }
        }
    
        public Collection<String> getParticipantNodes() throws Exception { 
            return LockInternals.getParticipantNodes(this.internals.getClient(), this.basePath, this.internals.getLockName(), this.internals.getDriver());
        }
    
        protected byte[] getLockNodeBytes() { 
            return null;
        }
    
        protected String getLockPath() { 
            InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(Thread.currentThread());
            return lockData != null ? lockData.lockPath : null;
        }
    
    	/** * 指定时间内获取锁 */
        private boolean internalLock(long time, TimeUnit unit) throws Exception { 
        	// 获取当前线程，用于支持可重入
            Thread currentThread = Thread.currentThread();
            InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);
            if (lockData != null) { 
            	// 锁计数器+1 (对于当前线程来说应线程安全，是否可以不要原子类，普通i++也可以)
                lockData.lockCount.incrementAndGet();
                return true;
            } else { 
            	// 阻塞调用attemptLock方法获取锁，获取成功则返回zk节点名称
                String lockPath = this.internals.attemptLock(time, unit, this.getLockNodeBytes());
                if (lockPath != null) { 
                    InterProcessMutex.LockData newLockData = new InterProcessMutex.LockData(currentThread, lockPath, null);
                    this.threadData.put(currentThread, newLockData);
                    return true;
                } else { 
                    return false;
                }
            }
        }
    
        private static class LockData { 
            final Thread owningThread;
            final String lockPath;
            // 锁计数器
            final AtomicInteger lockCount;
    
            private LockData(Thread owningThread, String lockPath) { 
                this.lockCount = new AtomicInteger(1);
                this.owningThread = owningThread;
                this.lockPath = lockPath;
            }
        }
    }

进一步进入 `LockInternals` 查看：

public class LockInternals { 
        private final WatcherRemoveCuratorFramework client;
     	// path=/basepath/lockname前缀+顺序号
        private final String path;
        private final String basePath;
        private final LockInternalsDriver driver;
        private final String lockName;
        private final AtomicReference<RevocationSpec> revocable = new AtomicReference((Object)null);
        private final CuratorWatcher revocableWatcher = new CuratorWatcher() { 
            public void process(WatchedEvent event) throws Exception { 
                if (event.getType() == EventType.NodeDataChanged) { 
                    LockInternals.this.checkRevocableWatcher(event.getPath());
                }
    
            }
        };
        private final Watcher watcher = new Watcher() { 
            public void process(WatchedEvent event) { 
            	// 唤醒所有的线程
                LockInternals.this.notifyFromWatcher();
            }
        };
        private volatile int maxLeases;
        static final byte[] REVOKE_MESSAGE = "__REVOKE__".getBytes();
    
        public void clean() throws Exception { 
            try { 
                this.client.delete().forPath(this.basePath);
            } catch (BadVersionException var2) { 
                ;
            } catch (NotEmptyException var3) { 
                ;
            }
    
        }
    
        LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) { 
            this.driver = driver;
            this.lockName = lockName;
            this.maxLeases = maxLeases;
            this.client = client.newWatcherRemoveCuratorFramework();
            this.basePath = PathUtils.validatePath(path);
            this.path = ZKPaths.makePath(path, lockName);
        }
    
        synchronized void setMaxLeases(int maxLeases) { 
            this.maxLeases = maxLeases;
            this.notifyAll();
        }
    
        void makeRevocable(RevocationSpec entry) { 
            this.revocable.set(entry);
        }
    
    	/** * 释放锁调用方法 */
        final void releaseLock(String lockPath) throws Exception { 
            this.client.removeWatchers();
            this.revocable.set((Object)null);
            this.deleteOurPath(lockPath);
        }
    
        CuratorFramework getClient() { 
            return this.client;
        }
    
        public static Collection<String> getParticipantNodes(CuratorFramework client, final String basePath, String lockName, LockInternalsSorter sorter) throws Exception { 
            List<String> names = getSortedChildren(client, basePath, lockName, sorter);
            Iterable<String> transformed = Iterables.transform(names, new Function<String, String>() { 
                public String apply(String name) { 
                    return ZKPaths.makePath(basePath, name);
                }
            });
            return ImmutableList.copyOf(transformed);
        }
    
    
        LockInternalsDriver getDriver() { 
            return this.driver;
        }
    
    	/** * 尝试获取锁 */
        String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception { 
            long startMillis = System.currentTimeMillis();
            Long millisToWait = unit != null ? unit.toMillis(time) : null;
            byte[] localLockNodeBytes = this.revocable.get() != null ? new byte[0] : lockNodeBytes;
            int retryCount = 0;
            String ourPath = null;
            boolean hasTheLock = false;  // 标记当前线程是否成功获取锁
            boolean isDone = false;
    
            while(!isDone) { 
                isDone = true;
    
                try { 
                	// 创建临时有序节点，返回属于当前线程的节点名称
                    ourPath = this.driver.createsTheLock(this.client, this.path, localLockNodeBytes);
                    // 循环获取锁，返回获取锁是否成功
                    hasTheLock = this.internalLockLoop(startMillis, millisToWait, ourPath);
                } catch (NoNodeException var14) { 
                    if (!this.client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) { 
                        throw var14;
                    }
    
                    isDone = false;
                }
            }
    
            return hasTheLock ? ourPath : null;
        }
    
        private void checkRevocableWatcher(String path) throws Exception { 
            RevocationSpec entry = (RevocationSpec)this.revocable.get();
            if (entry != null) { 
                try { 
                    byte[] bytes = (byte[])((BackgroundPathable)this.client.getData().usingWatcher(this.revocableWatcher)).forPath(path);
                    if (Arrays.equals(bytes, REVOKE_MESSAGE)) { 
                        entry.getExecutor().execute(entry.getRunnable());
                    }
                } catch (NoNodeException var4) { 
                    ;
                }
            }
    
        }
    
        private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception { 
            boolean haveTheLock = false;  // 获取成功标记
            boolean doDelete = false;  // 
    
            try { 
                if (this.revocable.get() != null) { 
                    ((BackgroundPathable)this.client.getData().usingWatcher(this.revocableWatcher)).forPath(ourPath);
                }
    
    			// 没有获取到锁会一直循环，发生异常会跳出循环
                while(this.client.getState() == CuratorFrameworkState.STARTED && !haveTheLock) { 
                	// 获取当前从小到大排序的节点
                    List<String> children = this.getSortedChildren();
                    // 获取当前线程所属的节点序号
                    String sequenceNodeName = ourPath.substring(this.basePath.length() + 1);
                    // 判断是否获取锁
                    PredicateResults predicateResults = this.driver.getsTheLock(this.client, children, sequenceNodeName, this.maxLeases);
                    if (predicateResults.getsTheLock()) { 
                        haveTheLock = true;
                    } else { 
                    	// 获取前一个节点
                        String previousSequencePath = this.basePath + "/" + predicateResults.getPathToWatch();
                        // 用this进行来实现线程等待唤醒
                        synchronized(this) { 
                            try { 
                                // 使用getData()接口而不是checkExists()是因为，如果前一个子节点已经被删除了那么会抛出异常而且不会设置事件监听器，而checkExists虽然也可以获取到节点是否存在的信息但是同时设置了监听器，这个监听器其实永远不会触发，对于zookeeper来说属于资源泄露
                                ((BackgroundPathable)this.client.getData().usingWatcher(this.watcher)).forPath(previousSequencePath);
                                if (millisToWait == null) { 
                                	// 未设置超时时间，一直等待直到被唤醒再次循环
                                    this.wait();
                                } else { 
                                	// 更新等待时间
                                    millisToWait = millisToWait.longValue() - (System.currentTimeMillis() - startMillis);
                                  	// 更新开始时间
                                    startMillis = System.currentTimeMillis();
                                    if (millisToWait.longValue() > 0L) { 
                                    	// 未超时则最多等待一段时间，等待被唤醒再次循环
                                        this.wait(millisToWait.longValue());
                                    } else { 
                                    	// 超时退出循环
                                        doDelete = true;
                                        break;
                                    }
                                }
                            } catch (NoNodeException var19) { 
                                ;
                            }
                        }
                    }
                }
            } catch (Exception var21) { 
                ThreadUtils.checkInterrupted(var21);
                doDelete = true;
                throw var21;
            } finally { 
            	// 如果超时了，则删除节点
                if (doDelete) { 
                    this.deleteOurPath(ourPath);
                }
    
            }
    
            return haveTheLock;
        }
    
        private void deleteOurPath(String ourPath) throws Exception { 
            try { 
                ((ChildrenDeletable)this.client.delete().guaranteed()).forPath(ourPath);
            } catch (NoNodeException var3) { 
                ;
            }
        }
    
        private synchronized void notifyFromWatcher() { 
            this.notifyAll();
        }
    }

[Link 1]: https://zhuanlan.zhihu.com/p/42056183
[How to do distributed locking]: http://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
[Is Redlock safe]: http://antirez.com/news/101
[Distributed locks with Redis]: https://redis.io/topics/distlock
[Distributed-locks-and-synchronizers]: https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers
[ZooKeeper]: https://juejin.im/post/5c01532ef265da61362232ed