• Zookeeper学习一

    初识 Zookeeper

    Zookeeper 是 Apache Hadoop 项目下的一个子项目，是一个树形目录服务（B树）。

    Zookeeper 翻译过来就是 动物园管理员，他是用来管 Hadoop（大象）、Hive(蜜蜂)、Pig(小 猪)的管理员。简称zk

    Zookeeper 是一个分布式的、开源的分布式应用程序的协调服务。

    Zookeeper 提供的主要功能包括： 配置管理 分布式锁 集群管理

    Zookeeper 安装与配置

    1.1 下载安装

    1、环境准备

    ZooKeeper服务器是用Java创建的，它运行在JVM之上。需要安装JDK 7或更高版本。

    2、上传

    将下载的ZooKeeper放到/opt/ZooKeeper目录下

    ​

    1. #上传zookeeper alt+p
    2. put f:/setup/apache-zookeeper-3.5.6-bin.tar.gz
    3. #打开 opt目录
    4. cd /opt
    5. #创建zooKeeper目录
    6. mkdir  zooKeeper
    7. #将zookeeper安装包移动到 /opt/zooKeeper
    8. mv apache-zookeeper-3.5.6-bin.tar.gz /opt/zookeeper/

    3、解压

    将tar包解压到/opt/zookeeper目录下

    tar -zxvf apache-ZooKeeper-3.5.6-bin.tar.gz

    1.2 配置启动

    1、配置zoo.cfg

    进入到conf目录拷贝一个zoo_sample.cfg并完成配置

    1. #进入到conf目录
    2. cd /opt/zooKeeper/apache-zooKeeper-3.5.6-bin/conf/
    3. #拷贝
    4. cp  zoo_sample.cfg  zoo.cfg

    修改zoo.cfg

    1. #打开目录
    2. cd /opt/zooKeeper/
    3. #创建zooKeeper存储目录
    4. mkdir  zkdata
    5. #修改zoo.cfg
    6. vim /opt/zooKeeper/apache-zooKeeper-3.5.6-bin/conf/zoo.cfg

    修改存储目录：dataDir=/opt/zookeeper/zkdata

    2、启动ZooKeeper

    1. cd /opt/zooKeeper/apache-zooKeeper-3.5.6-bin/bin/
    2. #启动
    3.  ./zkServer.sh  start

    看到上图表示ZooKeeper成功启动

    3、查看ZooKeeper状态

    ./zkServer.sh status

    zookeeper启动成功。standalone代表zk没有搭建集群，现在是单节点

    zookeeper没有启动

    Zookeeper 命令操作

    Zookeeper 数据模型

    ZooKeeper 是一个树形目录服务,其数据模型和Unix的文件系统目录树很类似，拥有一个层次化结构。

    这里面的每一个节点都被称为： ZNode，每个节点上都会保存自己的数据和节点信息。

     节点可以拥有子节点，同时也允许少量（1MB）数据存储在该节点之下。

    节点可以分为四大类：

    • PERSISTENT 持久化节点
    • EPHEMERAL 临时节点 ：-e
    • PERSISTENT_SEQUENTIAL 持久化顺序节点 ：-s
    • EPHEMERAL_SEQUENTIAL 临时顺序节点  ：-es

    Zookeeper服务端常用命令

    • 启动 ZooKeeper 服务: ./zkServer.sh start
    • 查看 ZooKeeper 服务状态: ./zkServer.sh status
    • 停止 ZooKeeper 服务: ./zkServer.sh stop
    • 重启 ZooKeeper 服务: ./zkServer.sh restart

    Zookeeper客户端常见命令

    连接ZooKeeper服务器：./zkCli.sh –server ip:port

    断开连接：quit

    设置节点值：set /节点path value

    查看命令帮助：help

    删除单个节点：delete /节点path

    显示指定目录下节点：ls 目录

    删除带有子节点的节点：deleteall /节点path

    创建节点：create /节点path value

    获取节点值：get /节点path

    Zookeeper JavaAPI 操作

    建立连接

    建立连接有两种方式，一种是调用工厂对象的newClient（）方法，另一种就是调用工厂对象的builder()，通过链式调用的方法就连接信息传入工厂中。

    下面是代码示例：

    1.     /**
    2.      * 建立连接
    3.      */
    4.     @Before
    5.     public void testConnect() {
    6.  
    7.         /*
    8.          *
    9.          * @param connectString       连接字符串。zk server 地址和端口 "192.168.149.135:2181,192.168.149.136:2181"
    10.          * @param sessionTimeoutMs    会话超时时间 单位ms
    11.          * @param connectionTimeoutMs 连接超时时间 单位ms
    12.          * @param retryPolicy         重试策略
    13.          */
    14.        /* //重试策略
    15.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
    16.         //1.第一种方式
    17.         CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.149.135:2181",
    18.                 60 * 1000, 15 * 1000, retryPolicy);*/
    19.         //重试策略
    20.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
    21.         //2.第二种方式
    22.         //CuratorFrameworkFactory.builder();
    23.         client = CuratorFrameworkFactory.builder()
    24.                 .connectString("192.168.149.135:2181")
    25.                 .sessionTimeoutMs(60 * 1000)
    26.                 .connectionTimeoutMs(15 * 1000)
    27.                 .retryPolicy(retryPolicy)
    28. //命名空间，使创建的节点都在命名空间的路径下面
    29.                 .namespace("kjz")
    30.                 .build();
    31.  
    32.         //开启连接
    33.         client.start();
    34.  
    35.     }

    注意此方法需要加上@Before注解，表示其他测试方法执行前需要先执行加了@Before注解的方法。因为每次进行crud操作时都需要与ZooKeeper Server建立连接。

    建立了连接，操作完毕后同时需要释放连接，在对应方法上面加一个@After注解，表示每次进行测试最后都要执行该方法。

    1.     @After
    2.     public void close() {
    3.         if (client != null) {
    4.             client.close();
    5.         }
    6.     }

    添加节点

    代码示例如下：

    1.    /**
    2.      * 创建节点：create 持久 临时 顺序 数据
    3.      * 1. 基本创建 ：create().forPath("")
    4.      * 2. 创建节点 带有数据:create().forPath("",data)
    5.      * 3. 设置节点的类型：create().withMode().forPath("",data)
    6.      * 4. 创建多级节点  /app1/p1 ：create().creatingParentsIfNeeded().forPath("",data)
    7.      */
    8.     @Test
    9.     public void testCreate() throws Exception {
    10.         //2. 创建节点 带有数据
    11.         //如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
    12.         String path = client.create().forPath("/app2", "hehe".getBytes());
    13.         System.out.println(path);
    14.  
    15.     }
    16.  
    17.     @Test
    18.     public void testCreate2() throws Exception {
    19.         //1. 基本创建
    20.         //如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
    21.         String path = client.create().forPath("/app1");
    22.         System.out.println(path);
    23.  
    24.     }
    25.  
    26.     @Test
    27.     public void testCreate3() throws Exception {
    28.         //3. 设置节点的类型
    29.         //默认类型：持久化
    30.         String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
    31.         System.out.println(path);
    32.  
    33.     }
    34.  
    35.     @Test
    36.     public void testCreate4() throws Exception {
    37.         //4. 创建多级节点  /app1/p1
    38.         //creatingParentsIfNeeded():如果父节点不存在，则创建父节点
    39.         String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");
    40.         System.out.println(path);
    41.     }

    Curator中提供了一个枚举类，里面定义了设置不同类型节点的常量

    运行testCreate3()时我发现ZooKeeper中并没有保存我创建的节点，原因是在创建节点时指定的节点类型为临时节点，临时节点在会话结束后就会删除。我通过运行testCreate3()这个方法来创建一个节点，运行结束，说明会话也结束了，ZooKeeper就会把节点删除了。

    删除节点

    删除节点： delete deleteall

    删除单个节点:delete().forPath("/app1");

    删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");

    必须成功的删除:为了防止网络抖动。本质就是重试。client.delete().guaranteed().forPath("/app2");

    回调：inBackground；

    代码示例：

    1. /**
    2.      * 删除节点： delete deleteall
    3.      * 1. 删除单个节点:delete().forPath("/app1");
    4.      * 2. 删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");
    5.      * 3. 必须成功的删除:为了防止网络抖动。本质就是重试。  client.delete().guaranteed().forPath("/app2");
    6.      * 4. 回调：inBackground
    7.      * @throws Exception
    8.      */
    9.  
    10.  
    11.     @Test
    12.     public void testDelete() throws Exception {
    13.         // 1. 删除单个节点
    14.         client.delete().forPath("/app1");
    15.     }
    16.  
    17.     @Test
    18.     public void testDelete2() throws Exception {
    19.         //2. 删除带有子节点的节点
    20.         client.delete().deletingChildrenIfNeeded().forPath("/app4");
    21.     }
    22.     @Test
    23.     public void testDelete3() throws Exception {
    24.         //3. 必须成功的删除
    25.         client.delete().guaranteed().forPath("/app2");
    26.     }
    27.  
    28.     @Test
    29.     public void testDelete4() throws Exception {
    30.         //4. 回调
    31.         client.delete().guaranteed().inBackground(new BackgroundCallback(){
    32.  
    33.             @Override
    34.             public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
    35.                 System.out.println("我被删除了~");
    36.                 System.out.println(event);
    37.             }
    38.         }).forPath("/app1");
    39.     }

    修改节点

    基本修改数据 setData().forPath()

    根据版本修改数据 setData().withVersion().forPath()

    version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。

    代码示例：

    1.  /**
    2.      * 修改数据
    3.      * 1. 基本修改数据：setData().forPath()
    4.      * 2. 根据版本修改: setData().withVersion().forPath()
    5.      * * version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。
    6.      *
    7.      * @throws Exception
    8.      */
    9.     @Test
    10.     public void testSet() throws Exception {
    11.         client.setData().forPath("/app1", "itcast".getBytes());
    12.     }
    13.  
    14.  
    15.     @Test
    16.     public void testSetForVersion() throws Exception {
    17.  
    18.         Stat status = new Stat();
    19.         //3. 查询节点状态信息：ls -s
    20.         client.getData().storingStatIn(status).forPath("/app1");
    21.  
    22.  
    23.         int version = status.getVersion();//查询出来的 3
    24.         System.out.println(version);
    25.         client.setData().withVersion(version).forPath("/app1", "hehe".getBytes());
    26.     }

    查询节点

    查询数据：get: getData().forPath()

    查询子节点： ls: getChildren().forPath()

    查询节点状态信息：ls -s:getData().storingStatIn(状态对象).forPath() 

    代码示例:

    1. @Test
    2.     public void testGet1() throws Exception {
    3.         //1. 查询数据：get
    4.         byte[] data = client.getData().forPath("/app1");
    5.         System.out.println(new String(data));
    6.     }
    7.  
    8.     @Test
    9.     public void testGet2() throws Exception {
    10.         // 2. 查询子节点： ls
    11.         List path = client.getChildren().forPath("/");
    12.  
    13.         System.out.println(path);
    14.     }
    15.  
    16.     @Test
    17.     public void testGet3() throws Exception {
    18.  
    19.  
    20.         Stat status = new Stat();
    21.         System.out.println(status);
    22.         //3. 查询节点状态信息：ls -s
    23.         client.getData().storingStatIn(status).forPath("/app1");
    24.  
    25.         System.out.println(status);
    26.  
    27.     }

    Stat里面封装了节点的状态信息

    完整代码如下：

    1. package com.kjz.curator;
    2.  
    3. import org.apache.curator.RetryPolicy;
    4. import org.apache.curator.framework.CuratorFramework;
    5. import org.apache.curator.framework.CuratorFrameworkFactory;
    6. import org.apache.curator.framework.api.BackgroundCallback;
    7. import org.apache.curator.framework.api.CuratorEvent;
    8. import org.apache.curator.retry.ExponentialBackoffRetry;
    9. import org.apache.zookeeper.CreateMode;
    10. import org.apache.zookeeper.data.Stat;
    11. import org.junit.After;
    12. import org.junit.Before;
    13. import org.junit.Test;
    14.  
    15. import java.util.List;
    16.  
    17. public class CuratorTest {
    18.  
    19.  
    20.     private CuratorFramework client;
    21.  
    22.     /**
    23.      * 建立连接
    24.      */
    25.     @Before
    26.     public void testConnect() {
    27.  
    28.         /*
    29.          *
    30.          * @param connectString       连接字符串。zk server 地址和端口 "192.168.149.135:2181,192.168.149.136:2181"
    31.          * @param sessionTimeoutMs    会话超时时间 单位ms
    32.          * @param connectionTimeoutMs 连接超时时间 单位ms
    33.          * @param retryPolicy         重试策略
    34.          */
    35.        /* //重试策略
    36.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
    37.         //1.第一种方式
    38.         CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.149.135:2181",
    39.                 60 * 1000, 15 * 1000, retryPolicy);*/
    40.         //重试策略
    41.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
    42.         //2.第二种方式
    43.         //CuratorFrameworkFactory.builder();
    44.         client = CuratorFrameworkFactory.builder()
    45.                 .connectString("192.168.149.135:2181")
    46.                 .sessionTimeoutMs(60 * 1000)
    47.                 .connectionTimeoutMs(15 * 1000)
    48.                 .retryPolicy(retryPolicy)
    49.                 .namespace("kjz")
    50.                 .build();
    51.  
    52.         //开启连接
    53.         client.start();
    54.  
    55.     }
    56. //==============================create=============================================================================
    57.  
    58.     /**
    59.      * 创建节点：create 持久 临时 顺序 数据
    60.      * 1. 基本创建 ：create().forPath("")
    61.      * 2. 创建节点 带有数据:create().forPath("",data)
    62.      * 3. 设置节点的类型：create().withMode().forPath("",data)
    63.      * 4. 创建多级节点  /app1/p1 ：create().creatingParentsIfNeeded().forPath("",data)
    64.      */
    65.     @Test
    66.     public void testCreate() throws Exception {
    67.         //2. 创建节点 带有数据
    68.         //如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
    69.         String path = client.create().forPath("/app2", "hehe".getBytes());
    70.         System.out.println(path);
    71.  
    72.     }
    73.  
    74.     @Test
    75.     public void testCreate2() throws Exception {
    76.         //1. 基本创建
    77.         //如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储
    78.         String path = client.create().forPath("/app1");
    79.         System.out.println(path);
    80.  
    81.     }
    82.  
    83.     @Test
    84.     public void testCreate3() throws Exception {
    85.         //3. 设置节点的类型
    86.         //默认类型：持久化
    87.         String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
    88.         System.out.println(path);
    89.  
    90.  
    91.     }
    92.  
    93.     @Test
    94.     public void testCreate4() throws Exception {
    95.         //4. 创建多级节点  /app1/p1
    96.         //creatingParentsIfNeeded():如果父节点不存在，则创建父节点
    97.         String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");
    98.         System.out.println(path);
    99.     }
    100. //===========================get================================================================================
    101.  
    102.     /**
    103.      * 查询节点：
    104.      * 1. 查询数据：get: getData().forPath()
    105.      * 2. 查询子节点： ls: getChildren().forPath()
    106.      * 3. 查询节点状态信息：ls -s:getData().storingStatIn(状态对象).forPath()
    107.      */
    108.  
    109.     @Test
    110.     public void testGet1() throws Exception {
    111.         //1. 查询数据：get
    112.         byte[] data = client.getData().forPath("/app1");
    113.         System.out.println(new String(data));
    114.     }
    115.  
    116.     @Test
    117.     public void testGet2() throws Exception {
    118.         // 2. 查询子节点： ls
    119.         List path = client.getChildren().forPath("/");
    120.  
    121.         System.out.println(path);
    122.     }
    123.  
    124.     @Test
    125.     public void testGet3() throws Exception {
    126.  
    127.  
    128.         Stat status = new Stat();
    129.         System.out.println(status);
    130.         //3. 查询节点状态信息：ls -s
    131.         client.getData().storingStatIn(status).forPath("/app1");
    132.  
    133.         System.out.println(status);
    134.  
    135.     }
    136.  
    137.     //===========================set================================================================================
    138.  
    139.     /**
    140.      * 修改数据
    141.      * 1. 基本修改数据：setData().forPath()
    142.      * 2. 根据版本修改: setData().withVersion().forPath()
    143.      * * version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。
    144.      *
    145.      * @throws Exception
    146.      */
    147.     @Test
    148.     public void testSet() throws Exception {
    149.         client.setData().forPath("/app1", "kjz".getBytes());
    150.     }
    151.  
    152.  
    153.     @Test
    154.     public void testSetForVersion() throws Exception {
    155.  
    156.         Stat status = new Stat();
    157.         //3. 查询节点状态信息：ls -s
    158.         client.getData().storingStatIn(status).forPath("/app1");
    159.  
    160.  
    161.         int version = status.getVersion();//查询出来的 3
    162.         System.out.println(version);
    163.         client.setData().withVersion(version).forPath("/app1", "hehe".getBytes());
    164.     }
    165.  
    166.     //===========================delete================================================================================
    167.  
    168.     /**
    169.      * 删除节点： delete deleteall
    170.      * 1. 删除单个节点:delete().forPath("/app1");
    171.      * 2. 删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");
    172.      * 3. 必须成功的删除:为了防止网络抖动。本质就是重试。  client.delete().guaranteed().forPath("/app2");
    173.      * 4. 回调：inBackground
    174.      * @throws Exception
    175.      */
    176.  
    177.  
    178.     @Test
    179.     public void testDelete() throws Exception {
    180.         // 1. 删除单个节点
    181.         client.delete().forPath("/app1");
    182.     }
    183.  
    184.     @Test
    185.     public void testDelete2() throws Exception {
    186.         //2. 删除带有子节点的节点
    187.         client.delete().deletingChildrenIfNeeded().forPath("/app4");
    188.     }
    189.     @Test
    190.     public void testDelete3() throws Exception {
    191.         //3. 必须成功的删除
    192.         client.delete().guaranteed().forPath("/app2");
    193.     }
    194.  
    195.     @Test
    196.     public void testDelete4() throws Exception {
    197.         //4. 回调
    198.         client.delete().guaranteed().inBackground(new BackgroundCallback(){
    199.  
    200.             @Override
    201.             public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
    202.                 System.out.println("我被删除了~");
    203.                 System.out.println(event);
    204.             }
    205.         }).forPath("/app1");
    206.     }
    207.  
    208.     @After
    209.     public void close() {
    210.         if (client != null) {
    211.             client.close();
    212.         }
    213.     }
    214.  
    215.  
    216. }

    Watch事件监听

    • ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。
    • ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
    • ZooKeeper 原生支持通过注册Watcher来进行事件监听，但是其使用并不是特别方便     需要开发人员自己反复注册Watcher，比较繁琐。
    • Curator引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。

    ZooKeeper提供了三种Watcher：

    •         NodeCache : 只是监听某一个特定的节点
    •          PathChildrenCache : 监控一个ZNode的子节点.
    •          TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache             的组合

    演示 NodeCache：给指定一个节点注册监听器

    1.     /**
    2.      * 演示 NodeCache：给指定一个节点注册监听器
    3.      */
    4.  
    5.     @Test
    6.     public void testNodeCache() throws Exception {
    7.         //1. 创建NodeCache对象
    8.         final NodeCache nodeCache = new NodeCache(client,"/app1");
    9.         //2. 注册监听
    10.         nodeCache.getListenable().addListener(new NodeCacheListener() {
    11.             @Override
    12.             public void nodeChanged() throws Exception {
    13.                 System.out.println("节点变化了~");
    14.  
    15.                 //获取修改节点后的数据
    16.                 byte[] data = nodeCache.getCurrentData().getData();
    17.                 System.out.println(new String(data));
    18.             }
    19.         });
    20.  
    21.         //3. 开启监听.如果设置为true，则开启监听是，加载缓冲数据
    22.         nodeCache.start(true);
    23.  
    24.  
    25.         while (true){
    26.  
    27.         }
    28.     }

    演示 PathChildrenCache：监听某个节点的所有子节点们

    1.  /**
    2.      * 演示 PathChildrenCache：监听某个节点的所有子节点们
    3.      */
    4.  
    5.     @Test
    6.     public void testPathChildrenCache() throws Exception {
    7.         //1.创建监听对象
    8.         PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"/app2",true);
    9.  
    10.         //2. 绑定监听器
    11.         pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
    12.             @Override
    13.             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
    14.                 System.out.println("子节点变化了~");
    15.                 System.out.println(event);
    16.                 //监听子节点的数据变更，并且拿到变更后的数据
    17.                 //1.获取类型
    18.                 PathChildrenCacheEvent.Type type = event.getType();
    19.                 //2.判断类型是否是update
    20.                 if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
    21.                     System.out.println("数据变了！！！");
    22.                     byte[] data = event.getData().getData();
    23.                     System.out.println(new String(data));
    24.  
    25.                 }
    26.             }
    27.         });
    28.         //3. 开启
    29.         pathChildrenCache.start();
    30.  
    31.         while (true){
    32.  
    33.         }
    34.     }

    数据变更类型的枚举类。

    演示 TreeCache：监听某个节点自己和所有子节点们，相当于PathChildrenCache和NodeCache的组合

    1.  /**
    2.      * 演示 NodeCache：给指定一个节点注册监听器
    3.      */
    4.  
    5.     @Test
    6.     public void testNodeCache() throws Exception {
    7.         //1. 创建NodeCache对象
    8.         final NodeCache nodeCache = new NodeCache(client,"/app1");
    9.         //2. 注册监听
    10.         nodeCache.getListenable().addListener(new NodeCacheListener() {
    11.             @Override
    12.             public void nodeChanged() throws Exception {
    13.                 System.out.println("节点变化了~");
    14.                 //获取修改节点后的数据
    15.                 byte[] data = nodeCache.getCurrentData().getData();
    16.                 System.out.println(new String(data));
    17.             }
    18.         });
    19.  
    20.         //3. 开启监听.如果设置为true，则开启监听是，加载缓冲数据
    21.         nodeCache.start(true);
    22.  
    23.  
    24.         while (true){
    25.  
    26.         }
    27.     }

    完整代码如下：

    1. package com.kjz.curator;
    2.  
    3. import org.apache.curator.RetryPolicy;
    4. import org.apache.curator.framework.CuratorFramework;
    5. import org.apache.curator.framework.CuratorFrameworkFactory;
    6. import org.apache.curator.framework.api.BackgroundCallback;
    7. import org.apache.curator.framework.api.CuratorEvent;
    8. import org.apache.curator.framework.recipes.cache.*;
    9. import org.apache.curator.retry.ExponentialBackoffRetry;
    10. import org.apache.zookeeper.CreateMode;
    11. import org.apache.zookeeper.data.Stat;
    12. import org.junit.After;
    13. import org.junit.Before;
    14. import org.junit.Test;
    15.  
    16. import java.util.List;
    17.  
    18. public class CuratorWatcherTest {
    19.  
    20.  
    21.     private CuratorFramework client;
    22.  
    23.     /**
    24.      * 建立连接
    25.      */
    26.     @Before
    27.     public void testConnect() {
    28.  
    29.         /*
    30.          *
    31.          * @param connectString       连接字符串。zk server 地址和端口 "192.168.149.135:2181,192.168.149.136:2181"
    32.          * @param sessionTimeoutMs    会话超时时间 单位ms
    33.          * @param connectionTimeoutMs 连接超时时间 单位ms
    34.          * @param retryPolicy         重试策略
    35.          */
    36.        /* //重试策略
    37.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);
    38.         //1.第一种方式
    39.         CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.149.135:2181",
    40.                 60 * 1000, 15 * 1000, retryPolicy);*/
    41.         //重试策略
    42.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
    43.         //2.第二种方式
    44.         //CuratorFrameworkFactory.builder();
    45.         client = CuratorFrameworkFactory.builder()
    46.                 .connectString("192.168.149.135:2181")
    47.                 .sessionTimeoutMs(60 * 1000)
    48.                 .connectionTimeoutMs(15 * 1000)
    49.                 .retryPolicy(retryPolicy)
    50.                 .namespace("kjz")
    51.                 .build();
    52.  
    53.         //开启连接
    54.         client.start();
    55.  
    56.     }
    57.  
    58.     @After
    59.     public void close() {
    60.         if (client != null) {
    61.             client.close();
    62.         }
    63.     }
    64.  
    65.     /**
    66.      * 演示 NodeCache：给指定一个节点注册监听器
    67.      */
    68.  
    69.     @Test
    70.     public void testNodeCache() throws Exception {
    71.         //1. 创建NodeCache对象
    72.         final NodeCache nodeCache = new NodeCache(client,"/app1");
    73.         //2. 注册监听
    74.         nodeCache.getListenable().addListener(new NodeCacheListener() {
    75.             @Override
    76.             public void nodeChanged() throws Exception {
    77.                 System.out.println("节点变化了~");
    78.  
    79.                 //获取修改节点后的数据
    80.                 byte[] data = nodeCache.getCurrentData().getData();
    81.                 System.out.println(new String(data));
    82.             }
    83.         });
    84.  
    85.         //3. 开启监听.如果设置为true，则开启监听是，加载缓冲数据
    86.         nodeCache.start(true);
    87.  
    88.  
    89.         while (true){
    90.  
    91.         }
    92.     }
    93.  
    94.  
    95.  
    96.     /**
    97.      * 演示 PathChildrenCache：监听某个节点的所有子节点们
    98.      */
    99.  
    100.     @Test
    101.     public void testPathChildrenCache() throws Exception {
    102.         //1.创建监听对象
    103.         PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"/app2",true);
    104.  
    105.         //2. 绑定监听器
    106.         pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
    107.             @Override
    108.             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
    109.                 System.out.println("子节点变化了~");
    110.                 System.out.println(event);
    111.                 //监听子节点的数据变更，并且拿到变更后的数据
    112.                 //1.获取类型
    113.                 PathChildrenCacheEvent.Type type = event.getType();
    114.                 //2.判断类型是否是update
    115.                 if(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
    116.                     System.out.println("数据变了！！！");
    117.                     byte[] data = event.getData().getData();
    118.                     System.out.println(new String(data));
    119.  
    120.                 }
    121.             }
    122.         });
    123.         //3. 开启
    124.         pathChildrenCache.start();
    125.  
    126.         while (true){
    127.  
    128.         }
    129.     }
    130.  
    131.  
    132.  
    133.  
    134.  
    135.     /**
    136.      * 演示 TreeCache：监听某个节点自己和所有子节点们
    137.      */
    138.  
    139.     @Test
    140.     public void testTreeCache() throws Exception {
    141.         //1. 创建监听器
    142.         TreeCache treeCache = new TreeCache(client,"/app2");
    143.  
    144.         //2. 注册监听
    145.         treeCache.getListenable().addListener(new TreeCacheListener() {
    146.             @Override
    147.             public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
    148.                 System.out.println("节点变化了");
    149.                 System.out.println(event);
    150.             }
    151.         });
    152.  
    153.         //3. 开启
    154.         treeCache.start();
    155.  
    156.         while (true){
    157.  
    158.         }
    159.     }
    160.  
    161. }

    分布式锁实现

    在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个JVM之下，没有任何问题。 但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。 那么就需要一种更加高级的锁机制，来处理种 跨机器的进程之间的数据同步问题——这就是分布式锁。
    分布式锁常见的实现方式：
    下面介绍ZooKeeper的实现方式：

    Zookeeper分布式锁原理

    核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。

    1. 客户端获取锁时，在lock节点下创建临时顺序节点。临时：防止获取到锁的服务宕机了导致锁无法释放，临时节点会在会话结束后自动删除。顺序：找到最小节点，创建最小节点的服务获取到锁。
    2. 然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。
    3. 如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。
    4. 如果发现比自己小的那个节点被删除，则客户端的 Watcher会收到相应通知，此时再次判断自己创建的节点     是否是lock子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点     并注册监听。

    Curator实现分布式锁API

    在Curator中有五种锁方案：

    • InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）
    • InterProcessMutex：分布式可重入排它锁
    • InterProcessReadWriteLock：分布式读写锁
    • InterProcessMultiLock：将多个锁作为单个实体管理的容器
    • InterProcessSemaphoreV2：共享信号量

    案例：模拟12306售票

    模拟12306服务

    1. package com.kjz.curator;
    2.  
    3. import org.apache.curator.RetryPolicy;
    4. import org.apache.curator.framework.CuratorFramework;
    5. import org.apache.curator.framework.CuratorFrameworkFactory;
    6. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    7. import org.apache.curator.retry.ExponentialBackoffRetry;
    8.  
    9. import java.util.concurrent.TimeUnit;
    10.  
    11. public class Ticket12306 implements Runnable{
    12.  
    13.     private int tickets = 10;//数据库的票数
    14.  
    15.     private InterProcessMutex lock ;
    16.  
    17.  
    18.     public Ticket12306(){
    19.         //重试策略
    20.         RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
    21.         //2.第二种方式
    22.         //CuratorFrameworkFactory.builder();
    23.         CuratorFramework client = CuratorFrameworkFactory.builder()
    24.                 .connectString("192.168.149.135:2181")
    25.                 .sessionTimeoutMs(60 * 1000)
    26.                 .connectionTimeoutMs(15 * 1000)
    27.                 .retryPolicy(retryPolicy)
    28.                 .build();
    29.  
    30.         //开启连接
    31.         client.start();
    32.  
    33.         lock = new InterProcessMutex(client,"/lock");
    34.     }
    35.  
    36.     @Override
    37.     public void run() {
    38.  
    39.         while(true){
    40.             //获取锁
    41.             try {
    42.                 lock.acquire(3, TimeUnit.SECONDS);
    43.                 if(tickets > 0){
    44.  
    45.                     System.out.println(Thread.currentThread()+":"+tickets);
    46.                     Thread.sleep(100);
    47.                     tickets--;
    48.                 }
    49.             } catch (Exception e) {
    50.                 e.printStackTrace();
    51.             }finally {
    52.                 //释放锁
    53.                 try {
    54.                     lock.release();
    55.                 } catch (Exception e) {
    56.                     e.printStackTrace();
    57.                 }
    58.             }
    59.  
    60.  
    61.  
    62.         }
    63.  
    64.     }
    65. }

    模拟买票

    1. package com.kjz.curator;
    2.  
    3. import org.apache.curator.RetryPolicy;
    4. import org.apache.curator.framework.CuratorFramework;
    5. import org.apache.curator.framework.CuratorFrameworkFactory;
    6. import org.apache.curator.framework.recipes.cache.*;
    7. import org.apache.curator.retry.ExponentialBackoffRetry;
    8. import org.junit.After;
    9. import org.junit.Before;
    10. import org.junit.Test;
    11.  
    12. public class LockTest {
    13.  
    14.  
    15.     public static void main(String[] args) {
    16.         Ticket12306 ticket12306 = new Ticket12306();
    17.  
    18.         //创建客户端
    19.         Thread t1 = new Thread(ticket12306,"携程");
    20.         Thread t2 = new Thread(ticket12306,"飞猪");
    21.  
    22.         t1.start();
    23.         t2.start();
    24.     }
    25.  
    26. }
  • 相关阅读:
    No6-5.从零搭建spring-cloud-alibaba微服务框架，添加用户鉴权逻辑，操作权限等（五，no6-5）
    依赖注入的正确打开方式 bilibili/kratos × google/wire
    高校教务系统登录页面JS分析——巢湖学院
    加密大崩盘，Web3游戏到底还有没有未来？5篇论文深度探讨
    计算机操作系统学习（六）设备管理
    Apache HttpClient 5 使用详细教程
    「PCB智能生产」MES系统在设备管理中的应用
    字节5年测试工程师经验分享，软件测试真的很简单么？
    tableau高级绘图（十）-Tableau绘制圆形图
    英语学术论文简短语句摘抄
  • 原文地址：https://blog.csdn.net/m0_74229735/article/details/137245523