备案控制台
开发者社区 开发与运维 文章 正文

【分布式】Zookeeper使用--开源客户端（二）

2022-05-02 146
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
简介： 上一篇博客已经介绍了如何使用Zookeeper提供的原生态Java API进行操作，本篇博文主要讲解如何通过开源客户端来进行操作。

　3.8 节点监听　　


package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
public class NodeCache_Sample {
    static String path = "/zk-book/nodecache";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, "init".getBytes());
        final NodeCache cache = new NodeCache(client, path, false);
        cache.start(true);
        cache.getListenable().addListener(new NodeCacheListener() {
            public void nodeChanged() throws Exception {
                System.out.println("Node data update, new data: " + new String(cache.getCurrentData().getData()));
            }
        });
        client.setData().forPath(path, "u".getBytes());
        Thread.sleep(1000);
        client.delete().deletingChildrenIfNeeded().forPath(path);
        Thread.sleep(Integer.MAX_VALUE);
    }
}

　运行结果：　　


Node data update, new data: u

　　当节点数据变更后收到了通知。NodeCache不仅可以监听数据节点的内容变更，也能监听指定节点是否存在。

　　

3.9 子节点监听　

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
public class PathChildrenCache_Sample {
    static String path = "/zk-book";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).sessionTimeoutMs(5000).build();
    public static void main(String[] args) throws Exception {
        client.start();
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                case CHILD_ADDED:
                    System.out.println("CHILD_ADDED," + event.getData().getPath());
                    break;
                case CHILD_UPDATED:
                    System.out.println("CHILD_UPDATED," + event.getData().getPath());
                    break;
                case CHILD_REMOVED:
                    System.out.println("CHILD_REMOVED," + event.getData().getPath());
                    break;
                default:
                    break;
                }
            }
        });
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        Thread.sleep(1000);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        Thread.sleep(1000);
        client.delete().forPath(path + "/c1");
        Thread.sleep(1000);
        client.delete().forPath(path);
        Thread.sleep(Integer.MAX_VALUE);
    }
}

　运行结果：

CHILD_ADDED,/zk-book/c1
CHILD_REMOVED,/zk-book/c1

　　监听节点的子节点，包括新增、数据变化、删除三类事件。


3.10 Master选举

　　借助Zookeeper，开发者可以很方便地实现Master选举功能，其大体思路如下：选择一个根节点，如/master_select，多台机器同时向该节点创建一个子节点/master_select/lock，利用Zookeeper特性，最终只有一台机器能够成功创建，成功的那台机器就是Master。

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class Recipes_MasterSelect {
    static String master_path = "/curator_recipes_master_path";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        LeaderSelector selector = new LeaderSelector(client, master_path, new LeaderSelectorListenerAdapter() {
            public void takeLeadership(CuratorFramework client) throws Exception {
                System.out.println("成为Master角色");
                Thread.sleep(3000);
                System.out.println("完成Master操作，释放Master权利");
            }
        });
        selector.autoRequeue();
        selector.start();
        Thread.sleep(Integer.MAX_VALUE);
    }
}

运行结果：

成为Master角色
完成Master操作，释放Master权利
成为Master角色

　　以上结果会反复循环，并且当一个应用程序完成Master逻辑后，另外一个应用程序的相应方法才会被调用，即当一个应用实例成为Master后，其他应用实例会进入等待，直到当前Master挂了或者推出后才会开始选举Master。


3.11 分布式锁

　　为了保证数据的一致性，经常在程序的某个运行点需要进行同步控制。以流水号生成场景为例，普通的后台应用通常采用时间戳方式来生成流水号，但是在用户量非常大的情况下，可能会出现并发问题。　

package com.hust.grid.leesf.curator.examples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
public class Recipes_NoLock {
    public static void main(String[] args) throws Exception {
        final CountDownLatch down = new CountDownLatch(1);
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                        down.await();
                    } catch (Exception e) {
                    }
                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
                    String orderNo = sdf.format(new Date());
                    System.err.println("生成的订单号是 : " + orderNo);
                }
            }).start();
        }
        down.countDown();
    }
}

　运行结果：　

生成的订单号是 : 16:29:10|590
生成的订单号是 : 16:29:10|590
生成的订单号是 : 16:29:10|591
生成的订单号是 : 16:29:10|591
生成的订单号是 : 16:29:10|590
生成的订单号是 : 16:29:10|590
生成的订单号是 : 16:29:10|591
生成的订单号是 : 16:29:10|590
生成的订单号是 : 16:29:10|592
生成的订单号是 : 16:29:10|591

　结果表示订单号出现了重复，即普通的方法无法满足业务需要，因为其未进行正确的同步。可以使用Curator来实现分布式锁功能。

package com.hust.grid.leesf.curator.examples;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class Recipes_Lock {
    static String lock_path = "/curator_recipes_lock_path";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        final InterProcessMutex lock = new InterProcessMutex(client, lock_path);
        final CountDownLatch down = new CountDownLatch(1);
        for (int i = 0; i < 30; i++) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                        down.await();
                        lock.acquire();
                    } catch (Exception e) {
                    }
                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
                    String orderNo = sdf.format(new Date());
                    System.out.println("生成的订单号是 : " + orderNo);
                    try {
                        lock.release();
                    } catch (Exception e) {
                    }
                }
            }).start();
        }
        down.countDown();
    }
}

　运行结果：


生成的订单号是 : 16:31:50|293
生成的订单号是 : 16:31:50|319
生成的订单号是 : 16:31:51|278
生成的订单号是 : 16:31:51|326
生成的订单号是 : 16:31:51|402
生成的订单号是 : 16:31:51|420
生成的订单号是 : 16:31:51|546
生成的订单号是 : 16:31:51|602
生成的订单号是 : 16:31:51|626
生成的订单号是 : 16:31:51|656
生成的订单号是 : 16:31:51|675
生成的订单号是 : 16:31:51|701
生成的订单号是 : 16:31:51|708
生成的订单号是 : 16:31:51|732
生成的订单号是 : 16:31:51|763
生成的订单号是 : 16:31:51|785
生成的订单号是 : 16:31:51|805
生成的订单号是 : 16:31:51|823
生成的订单号是 : 16:31:51|839
生成的订单号是 : 16:31:51|853
生成的订单号是 : 16:31:51|868
生成的订单号是 : 16:31:51|884
生成的订单号是 : 16:31:51|897
生成的订单号是 : 16:31:51|910
生成的订单号是 : 16:31:51|926
生成的订单号是 : 16:31:51|939
生成的订单号是 : 16:31:51|951
生成的订单号是 : 16:31:51|965
生成的订单号是 : 16:31:51|972
生成的订单号是 : 16:31:51|983

　结果表明此时已经不存在重复的流水号。


3.12 分布式计数器

　　分布式计数器的典型应用是统计系统的在线人数，借助Zookeeper也可以很方便实现分布式计数器功能：指定一个Zookeeper数据节点作为计数器，多个应用实例在分布式锁的控制下，通过更新节点的内容来实现计数功能。　

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
public class Recipes_DistAtomicInt {
    static String distatomicint_path = "/curator_recipes_distatomicint_path";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(client, distatomicint_path,
                new RetryNTimes(3, 1000));
        AtomicValue<Integer> rc = atomicInteger.add(8);
        System.out.println("Result: " + rc.succeeded());
    }
}

运行结果：

Result: true

　　结果表明已经将数据成功写入数据节点中。


3.13 分布式Barrier

　　如同JDK的CyclicBarrier，Curator提供了DistributedBarrier来实现分布式Barrier。

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class Recipes_Barrier {
    static String barrier_path = "/curator_recipes_barrier_path";
    static DistributedBarrier barrier;
    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                public void run() {
                    try {
                        CuratorFramework client = CuratorFrameworkFactory.builder()
                                .connectString("127.0.0.1:2181")
                                .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
                        client.start();
                        barrier = new DistributedBarrier(client, barrier_path);
                        System.out.println(Thread.currentThread().getName() + "号barrier设置");
                        barrier.setBarrier();
                        barrier.waitOnBarrier();
                        System.err.println("启动...");
                    } catch (Exception e) {
                    }
                }
            }).start();
        }
        Thread.sleep(2000);
        barrier.removeBarrier();
    }
}

　　运行结果：

Thread-1号barrier设置
Thread-2号barrier设置
Thread-4号barrier设置
Thread-3号barrier设置
Thread-0号barrier设置
启动...
启动...
启动...
启动...
启动...

结果表明通过DistributedBarrier可以实现类似于CyclicBarrier的分布式Barrier功能。


四、Curator工具类

　　4.1 ZKPaths

　　其提供了简单的API来构建znode路径、递归创建、删除节点等。 

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.curator.utils.ZKPaths.PathAndNode;
import org.apache.zookeeper.ZooKeeper;
public class ZKPaths_Sample {
    static String path = "/curator_zkpath_sample";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        ZooKeeper zookeeper = client.getZookeeperClient().getZooKeeper();
        System.out.println(ZKPaths.fixForNamespace(path, "sub"));
        System.out.println(ZKPaths.makePath(path, "sub"));
        System.out.println(ZKPaths.getNodeFromPath("/curator_zkpath_sample/sub1"));
        PathAndNode pn = ZKPaths.getPathAndNode("/curator_zkpath_sample/sub1");
        System.out.println(pn.getPath());
        System.out.println(pn.getNode());
        String dir1 = path + "/child1";
        String dir2 = path + "/child2";
        ZKPaths.mkdirs(zookeeper, dir1);
        ZKPaths.mkdirs(zookeeper, dir2);
        System.out.println(ZKPaths.getSortedChildren(zookeeper, path));
        ZKPaths.deleteChildren(client.getZookeeperClient().getZooKeeper(), path, true);
    }
}

　　运行结果：　

/curator_zkpath_sample/sub
/curator_zkpath_sample/sub
sub1
/curator_zkpath_sample
sub1
[child1, child2]

　借助ZKPaths可快速方便的完成节点的创建等操作。


4.2 EnsurePath

　　其提供了一种能够确保数据节点存在的机制，当上层业务希望对一个数据节点进行操作时，操作前需要确保该节点存在。　

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.EnsurePath;
public class EnsurePathDemo {
    static String path = "/zk-book/c1";
    static CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
            .sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
    public static void main(String[] args) throws Exception {
        client.start();
        client.usingNamespace("zk-book");
        EnsurePath ensurePath = new EnsurePath(path);
        ensurePath.ensure(client.getZookeeperClient());
        ensurePath.ensure(client.getZookeeperClient());
        EnsurePath ensurePath2 = client.newNamespaceAwareEnsurePath("/c1");
        ensurePath2.ensure(client.getZookeeperClient());
    }
}

EnsurePath采取了如下节点创建方式，试图创建指定节点，如果节点已经存在，那么就不进行任何操作，也不对外抛出异常，否则正常创建数据节点。


五、总结

　　本篇介绍了使用Zookeeper的开源客户端如何操作Zookeeper的方法，相应的源码也已经上传至github，谢谢各位园友的观看～


文章标签：
微服务引擎
Java
API
开发者
关键词：
分布式zookeeper
分布式开源
微服务引擎客户端
分布式微服务引擎
微服务引擎开源
相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作，您将了解并实现在线业务的微服务全链路灰度能力。
xleesf
目录
相关文章
长梦
|
1月前
|
监控 负载均衡 Cloud Native
ZooKeeper分布式协调服务详解：面试经验与必备知识点解析
【4月更文挑战第9天】本文深入剖析ZooKeeper分布式协调服务原理，涵盖核心概念如Server、Client、ZNode、ACL、Watcher，以及ZAB协议在一致性、会话管理、Leader选举中的作用。讨论ZooKeeper数据模型、操作、会话管理、集群部署与管理、性能调优和监控。同时，文章探讨了ZooKeeper在分布式锁、队列、服务注册与发现等场景的应用，并在面试方面分析了与其它服务的区别、实战挑战及解决方案。附带Java客户端实现分布式锁的代码示例，助力提升面试表现。
长梦
124 2
Cool架构
|
2月前
|
监控 Dubbo 前端开发
快速入门分布式系统与Dubbo+zookeeper Demo
快速入门分布式系统与Dubbo+zookeeper Demo
Cool架构
42 0
hsfxuebao
|
2月前
|
监控 NoSQL Java
Zookeeper分布式锁
Zookeeper分布式锁
hsfxuebao
90 1
1941623231718325
|
5天前
|
存储 大数据 Apache
深入理解ZooKeeper：分布式协调服务的核心与实践
【5月更文挑战第7天】ZooKeeper是Apache的分布式协调服务，确保大规模分布式系统中的数据一致性与高可用性。其特点包括强一致性、高可用性、可靠性、顺序性和实时性。使用ZooKeeper涉及安装配置、启动服务、客户端连接及执行操作。实际应用中，面临性能瓶颈、不可伸缩性和单点故障等问题，可通过水平扩展、集成其他服务和多集群备份来解决。理解ZooKeeper原理和实践，有助于构建高效分布式系统。
1941623231718325
25 4
老秦包你会
|
12天前
|
存储 搜索推荐 Java
Java远程连接本地开源分布式搜索引擎ElasticSearch
Java远程连接本地开源分布式搜索引擎ElasticSearch
老秦包你会
11 0
yuanzhengme
|
22天前
|
Java API Apache
ZooKeeper【基础 03】Java 客户端 Apache Curator 基础 API 使用举例（含源代码）
【4月更文挑战第11天】ZooKeeper【基础 03】Java 客户端 Apache Curator 基础 API 使用举例（含源代码）
yuanzhengme
33 11
日出等日落
|
22天前
|
分布式计算 Ubuntu 调度
如何本地搭建开源分布式任务调度系统DolphinScheduler并远程访问
如何本地搭建开源分布式任务调度系统DolphinScheduler并远程访问
日出等日落
37 0
薛伟同学
|
1月前
|
存储
ZooKeeper客户端常用命令
ZooKeeper客户端常用命令
薛伟同学
28 1
一只牛博
|
1月前
|
Java 网络安全 Apache
搭建Zookeeper集群：三台服务器，一场分布式之舞
搭建Zookeeper集群：三台服务器，一场分布式之舞
一只牛博
47 0
Zh.晨光
|
4月前
|
消息中间件 Java 网络安全
JAVAEE分布式技术之Zookeeper的第一次课
JAVAEE分布式技术之Zookeeper的第一次课
Zh.晨光
71 0

热门文章

最新文章

  • 1
    GaussDB分布式与单机模式的比较
  • 2
    缓存、分布式缓存和持久化
  • 3
    LLaMA-Factory 基于docker的大模型多卡分布式微调
  • 4
    【PolarDB-X从入门到精通】 第四讲：PolarDB分布式版安装部署（源码编译部署）
  • 5
    《MySQL 简易速速上手小册》第6章：MySQL 复制和分布式数据库（2024 最新版）
  • 6
    大数据技术与Python：结合Spark和Hadoop进行分布式计算
  • 7
    搞定了 6 种分布式ID，分库分表哪个适合做主键？
  • 8
    关于高并发下缓存失效的问题（本地锁 && 分布式锁 && Redission 详解）
  • 9
    分布式锁，Redission，其它实现问题讲解，以及面试题回答案例
  • 10
    如何本地搭建开源分布式任务调度系统DolphinScheduler并远程访问
  • 1
    ZooKeeper应用案例
    153
  • 2
    阿里云微服务引擎 MSE 及 API 网关 2024 年 02 月产品动态
    478
  • 3
    Kafka【环境搭建 02】kafka_2.11-2.4.1 基于 zookeeper 搭建高可用伪集群（一台服务器实现三个节点的 Kafka 集群）
    140
  • 4
    阿里云微服务引擎及 API 网关 2024 年 2 月产品动态
    629
  • 5
    ZooKeeper 项目
    78
  • 6
    Spring Boot整合Dubbo+Zookeeper实现RPC调用
    120
  • 7
    Zookeeper分布式锁
    90
  • 8
    【大厂面试演练】听说你很懂ZooKeeper？那我考考你
    146
  • 9
    阿里云微服务引擎 MSE 2024 年 01 月产品动态
    208
  • 10
    深度剖析dubbo和zookeeper关系
    19

    • 相关课程

    更多
  • MSE微服务测试最佳实践 - 自动化回归
  • Spring Boot+Vue.js+FastDFS实现分布式图片服务器
  • 基于Zookeeper、Dubbo构建互联网分布式基础架构
  • 大数据ZooKeeper快速入门
  • 分布式文件存储系统技术及实现
  • 分布式协调系统 Zookeeper 快速入门

    • 相关电子书

    更多
  • 《MSE 微服务网关》
  • 微服务引擎 MSE 治理中心重磅发布
  • 阿里云微服务引擎 MSE 2.0 线上发布

    • 相关实验场景

    更多
  • 基于MSE实现微服务的全链路灰度
  • 部署RabbitMQ
  • 基于MSE实现K8s集群内多服务的灰度发布
  • 搭建Hadoop环境
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）