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

本文涉及的产品
云原生网关 MSE Higress,422元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
注册配置 MSE Nacos/ZooKeeper,118元/月
简介: 一文带你理解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实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
相关文章
|
2月前
|
存储 缓存 算法
分布式锁服务深度解析:以Apache Flink的Checkpointing机制为例
【10月更文挑战第7天】在分布式系统中,多个进程或节点可能需要同时访问和操作共享资源。为了确保数据的一致性和系统的稳定性,我们需要一种机制来协调这些进程或节点的访问,避免并发冲突和竞态条件。分布式锁服务正是为此而生的一种解决方案。它通过在网络环境中实现锁机制,确保同一时间只有一个进程或节点能够访问和操作共享资源。
82 3
|
3月前
|
安全 应用服务中间件 API
微服务分布式系统架构之zookeeper与dubbo-2
微服务分布式系统架构之zookeeper与dubbo-2
|
3月前
|
负载均衡 Java 应用服务中间件
微服务分布式系统架构之zookeeper与dubbor-1
微服务分布式系统架构之zookeeper与dubbor-1
|
14天前
|
存储 运维 NoSQL
分布式读写锁的奥义:上古世代 ZooKeeper 的进击
本文作者将介绍女娲对社区 ZooKeeper 在分布式读写锁实践细节上的思考,希望帮助大家理解分布式读写锁背后的原理。
|
2月前
|
消息中间件 存储 监控
消息队列系统中的确认机制在分布式系统中如何实现
消息队列系统中的确认机制在分布式系统中如何实现
|
2月前
|
消息中间件 存储 监控
【10月更文挑战第2天】消息队列系统中的确认机制在分布式系统中如何实现
【10月更文挑战第2天】消息队列系统中的确认机制在分布式系统中如何实现
|
2月前
|
分布式计算 NoSQL Java
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
48 2
|
2月前
|
分布式计算 监控 Hadoop
Hadoop-29 ZooKeeper集群 Watcher机制 工作原理 与 ZK基本命令 测试集群效果 3台公网云服务器
Hadoop-29 ZooKeeper集群 Watcher机制 工作原理 与 ZK基本命令 测试集群效果 3台公网云服务器
43 1
|
2月前
|
分布式计算 Hadoop Unix
Hadoop-28 ZooKeeper集群 ZNode简介概念和测试 数据结构与监听机制 持久性节点 持久顺序节点 事务ID Watcher机制
Hadoop-28 ZooKeeper集群 ZNode简介概念和测试 数据结构与监听机制 持久性节点 持久顺序节点 事务ID Watcher机制
45 1
|
2月前
|
分布式计算 Hadoop
Hadoop-27 ZooKeeper集群 集群配置启动 3台云服务器 myid集群 zoo.cfg多节点配置 分布式协调框架 Leader Follower Observer
Hadoop-27 ZooKeeper集群 集群配置启动 3台云服务器 myid集群 zoo.cfg多节点配置 分布式协调框架 Leader Follower Observer
50 1