Zookeeper(5) - Curator编程-天翼云

一、集成

<dependencies>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.4.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>2.4.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>2.4.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-test</artifactId>
        <version>2.4.2</version>
    </dependency>
</dependencies>

二、简单应用

2.1、同步

public class zkCliTest {
    private static CuratorFramework client;
    private String path = "/zk-book/c1";
    static {
        //重试策略，可以通过实现RetryPolicy接口来定义自己的策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .namespace("base") //指定命名空间，则下面所有操作都是基于base来进行的
                .build();
        client.start();
    }

    /*withMode, 这里区分临时节点和持久化节点*/
    @Test
    public void nodeCreate() throws Exception {
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
    }

    @Test
    public void nodeRead() throws Exception {
        Stat stat = new Stat();
        System.out.println(new String(client.getData().storingStatIn(stat).forPath(path)));
    }

    @Test
    public void nodeDel() throws Exception {
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        client.delete().deletingChildrenIfNeeded()
                .withVersion(stat.getVersion()).forPath(path);
    }
@Test
public void nodeSet() throws Exception {
    Stat stat = new Stat();
    client.getData().storingStatIn(stat).forPath(path);
    System.out.println("Success set node for : " + path + ", new version: "
            + client.setData().withVersion(stat.getVersion()).forPath(path).getVersion());
    try {
        client.setData().withVersion(stat.getVersion()).forPath(path);
    } catch (Exception e) {
        System.out.println("Fail set node due to " + e.getMessage());
    }
}
}

2.2、异步

在Zookeeper中，所有的异步通知事件处理都是由EventThread这个线程来处理的，EventThread线程用于串行处理所有的事件通知，其可以保证对事件处理的顺序性，但是一旦碰上复杂的处理单元，会消耗过长的处理时间，从而影响其他事件的处理，Curator允许用户传入Executor实例，这样可以将比较复杂的事件处理放到一个专门的线程池中去。

public class ZkBackTest {

    private static String path = "/zk-book/c1";

    private 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();
        /*
        * Main thread: main
            event[code: -110, type: CREATE]
            Thread of processResult: main-EventThread
            event1[code: -110, type: CREATE]
            Thread of processResult: pool-3-thread-1
        * */
        System.out.println("Main thread: " + Thread.currentThread().getName());

        final CountDownLatch semaphore = new CountDownLatch(2);
        ExecutorService tp = Executors.newFixedThreadPool(2);

        /**
         * BackgroundCallback接口 用来处理异步接口调用之后服务端返回的结果
         * */

        // 此处传入了自定义的Executor
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("event1[code: " + event.getResultCode() + ", type: " + event.getType() + "]");
                System.out.println("Thread of processResult: " + Thread.currentThread().getName());
                semaphore.countDown();
            }
        }, tp).forPath(path, "init".getBytes());

        // 此处没有传入自定义的Executor
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("event[code: " + event.getResultCode() + ", type: " + event.getType() + "]");
                System.out.println("Thread of processResult: " + Thread.currentThread().getName());
                semaphore.countDown();
            }
        }).forPath(path, "init".getBytes());

        semaphore.await();
        tp.shutdown();
    }
}

2.3、ZKPaths

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

   static String path = "/curator_zkpath_sample";
   static CuratorFramework client = CuratorFrameworkFactory.builder()
         .connectString( "domain1.book.zookeeper: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);
   }
}

2.4、EnsurePath

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

    static String path = "/zk-book/c1";
    static CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("domain1.book.zookeeper: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());
   }
}

三、高级应用

3.1、分布式锁

//使用Curator实现分布式锁功能
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();//+1
   }
}

3.2、计时器

public class Recipes_DistAtomicInt {

   static String distatomicint_path = "/curator_recipes_distatomicint_path";
    static CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("domain1.book.zookeeper: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() );
   }
}

3.3、分布式Barrier

//使用Curator实现分布式Barrier
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("domain1.book.zookeeper: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();
   }
}

public class Recipes_Barrier2 {
   static String barrier_path = "/curator_recipes_barrier_path";
   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("domain1.book.zookeeper:2181")
                           .retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
                  client.start();
                  DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, barrier_path,5);
                  Thread.sleep( Math.round(Math.random() * 3000) );
                  System.out.println(Thread.currentThread().getName() + "号进入barrier" );
                  barrier.enter();
                  System.out.println("启动...");
                  Thread.sleep( Math.round(Math.random() * 3000) );
                  barrier.leave();
                  System.out.println( "退出..." );
               } catch (Exception e) {}
            }
         }).start();
      }
   }
}

public class Recipes_CyclicBarrier {

   public static CyclicBarrier barrier = new CyclicBarrier( 3 );
   public static void main( String[] args ) throws IOException, InterruptedException {
      ExecutorService executor = Executors.newFixedThreadPool( 3 );
      executor.submit( new Thread( new Runner( "1号选手" ) ) );
      executor.submit( new Thread( new Runner( "2号选手" ) ) );
      executor.submit( new Thread( new Runner( "3号选手" ) ) );
      executor.shutdown();
   }
}
class Runner implements Runnable {
   private String name;
   public Runner( String name ) {
      this.name = name;
   }
   public void run() {
      System.out.println( name + " 准备好了." );
      try {
         Recipes_CyclicBarrier.barrier.await();
      } catch ( Exception e ) {}
      System.out.println( name + " 起跑!" );
   }
}

四、节点管理

4.1、监听

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()));
    }
});

PathChildrenCache cache = new PathChildrenCache(client, path, true);
cache.start(PathChildrenCache.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;
        }
    }
});

4.2、选举

String master_path = "/curator_recipes_master_path";
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 );

五、测试管理

5.1、单机模拟

public class TestingServer_Sample {
   static String path = "/zookeeper";
   public static void main(String[] args) throws Exception {
      TestingServer server = new TestingServer(2181,new File("/home/admin/zk-book-data"));
      
      CuratorFramework client = CuratorFrameworkFactory.builder()
               .connectString(server.getConnectString())
               .sessionTimeoutMs(5000)
               .retryPolicy(new ExponentialBackoffRetry(1000, 3))
               .build();
        client.start();
        System.out.println( client.getChildren().forPath( path ));
        server.close();
   }
}

5.2、集群模拟

public class TestingCluster_Sample {

   public static void main(String[] args) throws Exception {
      TestingCluster cluster = new TestingCluster(3);
      cluster.start();
      Thread.sleep(2000);
      
      TestingZooKeeperServer leader = null;
      for(TestingZooKeeperServer zs : cluster.getServers()){
         System.out.print(zs.getInstanceSpec().getServerId()+"-");
         System.out.print(zs.getQuorumPeer().getServerState()+"-");  
         System.out.println(zs.getInstanceSpec().getDataDirectory().getAbsolutePath());
         if( zs.getQuorumPeer().getServerState().equals( "leading" )){
            leader = zs;
         }
      }
        leader.kill();
        System.out.println( "--After leader kill:" );
        for(TestingZooKeeperServer zs : cluster.getServers()){
         System.out.print(zs.getInstanceSpec().getServerId()+"-");
         System.out.print(zs.getQuorumPeer().getServerState()+"-");  
         System.out.println(zs.getInstanceSpec().getDataDirectory().getAbsolutePath());
      }
        cluster.stop();
   }
}

活动

智算服务

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

Zookeeper(5) - Curator编程

Zookeeper(5) - Curator编程

一、集成

二、简单应用

2.1、同步

2.2、异步

2.3、ZKPaths

2.4、EnsurePath

三、高级应用

3.1、分布式锁

3.2、计时器

3.3、分布式Barrier

四、节点管理

4.1、监听

4.2、选举

五、测试管理

5.1、单机模拟

5.2、集群模拟

相关文章

Podman安装部署kafka和管理界面（快速跑起来）

【dubbo】分布式和dubbo

kafka集群重启方法

zookeeper配置systemctl

zookeeper节点监视Watcher

Nginx系列：配置跳转的常用方式

Dubbo及Zookeeper安装

一键式完全删除CDH 6.3

聊一下zookeeper的同步算法。

金币路径。

作者介绍

最新文章

[redis] 生成rdb对磁盘io的压力测试

Java操作Zookeeper Maven

zookeeper API 基础

Python：kazoo模块与Zookeeper交互

热门文章

zookeeper API 基础

Python：kazoo模块与Zookeeper交互

Java操作Zookeeper Maven

[redis] 生成rdb对磁盘io的压力测试

热门标签

相关产品

弹性云主机

天翼云电脑（公众版）

对象存储

云硬盘

随机文章

zookeeper API 基础

[redis] 生成rdb对磁盘io的压力测试

Python：kazoo模块与Zookeeper交互

Java操作Zookeeper Maven