一文带你理解Zookeeper实现分布式锁的机制

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 一文带你理解Zookeeper实现分布式锁的机制

概述


提到锁,想必大家可能最先想到的是Java JUC中的synchronized关键字或者可重入锁ReentrantLock。它能够保证我们的代码在同一个时刻只有一个线程执行,保证数据的一致性和完整性。但是它仅限于单体项目,也就是说它们只能保证单个JVM应用内线程的顺序执行。

如果你部署了多个节点,也就是分布式场景下如何保证不同节点在同一时刻只有一个线程执行呢?场景的业务场景比如秒杀、抢优惠券等,这就引入了我们的分布式锁,本文我们主要讲解利用Zookeeper的特性如何来实现我们的分布式锁。


Zookeeper分布式锁实现原理


利用Zookeeper的临时顺序节点和监听机制两大特性,可以帮助我们实现分布式锁。

1671157876838.jpg

  1. 首先得有一个持久节点/locks, 路径服务于某个使用场景,如果有多个使用场景建议路径不同。
  2. 请求进来时首先在/locks创建临时有序节点,所有会看到在/locks下面有seq-000000000, seq-00000001 等等节点。
  3. 然后判断当前创建得节点是不是/locks路径下面最小的节点,如果是,获取锁,不是,阻塞线程,同时设置监听器,监听前一个节点。
  4. 获取到锁以后,开始处理业务逻辑,最后delete当前节点,表示释放锁。
  5. 后一个节点就会收到通知,唤起线程,重复上面的判断。

大家有没有想过为什么要设置对前一个节点的监听?

主要为了避免羊群效应。所谓羊群效应就是一个节点挂掉,所有节点都去监听,然后做出反应,这样会给服务器带来巨大压力,所以有了临时顺序节点,当一个节点挂掉,只有它后面的那一个节点才做出反应。


原生Zookeeper客户端实现分布式锁


通过原生zookeeper api方式的实现,可以加强我们对zk实现分布式锁原理的理解。

public class DistributedLock {
    private String connectString = "10.100.1.176:2281";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;
    private String rootNode = "lock";
    private String subNode = "seq-";
    private String waitPath;
    // 当前client创建的子节点
    private String currentNode;
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    private CountDownLatch waitDownLatch = new CountDownLatch(1);
    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 如果连接建立时,唤醒 wait 在该 latch 上的线程
                if(event.getState() == Event.KeeperState.SyncConnected) {
                    countDownLatch.countDown();
                }
                //  发生了 waitPath 的删除事件
                if(event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                    waitDownLatch.countDown();
                }
            }
        });
        // 等待连接建立,因为连接建立时异步过程
        countDownLatch.await();
        // 获取根节点
        Stat stat = zk.exists("/" + rootNode, false);
        // 如果根节点不存在,则创建根节点
        if(stat == null) {
            System.out.println("创建根节点");
            zk.create("/" + rootNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
    public void zkLock() {
        try {
            // 在根节点创建临时顺序节点
            currentNode = zk.create("/" + rootNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 获取子节点
            List<String> childrenNodes = zk.getChildren("/" + rootNode, false);
            // 如果只有一个子节点,说明是当前节点,直接获得锁
            if(childrenNodes.size() == 1) {
                return;
            } else {
                //对根节点下的所有临时顺序节点进行从小到大排序
                Collections.sort(childrenNodes);
                //当前节点名称
                String thisNode = currentNode.substring(("/" + rootNode + "/").length());
                //获取当前节点的位置
                int index = childrenNodes.indexOf(thisNode);
                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    // index == 0, 说明 thisNode 在列表中最小, 当前client 获得锁
                    return;
                } else {
                    // 获得排名比 currentNode 前 1 位的节点
                    this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);
                    // 在 waitPath节点上注册监听器, 当 waitPath 被删除时,zookeeper 会回调监听器的 process 方法
                    zk.getData(waitPath, true, new Stat());
                    //进入等待锁状态
                    waitDownLatch.await();
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void zkUnlock() {
        try {
            zk.delete(this.currentNode, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}

测试代码如下:

public class DistributedLockTest {
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        DistributedLock lock1 = new DistributedLock();
        DistributedLock lock2 = new DistributedLock();
        new Thread(() -> {
            // 获取锁对象
            try {
                lock1.zkLock();
                System.out.println("线程 1 获取锁");
                Thread.sleep(5 * 1000);
                System.out.println("线程 1 释放锁");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock1.zkUnlock();
            }
        }).start();
        new Thread(() -> {
            // 获取锁对象
            try {
                lock2.zkLock();
                System.out.println("线程 2 获取锁");
                Thread.sleep(5 * 1000);
                System.out.println("线程 2 释放锁");
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock2.zkUnlock();
            }
        }).start();
    }
}

测试结果:

线程 2 获取锁
线程 2 释放锁
线程 1 获取锁
线程 1 释放锁

获取锁和释放锁成对出现,说明分布式锁生效了。


Curator框架实现分布式锁


在实际的开发钟,我们会直接使用成熟的框架Curator客户端,它里面封装了分布式锁的实现,避免我们去重复造轮子。

  1. pom.xml添加如下依赖
<dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>5.2.1</version>
        </dependency>
  1. 通过InterProcessLock实现分布式锁
public class CuratorLockTest {
    private String connectString = "10.100.1.14:2181";
    private String rootNode = "/locks";
    public static void main(String[] args) {
        new CuratorLockTest().testLock();
    }
    public void testLock() {
        // 分布式锁1
        InterProcessLock lock1 = new InterProcessMutex(getCuratorFramework(), rootNode);
        // 分布式锁2
        InterProcessLock lock2 = new InterProcessMutex(getCuratorFramework(), rootNode);
        // 第一个线程
        new Thread(() -> {
            // 获取锁对象
            try {
                lock1.acquire();
                System.out.println("线程 1 获取锁");
                // 测试锁重入
                lock1.acquire();
                System.out.println("线程 1 再次获取锁");
                Thread.sleep(5 * 1000);
                lock1.release();
                System.out.println("线程 1 释放锁");
                lock1.release();
                System.out.println("线程 1 再次释放锁");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        // 第二个线程
        new Thread(() -> {
            // 获取锁对象
            try {
                lock2.acquire();
                System.out.println("线程 2 获取锁");
                // 测试锁重入
                lock2.acquire();
                System.out.println("线程 2 再次获取锁");
                Thread.sleep(5 * 1000);
                lock2.release();
                System.out.println("线程 2 释放锁");
                lock2.release();
                System.out.println("线程 2 再次释放锁");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }
    public CuratorFramework getCuratorFramework() {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString).connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(new ExponentialBackoffRetry(3000, 3)).build();
        // 连接
        client.start();
        System.out.println("zookeeper 初始化完成...");
        return client;
    }
}
  1. 结果展示
线程 1 释放锁
线程 1 再次释放锁
线程 2 获取锁
线程 2 再次获取锁
线程 2 释放锁
线程 2 再次释放锁

有兴趣的看下源码,它是通过wait、notify来实现阻塞。

代码github.com/alvinlkk/aw…


总结


ZooKeeper分布式锁(如InterProcessMutex),能有效的解决分布式锁问题,但是性能并不高。

因为每次在创建锁和释放锁的过程中,都要动态创建、销毁瞬时节点来实现锁功能。大家知道,ZK中创建和删除节点只能通过Leader服务器来执行,然后Leader服务器还需要将数据同不到所有的Follower机器上,这样频繁的网络通信,性能的短板是非常突出的。

在高性能,高并发的场景下,不建议使用ZooKeeper的分布式锁,可以使用Redis的分布式锁。而由于ZooKeeper的高可用特性,所以在并发量不是太高的场景,推荐使用ZooKeeper的分布式锁。

相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
相关文章
|
14天前
|
监控 负载均衡 Cloud Native
ZooKeeper分布式协调服务详解:面试经验与必备知识点解析
【4月更文挑战第9天】本文深入剖析ZooKeeper分布式协调服务原理,涵盖核心概念如Server、Client、ZNode、ACL、Watcher,以及ZAB协议在一致性、会话管理、Leader选举中的作用。讨论ZooKeeper数据模型、操作、会话管理、集群部署与管理、性能调优和监控。同时,文章探讨了ZooKeeper在分布式锁、队列、服务注册与发现等场景的应用,并在面试方面分析了与其它服务的区别、实战挑战及解决方案。附带Java客户端实现分布式锁的代码示例,助力提升面试表现。
30 2
|
1月前
|
数据库 数据安全/隐私保护
TiDB分布式事务处理机制
【2月更文挑战第28天】TiDB作为开源的分布式HTAP数据库产品,其分布式事务处理机制是其核心功能之一。本章节将深入解析TiDB分布式事务处理机制的工作原理,包括其采用的分布式事务协议、事务的提交与回滚过程、以及如何处理并发事务等关键内容。通过了解TiDB的分布式事务处理机制,我们可以更好地理解其在分布式环境下如何确保数据一致性和事务正确性。
|
1月前
|
SQL 弹性计算 分布式计算
TiDB计算层详解:分布式计算框架与查询优化机制
【2月更文挑战第26天】本文将深入剖析TiDB的计算层,详细解析其分布式计算框架和查询优化机制。通过了解计算层的核心组件和工作原理,我们可以更好地理解TiDB如何高效处理SQL查询和计算任务。本文将从计算层的架构、任务分发、查询优化等方面展开介绍,帮助读者全面掌握TiDB计算层的关键技术和优势。
|
3月前
|
消息中间件 Java 网络安全
JAVAEE分布式技术之Zookeeper的第一次课
JAVAEE分布式技术之Zookeeper的第一次课
70 0
|
1月前
|
监控 NoSQL Java
Zookeeper分布式锁
Zookeeper分布式锁
90 1
|
28天前
|
存储 监控 安全
金石推荐 | 【分布式技术专题】「单点登录技术架构」一文带领你好好认识以下Saml协议的运作机制和流程模式
金石推荐 | 【分布式技术专题】「单点登录技术架构」一文带领你好好认识以下Saml协议的运作机制和流程模式
63 1
|
1月前
|
Dubbo 网络协议 应用服务中间件
分布式微服务框架dubbo原理与机制
分布式微服务框架dubbo原理与机制
|
2月前
|
Java Linux Spring
Zookeeper实现分布式服务配置中心
Zookeeper实现分布式服务配置中心
48 0
|
2月前
|
存储 分布式计算 Hadoop
ZooKeeper初探:分布式世界的守护者
ZooKeeper初探:分布式世界的守护者
64 0
|
2月前
|
NoSQL Java API
分布式锁【数据库乐观锁实现的分布式锁、Zookeeper分布式锁原理、Redis实现的分布式锁】(三)-全面详解(学习总结---从入门到深化)
分布式锁【数据库乐观锁实现的分布式锁、Zookeeper分布式锁原理、Redis实现的分布式锁】(三)-全面详解(学习总结---从入门到深化)
298 0