【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理

栏目: IT技术 · 发布时间: 6年前

内容简介：前面已经讲解了Zookeeper可重入锁的实现原理，自己对分布式锁也有了更深的认知。我在公众号中发了一个疑问，相比于Redis来说，Zookeeper的实现方式要更好一些，即便Redis作者实现了RedLock算法来解决Redis集群模式下分布式锁的弊端，但Redis实现的分布式锁仍然不是那么完美。比如有5台Redis集群，按照n/2 + 1代表获取锁成功，如果客户端A此时获取锁，Redis集群（1,2,3）返回成功，客户端A获取锁成功。

前言

前面已经讲解了Zookeeper可重入锁的实现原理，自己对分布式锁也有了更深的认知。

我在公众号中发了一个疑问，相比于 Redis 来说，Zookeeper的实现方式要更好一些，即便Redis作者实现了RedLock算法来解决Redis集群模式下分布式锁的弊端，但Redis实现的分布式锁仍然不是那么完美。

比如有5台Redis集群，按照n/2 + 1代表获取锁成功，如果客户端A此时获取锁，Redis集群（1,2,3）返回成功，客户端A获取锁成功。

此时Redis 1 master宕机，切换到slave，而slave并未来得及同步客户端A加锁成功的信息到slave。

客户端B获取锁，Redis集群(1,4,5)返回成功，客户端B仍然可以成功获取锁。

即使如此，为何在实际生产项目中分布式锁大多还是由Redis来完成？

这一点我仍然有些疑惑，我接触过的公司和项目都普遍用Redis来实现分布式锁。

这里就不再纠结了，接着继续学习Zookeeper剩下几个实现分布式锁的组件吧。

Semaphore实现原理

前面已经讲过Redisson中Semaphore的实现原理( 【分布式锁】05-使用Redisson中Semaphore和CountDownLatch原理 )，现在学习下ZK中Semaphore是如何实现的

Semaphore 使用案例

使用示例很简单，Curator官网上有对应代码，使用 InterProcessSemaphoreV2 类即可，代码如下： 

/**
*  Zookeeper分布式锁测试代码
*
 * @author wangmeng
 * @date 2020/03/30 18:59
 */
public class Application {

    /** Zookeeper info */
    private static final String ZK_ADDRESS = "YourZkIP:2181";
    private static final String ZK_LOCK_PATH = "/locks/lock_01";
    private static final String ZK_SEMAPHORE_LOCK_PATH = "/semaphore/semaphore_01";

    public static void main(String[] args) throws InterruptedException {
        // 1.Connect to zk
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new RetryNTimes(10, 5000)
        );
        client.start();
        System.out.println("zk client start successfully!");

        Thread t1 = new Thread(() -> {
            testSemaphore(client);
        }, "t1");
        Thread t2 = new Thread(() -> {
            testSemaphore(client);
        }, "t2");
        Thread t3 = new Thread(() -> {
            testSemaphore(client);
        }, "t3");

        t1.start();
        t2.start();
        t3.start();
    }


    /**
     * 测试Semaphore
     */
    private static void testSemaphore(CuratorFramework client) {
        InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, ZK_SEMAPHORE_LOCK_PATH, 2);
        try {
            Lease lease = semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + " hold lock");
            Thread.sleep(5000L);
            semaphore.returnLease(lease);
            System.out.println(Thread.currentThread().getName() + " release lock");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

打印结果为：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png

因为设置的只允许最多2个客户端同时获取锁。

从效果上看t3和t2同时获取到了锁，接着t3释放了锁后t1才获取锁。

Semaphore加锁源码解析

源码面前出真知，我们直接看下源码：

public class InterProcessSemaphoreV2 {

    private static final String LOCK_PARENT = "locks";
    private static final String LEASE_PARENT = "leases";
    private static final String LEASE_BASE_NAME = "lease-";

    public Collection<Lease> acquire(int qty, long time, TimeUnit unit) throws Exception
    {
        long startMs = System.currentTimeMillis();
        boolean hasWait = (unit != null);
        long waitMs = hasWait ? TimeUnit.MILLISECONDS.convert(time, unit) : 0;

        Preconditions.checkArgument(qty > 0, "qty cannot be 0");

        ImmutableList.Builder<Lease> builder = ImmutableList.builder();
        boolean success = false;
        try
        {
            while ( qty-- > 0 )
            {
                int retryCount = 0;
                long startMillis = System.currentTimeMillis();
                boolean isDone = false;
                while ( !isDone )
                {
                    switch ( internalAcquire1Lease(builder, startMs, hasWait, waitMs) )
                    {
                        case CONTINUE:
                        {
                            isDone = true;
                            break;
                        }

                        // 省略其他分支逻辑
                    }
                }
            }
            success = true;
        }
        finally
        {
            if ( !success )
            {
                returnAll(builder.build());
            }
        }

        return builder.build();
    }

    private InternalAcquireResult internalAcquire1Lease(ImmutableList.Builder<Lease> builder, long startMs, boolean hasWait, long waitMs) throws Exception
    {
        if ( client.getState() != CuratorFrameworkState.STARTED )
        {
            return InternalAcquireResult.RETURN_NULL;
        }

        if ( hasWait )
        {
            long thisWaitMs = getThisWaitMs(startMs, waitMs);
            if ( !lock.acquire(thisWaitMs, TimeUnit.MILLISECONDS) )
            {
                return InternalAcquireResult.RETURN_NULL;
            }
        }
        else
        {
            lock.acquire();
        }
        try
        {
            PathAndBytesable<String> createBuilder = client.create().creatingParentsIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL);
            String path = (nodeData != null) ? createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME), nodeData) : createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME));
            String nodeName = ZKPaths.getNodeFromPath(path);
            builder.add(makeLease(path));

            synchronized(this)
            {
                for(;;)
                {
                    List<String> children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);
                    if ( !children.contains(nodeName) )
                    {
                        log.error("Sequential path not found: " + path);
                        return InternalAcquireResult.RETRY_DUE_TO_MISSING_NODE;
                    }

                    if ( children.size() <= maxLeases )
                    {
                        break;
                    }
                    if ( hasWait )
                    {
                        long thisWaitMs = getThisWaitMs(startMs, waitMs);
                        if ( thisWaitMs <= 0 )
                        {
                            return InternalAcquireResult.RETURN_NULL;
                        }
                        wait(thisWaitMs);
                    }
                    else
                    {
                        wait();
                    }
                }
            }
        }
        finally
        {
            lock.release();
        }
        return InternalAcquireResult.CONTINUE;
    }
}

代码有点长，我们一点点分析，我们以客户端A、B、C同时进入获取锁逻辑来举例，这里Semaphore最大可允许2个客户端同时获取锁。

  1. 三个客户端同时进入switch逻辑，执行 internalAcquire1Lease() 方法
  2. 在 internalAcquire1Lease() 方法中，先使用 lock.acquire() 执行加锁逻辑，这个lock是我们上一章讲的可重入锁逻辑，不再赘述
  3. 这个 lock 是哪里初始化的呢？在 InterProcessSemaphoreV2 构造函数中：
    lock = new InterProcessMutex(client, ZKPaths.makePath(path, LOCK_PARENT));
    this.maxLeases = (count != null) ? count.getCount() : maxLeases;
  4. 注意lock的path为： /semaphore/semaphore_01/locks , maxLeases为传入的3
  5. 此时客户端A、B、C执行 lock.acquire() 只会有一个可以成功获取锁，其他两个客户端会 wait()

到了这里，Zookeeper中就会有三条类似于：

/semaphores/semaphore_01/locks/_c_a9302e20-de9c-4356-923a-274664d7676c-lock-0000000001 的数据

接着客户端A继续往下执行，具体逻辑如图：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理

  1. 首先是客户端A创建一个/locks/lock-xxxx01节点，获取锁成功过
  2. 接着创建临时顺序节点/leases/lease-xxxx01
  3. 判断/leases目录下节点数量（数量为1）是否小于等于maxLeases（maxLeases=2）
  4. 如果成功则退出循环，释放/locks加的锁，返回 InternalAcquireResult.CONTINUE ，状态，执行 lock.release() 通知客户端B、C争抢 /locks 节点下的锁
  5. 此时如果客户端B抢到锁，然后同样创建/leases/lease-xxxx02,
  6. 判断/leases目录下节点数量（数量为2）是否小于等于maxLeases（maxLeases=2）
  7. 客户端B也退出循环，返回 InternalAcquireResult.CONTINUE ，接着客户端C来获取锁
  8. 客户端C执行时，判断/leases目录下节点数量（数量为3）是否小于等于maxLeases（maxLeases=2）

此时客户端C会进入到 wait() 方法，直到客户端A或者客户端B释放 leases 节点下锁时才会重试获取锁。

返回 InternalAcquireResult.CONTINUE 后，就标志获取锁成功。

Semaphore释放锁源码分析

我们直接看代码，释放锁代码很简单：

/**
 * Convenience method. Closes the lease
 *
 * @param lease lease to close
 */
public void returnLease(Lease lease)
{
    Closeables.closeQuietly(lease);
}

一路跟下去，可以看到 closeQuietly 实现方法：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png 【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理

最后用到 Lease 中的 close() 方法，删除创建的 /leases/lease-xxxx 节点数据，然后通知其他节点客户端，使用 notifyAll()

ZK-Semaphore总结

一张图总结下：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 05_Zookeeper中Semaphore实现原理 _1_.jpg

Zookeeper 非重入锁实现原理

之前听小伙伴说过一个面试题，请说出你所知道的非重入锁？

在脑子中搜索JDK中非重入锁？好像没有？

Zookeeper中提供了一个非重入锁的实现方式，实现原理使用Semaphore，最大允许1个客户端获取锁

按理说JDK中的Semaphore也可以实现此功能，哈哈，感觉自己被忽悠了，接着还是勉为其难的看下ZK中"非重入锁"的实现方式吧：

使用示例

/**
 * 测试非重入锁
 */
private static void testSemaphoreMutex(CuratorFramework client) {
    InterProcessSemaphoreMutex semaphoreMutex = new InterProcessSemaphoreMutex(client, ZK_SEMAPHORE_LOCK_PATH);
    try {
        semaphoreMutex.acquire();
        Thread.sleep(5000L);
        semaphoreMutex.release();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

源码分析

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png

实际上就是设置 maxLeases 为1，原理同上面的 Semaphore 源码分析

Zookeeper读写锁原理

之前在Redisson中已经见过它对读写锁的实现，分别举例了读读、写写、读写、写读这几种场景锁的互斥性以及可重入性，这里也采用类似的场景分析。

读写锁使用案例

直接看案例，可以针对案例修改几种场景进行测试：

/**
 * @author wangmeng
 * @date 2020/03/30 18:59
 */
public class Application {

    /** Zookeeper info */
    private static final String ZK_ADDRESS = "yourZkIP:2181";
    private static final String ZK_LOCK_PATH = "/locks/lock_01";
    private static final String ZK_SEMAPHORE_LOCK_PATH = "/semaphore/semaphore_01";
    private static final String ZK_READ_WRITE_LOCK_PATH = "/readwrite/readwrite_01";

    public static void main(String[] args) throws InterruptedException {
        // 1.Connect to zk
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new RetryNTimes(10, 5000)
        );
        client.start();
        System.out.println("zk client start successfully!");

        Thread t1 = new Thread(() -> {
            testReadWriteLock(client);
        }, "t1");
        Thread t2 = new Thread(() -> {
            testReadWriteLock(client);
        }, "t2");

        t1.start();
        t2.start();
    }


    /**
     * 测试读写锁
     */
    private static void testReadWriteLock(CuratorFramework client) {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, ZK_READ_WRITE_LOCK_PATH);
        try {
            // 获取读锁
            InterProcessMutex readLock = readWriteLock.readLock();
            readLock.acquire();
            System.out.println(Thread.currentThread().getName() + " hold read lock");
            Thread.sleep(5000);
            readLock.release();
            System.out.println(Thread.currentThread().getName() + " release read lock");

            // 获取写锁
            InterProcessMutex writeLock = readWriteLock.writeLock();
            writeLock.acquire();
            System.out.println(Thread.currentThread().getName() + " hold write lock");
            Thread.sleep(5000);
            writeLock.release();
            System.out.println(Thread.currentThread().getName() + " release write lock");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

运行后结果：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png

从结果可以看出来： 读读不互斥、 写写互斥

读写锁源码解析

首先看下 InterProcessReadWrite 的构造函数： 

public class InterProcessReadWriteLock {    
    public InterProcessReadWriteLock(CuratorFramework client, String basePath)
    {
        writeMutex = new InternalInterProcessMutex
        (
            client,
            basePath,
            WRITE_LOCK_NAME,
            1,
            new SortingLockInternalsDriver()
            {
                @Override
                public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                {
                    return super.getsTheLock(client, children, sequenceNodeName, maxLeases);
                }
            }
        );

        readMutex = new InternalInterProcessMutex
        (
            client,
            basePath,
            READ_LOCK_NAME,
            Integer.MAX_VALUE,
            new SortingLockInternalsDriver()
            {
                @Override
                public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
                {
                    return readLockPredicate(children, sequenceNodeName);
                }
            }
        );
    }
}

因为ZK中的读写锁底层也是基于第一讲中 InterProcessMutex.internalLock() 去实现的，所以 InterProcessReadWriteLock 读锁和写锁分别初始化了 maxLeases 及重写了 getsTheLock() 方法，这个方法是判断是否可以获取锁的核心代码，类似于： 

int ourIndex = children.indexOf(sequenceNodeName);
boolean getsTheLock = ourIndex < maxLeases;

不清楚的可以回头看看： 【分布式锁】06-Zookeeper实现分布式锁：可重入锁源码分析

另外写锁和读锁的path会有区别：

private static final String READ_LOCK_NAME  = "__READ__";
private static final String WRITE_LOCK_NAME = "__WRIT__";

写锁的maxLeases是1，加了写锁就不允许再加其他读锁（但可重入加写锁和读锁）

读锁的maxLeases是Integer.MAX_VALUE，读读锁不互斥

读读互斥及重入

查看读锁中判断获取锁成功的核心逻辑：

private PredicateResults readLockPredicate(List<String> children, String sequenceNodeName) throws Exception
{
    // 如果当前线程获取写锁，那么直接返回true，获取读锁成功
    if ( writeMutex.isOwnedByCurrentThread() )
    {
        return new PredicateResults(null, true);
    }

    int         index = 0;
    int         firstWriteIndex = Integer.MAX_VALUE;
    int         ourIndex = Integer.MAX_VALUE;
    for ( String node : children )
    {
        if ( node.contains(WRITE_LOCK_NAME) )
        {
            firstWriteIndex = Math.min(index, firstWriteIndex);
        }
        else if ( node.startsWith(sequenceNodeName) )
        {
            ourIndex = index;
            break;
        }

        ++index;
    }
    StandardLockInternalsDriver.validateOurIndex(sequenceNodeName, ourIndex);

    boolean     getsTheLock = (ourIndex < firstWriteIndex);
    String      pathToWatch = getsTheLock ? null : children.get(firstWriteIndex);
    return new PredicateResults(pathToWatch, getsTheLock);
}

如果客户端A已经获取了读锁

此时客户端B再来获取读锁

  1. children：[xxx_READ_0001, xxxx_READ_0002]，此时都是读锁，不包含WRITE锁标识
  2. sequenceNodeName就是node创建的节点名称，这里ourIndex=index0
  3. ourIndex<firstWriteIndex = Integer.MAX_VALUE 获取锁成功

执行debug流程如下图：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png

因为读读不互斥，所以这里读锁也是可重入的

写读互斥及重入

上面已经分析过读读的逻辑了，这里接着按照上面的代码分析下读写的逻辑：

客户端A加写锁成功

客户端B加读锁

  1. node.contains(WRITE_LOCK_NAME) ，此时客户端B中含有WRITE标识
  2. firstWriteIndex = Math.min(index, firstWriteIndex)=0
  3. boolean getsTheLock = (ourIndex < firstWriteIndex);
    此时ourIndex = Integer.MAX_VALUE，判断条件不成立，所以加写锁失败

不同客户端写读锁互斥

接着看看同一个客户端逻辑：

if ( writeMutex.isOwnedByCurrentThread() )
{
    return new PredicateResults(null, true);
}

如果当前线程获取了写锁，那么再加读写直接返回成功。

所以同一个客户端同一线程：先加写锁、再加读锁可重入，这一点和Redisson中是一致的，具体可以看： 【分布式锁】04-使用Redisson实现ReadWriteLock原理

写写互斥及重入

写锁完全可以看做成 InterProcessMutex ，这里 maxLeases 为1，所以同一个线程写是可重入的，不同客户端获取锁时互斥的

读写互斥及重入

客户端A加读锁

客户端B加写锁

同样道理，此时children数据结构如：

[_c_13bf63d6-43f3-4c2f-ba98-07a641d351f2-__READ__0000000004,

_c_73b60882-9361-4fb7-8420-a8d4911d2c99-__WRIT__0000000005]

判断写锁在"/readwrite/readwrite_01"目录下的位置，不是在首位，加锁失败

可重入锁也是同样原理，不可重入

Zookeeper中MultiLock实现原理

我们在Redisson中已经见过MultiLock原理，其中Redissoon为了实现 RedLock 算法，也有MultiLock的实现（可以参考 【分布式锁】03-使用Redisson实现RedLock原理 ）当多个资源需要统一加锁的时候，我们就可以使用MultiLock

Zookeeper中的MultiLock实现非常简单，就是依次加锁，实现如下图：

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理 image.png

总结

Zookeeper实现分布式锁的相关原理全都讲完了，仔细阅读Curator源码觉得还挺有意思，再来会先Curator官网那句话：

Guava is to Java what Curator is to Zookeeper

Curator真的很强，分布式锁实现的很棒！

申明

本文章首发自本人公众号： 壹枝花算不算浪漫 ，如若转载请标明来源!

感兴趣的小伙伴可关注个人公众号：壹枝花算不算浪漫

【分布式锁】07-Zookeeper实现分布式锁：Semaphore、读写锁实现原理


以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网

查看所有标签

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。

码农书籍
Computational Geometry

Computational Geometry

Mark de Berg、Otfried Cheong、Marc van Kreveld、Mark Overmars / Springer / 2008-4-16 / USD 49.95

This well-accepted introduction to computational geometry is a textbook for high-level undergraduate and low-level graduate courses. The focus is on algorithms and hence the book is well suited for st......一起来看看 《Computational Geometry》 这本书的介绍吧!

码农工具
XML 在线格式化
XML 在线格式化

在线 XML 格式化压缩工具

Markdown 在线编辑器
Markdown 在线编辑器

Markdown 在线编辑器

HSV CMYK 转换工具
HSV CMYK 转换工具

HSV CMYK互换工具

  • New
  • 文章
  • 话题
  • 教程
    • 关注 码农网 公众号
    版权所有，保留一切权利！© 2018-2026 码农网 粤ICP备17054400号-3