分布式锁的实现原理

作者:来自 vivo 互联网服务器团队- Xu Yaoming

介绍分布式锁的实现原理。

一、分布式锁概述

分布式锁,顾名思义,就是在分布式环境下使用的锁。众所周知,在并发编程中,我们经常需要借助并发控制工具,如 mutex、synchronized 等,来保障线程安全。但是,这种线程安全仅作用在同一内存环境中。在实际业务中,为了保障服务的可靠性,我们通常会采用多节点进行部署。在这种分布式情况下,各实例间的内存不共享,线程安全并不能保证并发安全,如下例,同一实例中线程A与线程B之间的并发安全并不能保证实例1与实例2之间的并发安全:

图1.png

因此,当遇到分布式系统的并发安全问题时,我们就可能会需要引入分布式锁来解决。

用于实现分布式锁的组件通常都会具备以下的一些特性:

  • 互斥性:提供分布式环境下的互斥原语来加锁/释放锁,当然是分布式锁最基本的特性。

  • 自动释放:为了应对分布式系统中各实例因通信故障导致锁不能释放的问题,自动释放的特性通常也是很有必要的。

  • 分区容错性:应用在分布式系统的组件,具备分区容错性也是一项重要的特性,否则就会成为整个系统的瓶颈。

目前开源社区中常见的分布式锁解决方案,大多是基于具备集群部署能力的 key-value 存储中间件来实现,最为常用的方案基本上是基于 Redis、zookeeper 来实现,笔者将从上述分布式锁的特性出发,介绍一下这两类的分布式锁解决方案的优缺点。

二、分布式锁的实现原理

2.1 Redis 实现分布式锁

Redis 由于其高性能、使用及部署便利性,在很多场景下是实现分布式锁的首选。首先我们看下 Redis 是如何实现互斥性的。在单机部署的模式下,Redis 由于其单线程处理命令的线程模型,天然的具备互斥能力;而在哨兵/集群模式下,写命令也是单独发送到某个单独节点上进行处理,可以保证互斥性;其核心的命令是 set if not exist:

SET lockKey lockValue NX

成功设置 lockValue 的实例,就相当于抢锁成功。但如果持有锁的实例宕机,因为 Redis 服务端并没有感知客户端状态的能力,因此会出现锁无法释放的问题:

图2.png

这种情况下,就需要给 key 设置一个过期时间 expireTime:

SET lockKey lockValue EX expireTime NX

如果持有锁的实例宕机无法释放锁,则锁会自动过期,这样可以就避免锁无法释放的问题。在一些简单的场景下,通过该方式实现的分布式锁已经可以满足需求。但这种方式存在一个明显问题:如果业务的实际处理时间比锁过期时间长,锁就会被误释放,导致其他实例也可以加锁:

图3.png

这种情况下,就需要通过其他机制来保证锁在业务处理结束后再释放,一个常用的方式就是通过后台线程的方式来实现锁的自动续期。

图4.png

Redssion 是开源社区中比较受欢迎的一个 Java 语言实现的 Redis 客户端,其对 Java 中 Lock 接口定义进行扩展,实现了 Redis 分布式锁,并通过 watchDog 机制(本质上即是后台线程运作)来对锁进行自动续期。以下是一个简单的 Reddison 分布式锁的使用例子:

RLock rLock = RedissonClient.getLock("test-lock");
try {
    if (rLock.tryLock()) {
        // do something
    }
} finally {
    rLock.unlock();
}

Redssion 的默认实现 RedissonLock 为可重入互斥非公平锁,其 tryLock 方法会基于三个可选参数执行:

  • waitTime(获取锁的最长等待时长):默认为-1,waitTime 参数决定在获取锁的过程中是否需要进行等待,如果 waitTime>0,则在获取锁的过程中线程会等待一定时间并持续尝试获取锁,否则获取锁失败会直接返回。

  • leaseTime(锁持有时长):默认为-1。当 leaseTime<=0 时,会开启 watchDog 机制进行自动续期,而 leaseTime>0 时则不会进行自动续期,到达 leaseTime 锁即过期释放

  • unit(时间单位):标识 waitTime 及 leaseTime 的时间单位

我们不妨通过参数最全的 RedissonLock#tryLock(long waitTime, long leaseTime, TimeUnit unit) 方法源码来一探其完整的加锁过程:

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    ...
    // tryAcquire方法返回锁的剩余有效时长ttl,如果未上锁,则为null
    Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
    if (ttl == null) {
        // 获取锁成功
        return true;
    }
     
    // 计算剩余等待时长,剩余等待时长小于0,则不再尝试获取锁,获取锁失败,后续有多处同样的判断逻辑,将精简省略
   time -= System.currentTimeMillis() - current;
    if (time <= 0) {
        acquireFailed(waitTime, unit, threadId);
        return false;
    }
     
    // 等待时长大于0,则会对锁释放的事件进行订阅,持有锁的客户端在锁释放时会发布锁释放事件通知其他客户端抢锁,由此可得知该默认实现为非公平锁。
    // Redisson对Redis发布订阅机制的实现,底层大量使用了CompletableFuture、CompletionStage等接口来编写异步回调代码,感兴趣的读者可以详细了解,此处不作展开
    CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
    try {
        subscribeFuture.get(time, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        ...
    } catch (ExecutionException e) {
        ...
    }
 
    try {
        ...
        // 循环尝试获取锁
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                return true;
            }
            ...
            // 此处通过信号量来将线程阻塞一定时间,避免无效的申请锁浪费资源;在阻塞期间,如果收到了锁释放的事件,则会通过信号量提前唤起阻塞线程,重新尝试获取锁;
            currentTime = System.currentTimeMillis();
            if (ttl >= 0 && ttl < time) {
                // 若ttl(锁过期时长)小于time(剩余等待时长),则将线程阻塞ttl
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                // 若等待时长小于ttl,则将线程阻塞time
                commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
            }
            ...
        }
    } finally {
        // 取消订阅
        unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
    }
}

上述代码逻辑主要集中在处理 waitTime 参数,在并发竞争不激烈、可以容忍一定的等待时间的情况下,合理设置 waitTime 参数可以提高业务并发运行成功率,避免抢锁失败直接返回错误;但在并发竞争激烈、对性能有较高要求时,建议不设置 waitTime,或者直接使用没有 waitTime 参数的 lock() 方法,通过快速失败来提高系统吞吐量。

一个比较值得注意的点是,如果设置了 waitTime 参数,则 Redisson 通过将 RedissonLockEntry 中信号量(Semaphore)的许可证数初始化为0来达到一定程度的限流,保证锁释放后只有一个等待中的线程会被唤醒去请求 Redis 服务端,把唤醒等待线程的工作分摊到各个客户端实例上,可以很大程度上缓解非公平锁给 Redis 服务端带来的惊群效应压力。

public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {
    ...
    private final Semaphore latch;
 
    public RedissonLockEntry(CompletableFuture<RedissonLockEntry> promise) {
        super();
        //  RedissonLockEntry 中的Semaphore的许可证数初始化为0
        this.latch = new Semaphore(0);
        this.promise = promise;
    }
    ...
}

获取锁的核心逻辑,会通过 RedissonLock#tryAcquire 方法调用到 RedissonLock#tryAcquireAsync 方法。

private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    RFuture<Long> ttlRemainingFuture;
    if (leaseTime > 0) {
        // 若leaseTime大于零,会设置锁的租期为leaseTime
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        // 若leaseTime小于或等于零,会设置锁的租期为internalLockLeaseTime,这是一个通过lockWatchdogTimeout配置的值,默认为30s
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    }
 
    // 此处的handleNoSync方法是为了解决Redis发生故障转移,集群拓扑改变后,只有持有锁的客户端能再次获得锁的bug,为3.20.1版本修复,详见Redisson issue#4822
    CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
    ttlRemainingFuture = new CompletableFutureWrapper<>(s);
 
    // 根据加锁情况来进行后续处理
    CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
        // lock acquired
        // 若ttl为空,说明加锁不成功
        if (ttlRemaining == null) {
            if (leaseTime > 0) {
                // 若leaseTime>0,则将internalLockLeaseTime变量设置为leaseTime,以便后续解锁使用
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                // 若leaseTime<=0,则开启看门狗机制,通过定时任务进行锁续期
                scheduleExpirationRenewal(threadId);
            }
        }
        return ttlRemaining;
    });
    return new CompletableFutureWrapper<>(f);
}
 
// 加锁的lua脚本
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
            "if ((Redis.call('exists', KEYS[1]) == 0) " +
                        "or (Redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then " +
                    "Redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                    "Redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    "return nil; " +
                "end; " +
                "return Redis.call('pttl', KEYS[1]);",
            Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}

可以看到,若 leaseTime 大于0,则不会开启看门狗机制,锁在过期后即失效,在使用时请务必留意。上述代码中执行的 scheduleExpirationRenewal 方法即为看门狗机制的实现逻辑:

protected void scheduleExpirationRenewal(long threadId) {
    // 每个锁都会对应一个ExpirationEntry类,第一次加锁时不存在oldEntry
    ExpirationEntry = new ExpirationEntry();
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        // 非首次加锁,重入计数,不作其他操作
        oldEntry.addThreadId(threadId);
    } else {
        // 首次加锁,调用renewExpiration()方法进行自动续期
        entry.addThreadId(threadId);
        try {
            renewExpiration();
        } finally {
            // 若当前线程被中断,则取消对锁的自动续期。
            if (Thread.currentThread().isInterrupted()) {
                cancelExpirationRenewal(threadId);
            }
        }
    }
}
 
private void renewExpiration() {
    ...
    // 此处使用的是netty的时间轮来执行定时续期,此处不对时间轮做展开,感兴趣的读者可详细了解
    Timeout task = getServiceManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {
            ...
            CompletionStage<Boolean> future = renewExpirationAsync(threadId);
            future.whenComplete((res, e) -> {
                if (e != null) {
                    log.error("Can't update lock {} expiration", getRawName(), e);
                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                    return;
                }
                 
                if (res) {
                    // 若续期成功,则递归调用,等待任务的下一次执行
                    renewExpiration();
                } else {
                    // 若续期结果为false,说明锁已经过期了,或锁易主了,则清理当前线程关联的信息,等待线程结束
                    cancelExpirationRenewal(null);
                }
            });
        }
        // 时间轮的执行周期为internalLockLeaseTime / 3,即默认情况下,internalLockLeaseTime为30s时,每10s触发一次自动续期
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
     
    ee.setTimeout(task);
}
 
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
    // 执行重置过期时间的lua脚本
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (Redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                    "Redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    "return 1; " +
                    "end; " +
                    "return 0;",
            Collections.singletonList(getRawName()),
            internalLockLeaseTime, getLockName(threadId));
}

上面一段代码即是看门狗调度的核心代码,本质上即是通过定时调度线程执行 lua 脚本来进行锁续期。值得留意的是 scheduleExpirationRenewal

方法中的 ExpirationEntry,该对象与锁一一关联,会存储尝试获取该锁的线程(无论是否获取成功)以及重入锁的次数,在锁失效/锁释放时,会根据该对象中存储的线程逐一进行资源释放操作,以保证资源的正确释放。

最后,对上述 Redisson 可重入非公平锁源码进行一下总结:

  • Redisson 加锁时,根据 waitTime 参数是否大于0来决定加锁失败时采用等待并再次尝试/快速失败的策略;

  • Redisson 加锁时根据 leaseTime 参数是否小于等于0来决定是否开启看门狗机制进行定时续期;

  • Redisson 底层使用了 netty 实现的时间轮来进行定时续期任务的调度,执行周期为 internalLockLeaseTime / 3,默认为10s。

2.2 zookeeper 实现分布式锁

zookeeper(后文均简称 zk )基于 zab 协议实现的分布式协调服务,天生具备实现分布式锁的基础条件。我们可以从zk的一些基本机制入手,了解其是如何实现分布式锁的。

  • zab:为了保证分布式一致性,zk 实现了 zab(Zk Atomic Broadcast,zk 原子广播)协议,在 zab 协议下,zk集群分为 Leader 节点及 Follower 节点,其中,负责处理写请求的 Leader 节点在集群中是唯一的,多个 Follower 则负责同步 Leader 节点的数据,处理客户端的读请求。同时,zk 处理写请求时底层数据存储使用的是 ConcurrentHashMap,以保证并发安全;
public class NodeHashMapImpl implements NodeHashMap {
 
    private final ConcurrentHashMap<String, DataNode> nodes;
    private final boolean digestEnabled;
    private final DigestCalculator digestCalculator;
    private final AdHash hash;
     
    ...
 
}
  • 临时顺序节点:zk 的数据呈树状结构,树上的每一个节点为一个基本数据单元,称为 Znode。zk 可以创建一类临时顺序(EPHEMERAL_SEQUENTIAL)节点,在满足一定条件时会可以自动释放;同时,同一层级的节点名称会按节点的创建顺序进行命名,第一个节点为xxx-0000000000,第二个节点则为xxx-0000000001,以此类推;
图5.png
  • session:zk 的服务端与客户端使用 session 机制进行通信,简单来说即是通过长连接来进行交互,zk 服务端会通过心跳来监控客户端是否处于活动状态。若客户端长期无心跳或断开连接,则 zk 服务端会定期关闭这些 session,主动断开与客户端的通信。

了解了上述 zk 特点,我们不难发现 zk 也是具备互斥性、自动释放的特性的。同时,zk 由于 session 机制的存在,服务端可以感知到客户端的状态,因此不需要有由客户端来进行节点续期,zk 服务端可以主动地清理失联客户端创建的节点,避免锁无法释放的问题。zk 实现分布式锁的主要步骤如下:

  1. client1 申请加锁,创建 /lock/xxx-lock-0000000000节点(临时顺序节点),并监听其父节点 /lock;

  2. client1 查询 /lock 节点下的节点列表,并判断自己创建的 /xxx-lock-0000000000 是否为 /lock 节点下的第一个节点;当前没有其他客户端加锁,所以 client1 获取锁成功;

  3. 若 client2 此时来加锁,则会创建 /lock/xxx-lock-0000000001 节点;此时 client2 查询 /lock 节点下的节点列表,此时 /xxx-lock-0000000001 并非 /lock 下的第一个节点,因此加锁不成功,此时 client2 则会监听其上一个节点 /xxx-lock-0000000000;

  4. client1 释放锁,client1 删除 /xxx-lock-0000000000 节点,zk 服务端通过长连接 session 通知监听了 /xxx-lock-0000000000 节点的 client2 来获取锁

  5. 收到释放事件的 client2 查询 /lock 节点下的节点列表,此时自己创建的 /xxx-lock-0000000001 为最小节点,因此获取锁成功。

图6-1.png
图6-2.png
图6-3.png
图6-4.png

上述是 zk 公平锁的一种常见实现方式。值得注意的是, zk 客户端通常并不会实现非公平锁。事实上,zk 上锁的粒度不局限于上述步骤中的客户端,zk 客户端每次获取锁请求(即每一个尝试获取锁的线程)都会向 zk 服务端请求创建一个临时顺序节点。

以上述步骤为例,如果需要实现非公平锁,则会导致其余的所有节点都需要监听第一个节点 /xxx-lock-0000000000 的释放事件,相当于所有等待锁释放的线程都会监听同一个节点,这种机制无法像 Redisson 一样把唤醒锁的压力分摊到客户端上(或者说实现起来比较困难),会产生比较严重的惊群效应,因此使用 zk 实现的分布式锁一般情况下都是公平锁。

Curator 是一个比较常用的 zk 客户端,我们可以通过 Curator 的加锁过程,来了解 zk 分布式锁的设计原理。Curator 中比较常用的是可重入互斥公平锁 InterProcessMutex:

InterProcessMutex mutex = new InterProcessMutex(zkClient, "/lock");
try {
    // acquire方法的两个参数:等待时长及时间单位
    if (mutex.acquire(3, TimeUnit.SECONDS)) {
        log.info("加锁成功");
    } else {
        log.info("加锁失败");
    }
} finally {
    mutex.release();
}

InterProcessMutex 同样提供了等待时长参数,用于设置没有立即获取到锁时是快速失败还是阻塞等待,下一步,方法会调用到 InterProcessMutex#internalLock 方法中:

private boolean internalLock(long time, TimeUnit unit) throws Exception
{
    // 注释的意思:一个LockData对象只会被一个持有锁的线程进行修改,因此不需要对LockData进行并发控制。如此说明的原因是zk的互斥特性保证了下方attemptLock方法的互斥,由此保证了LockData不会被并发修改
    /*
        Note on concurrency: a given lockData instance
        can be only acted on by a single thread so locking isn't necessary
    */
 
    Thread currentThread = Thread.currentThread();
     
    // LockData用于记录当前持有锁的线程数据
    LockData lockData = threadData.get(currentThread);
    if ( lockData != null )
    {
        // 线程不为空,则进行重入,重入次数+1
        // re-entering
        lockData.lockCount.incrementAndGet();
        return true;
    }
     
    // 向zk服务获取分布式锁,getLockNodeBytes
    String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
    if ( lockPath != null )
    {
        // 若lockPath不为空,则获取锁成功,记录当前持有锁的线程
        LockData newLockData = new LockData(currentThread, lockPath);
        threadData.put(currentThread, newLockData);
        return true;
    }
 
    return false;
}

InterProcessMutex#internalLock会调用到 LockInternals#attemptLock 方法:

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{
    ...
    while ( !isDone )
    {
        isDone = true;
 
        try
        {
            // 创建锁节点
            ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
            // 判断是否成功获取锁
            hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
        }
        catch ( KeeperException.NoNodeException e )
        {
            // 捕获由于网络中断、session过期等原因导致的无法获得节点异常,此处根据配置的zk客户端重试策略决定是否重试,默认重试策略为Exponential Backoff
            ...retry or not...
        }
    }
 
    if ( hasTheLock )
    {
        return ourPath;
    }
 
    return null;
}
 
public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception
{
    String ourPath;
    if ( lockNodeBytes != null )
    {  
        // 在其他类型的锁实现中,lockNodeBytes可能不为空,则根据lockNodeBytes来获取节点路径,此处暂不作展开
        ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, lockNodeBytes);
    }
    else
    {
        // 在可重入互斥锁中,客户端向zk服务端请求创建一个 EPHEMERAL_SEQUENTIAL 临时顺序节点
        ourPath = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
    }
    return ourPath;
}

上述代码中,创建锁节点并不会产生互斥,而是会直接向 zk 服务端请求创建临时顺序节点。此时,客户端还未真正的获得锁,判断加锁成功的核心逻辑在 LockInternals#internalLockLoop 方法中:

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
{
    boolean     haveTheLock = false;
    boolean     doDelete = false;
    try
    {
        if ( revocable.get() != null )
        {  
            // curator锁撤销机制,通过实现Curator中的Revocable接口的makeRevocable方法,可以将锁设置为可撤销锁,其他线程可以在符合条件时将锁撤销,此处暂不涉及
            client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
        }
         
        // 客户端实例就绪,则尝试循环获取锁
        while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) 
        {
            // 获取当前父节点下的排好序的子节点
            List<String>        children = getSortedChildren();
            // 得到当前节点名
            String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
            // 根据 children 列表与当前节点名,计算当前节点是否为第一个节点,若不是第一个节点,则在 PredicateResults中返回需要监听的前一个节点节点,若为最小节点,则获取锁成功
            PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
            if ( predicateResults.getsTheLock() )
            {
                // 获取锁成功
                haveTheLock = true;
            }
            else
            {
                // 拼接前一个节点的节点路径
                String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
                 
                synchronized(this)
                {
                    try
                    {
                        // 将前一个节点的监听器放到当前客户端中,当前一个节点被释放时,就会唤醒当前客户端
                        client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                        if ( millisToWait != null )
                        {
                            millisToWait -= (System.currentTimeMillis() - startMillis);
                            startMillis = System.currentTimeMillis();
                            // 计算剩余等待时长,若等待时长小于0,则不再尝试获取锁,并标记当前线程创建的节点需要删除
                            if ( millisToWait <= 0 )
                            {
                                doDelete = true;    // timed out - delete our node
                                break;
                            }
                            // 若等待时长大于0,则阻塞线程,等待锁释放
                            wait(millisToWait);
                        }
                        else
                        {
                            // 在其他的一些加锁场景中,默认会持久等待到锁释放位置,当前可重入互斥锁暂不涉及
                            wait();
                        }
                    }
                    catch ( KeeperException.NoNodeException e )
                    {
                        // it has been deleted (i.e. lock released). Try to acquire again
                    }
                }
            }
        }
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        doDelete = true;
        throw e;
    }
    finally
    {
        if ( doDelete )
        {
            // 删除当前节点
            deleteOurPath(ourPath);
        }
    }
    return haveTheLock;
}
 
private synchronized void notifyFromWatcher()
{
    // 当zk客户端收到锁释放事件时,会遍历当前客户端注册过的所有的监听器,并找到合适的监听器进行回调,最终通过notifyAll唤醒监听被释放节点的线程
    notifyAll();
}

上述 curator 加锁的核心代码虽然比较长,但整体逻辑与我们前面分析过的加锁逻辑是一致的,主要做了三件事:

  • 获取当前父节点的有序子节点序列;

  • 判断当前节点是否为第一个节点;

  • 若为第一个节点,则获取锁成功,否则为当前 zk 客户端增加一个前一节点的监听器,如果此时还在等待时长内,则使用wait方法挂起线程,否则删除当前节点。

三、总结——如何选择合适的分布式并发安全解决方案?

  • 绕不过的 CAP 理论

Redis 与 zk 由于客户端与服务端的交互机制上存在比较大的差异,相应的分布式锁实现原理也有所不同。两者都是优秀的支持分布式部署的系统,自然具备分区容错性,但分布式系统总绕不过去一个经典的问题——CAP理论:在满足了分区容错性的前提下,分布式系统只能满足可用性、数据一致性两者其一。

图7.png

对比之下,Redis 在可用性上更胜一筹,属于 AP 系统;zk 具备更强的数据一致性,属于 CP 系统,而基于 AP、CP 的特性去实现的分布式锁,自然也会存在不同程度的问题。

  • Redis 分布式锁的一致性问题

Redis 的集群模式并没有严格地实现分布式共识算法,因此 Redis 是不具备一致性的。为了保证高可用性,Redis 集群的主从节点使用的是异步复制,从节点并不保证与主节点数据一致,只能尽量的追赶主节点的最新数据;因此,当主节点发生故障,进行主从切换时,实际上有可能会发生数据丢失问题:

图8.png
  • zk 性能及可用性问题

zk 实现了 zab 算法,在数据一致性上给出了比较可靠的方案,但是由于 zab 协议的两阶段提交要求所有节点的写请求处理就绪后,才算写入成功,这无疑会导致性能的下降。此外,在zk集群发生 leader 重选举的过程中,对外会表现为不可用状态,此时可用性上就会存在问题:

图9.png

由上可知,分布式并发安全解决方案并不存在完美的“银弹”,因此更多时候我们应当根据自身业务情况,合理地选择合适的解决方案。

显而易见地,如果业务场景有较高的请求量,并发竞争比较激烈,对性能有较高要求,此时通过 Redis 来实现分布式锁会是比较合适的方案。但是如果业务场景对数据一致性要求比较高,或是系统交互链路比较长,一但发生数据不一致时,会导致系统出现难以恢复的问题时,采用zk来实现分布式锁则是更优的解决方案。

  • 上述方案都无法满足要求?

总体上看,Redis 由于其本身的高性能可以满足大多数场景下的性能要求,而 zk 则保证了较高数据一致性。但倘若遇到了既要求高性能、又要求数据一致性、还要引入锁机制来保障并发安全的场景,这时候就必须重新审视系统设计是否合理了,毕竟高并发与锁是一对矛盾,可用性与数据一致性是一对矛盾,我们应该通过良好的方案、系统设计,来避免让我们的系统陷入这些矛盾的困境中。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,029评论 6 492
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,395评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,570评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,535评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,650评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,850评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,006评论 3 408
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,747评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,207评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,536评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,683评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,342评论 4 330
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,964评论 3 315
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,772评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,004评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,401评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,566评论 2 349

推荐阅读更多精彩内容