大数据学习06_zookeeper3_javaAPI操作

这里操作Zookeeper的JavaAPI使用的是一套zookeeper客户端框架 Curator ，解决了很多 Zookeeper客户端非常底层的细节开发工作

Curator包含了几个包：

• curator-framework：对zookeeper的底层api的一些封装
• curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式 计数器等

先创建一个java工程，并且导入maven依赖。Maven依赖(使用curator的版本：2.12.0，对应Zookeeper的版本为：3.4.x，如果跨版本会有兼 容性问题，很有可能导致节点操作失败)

<dependencies>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.12.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.12.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.collections</groupId>
            <artifactId>google-collections</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>

            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.25</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <!-- java编译插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>

如果改变了curator要手动重新reimport

 接下来进行API操作：

• 创建永久节点
• 

@Test
public void createNode() throws Exception {
   RetryPolicy retryPolicy = new  ExponentialBackoffRetry(1000, 1);
   //获取客户端对象
   CuratorFramework client =    
CuratorFrameworkFactory.newClient("192.168.xxx.xxx:2181,192.168.xxx.xxx:2
181,192.168.xxx.xxx:2181", 1000, 1000, retryPolicy);
    
  //调用start开启客户端操作
  client.start();
    
  //通过create来进行创建节点，并且需要指定节点类型
  client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTEN
T).forPath("/hello3/world");
client.close();
}

• 创建临时节点
• 

public void createNode2() throws Exception {
 RetryPolicy retryPolicy = new  ExponentialBackoffRetry(3000, 1);
  CuratorFramework client =
CuratorFrameworkFactory.newClient("node01:2181,node02:2181,node03:2181",
3000, 3000, retryPolicy);
client.start();
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).
forPath("/hello5/world");
Thread.sleep(5000);
client.close();
}

• 修改节点数据
• 

@Test
 public void nodeData() throws Exception {
 RetryPolicy retryPolicy = new  ExponentialBackoffRetry(3000, 1);
 CuratorFramework client =
CuratorFrameworkFactory.newClient("node01:2181,node02:2181,node03:2181",
3000, 3000, retryPolicy);
 client.start();
 client.setData().forPath("/hello5", "hello7".getBytes());
 client.close();
}

• 节点数据查询

@Test
 public void updateNode() throws Exception {
RetryPolicy retryPolicy = new  ExponentialBackoffRetry(3000, 1);
 CuratorFramework client =
CuratorFrameworkFactory.newClient("node01:2181,node02:2181,node03:2181",
3000, 3000, retryPolicy);
 client.start();
 byte[] forPath = client.getData().forPath("/hello3");
 System.out.println(new String(forPath));
 client.close();
 }

• 节点watch机制

@Test
    public void watchZnode() throws Exception {
        RetryPolicy policy = new ExponentialBackoffRetry(3000, 3);
        String connectstr="192.168.xx.xxx:2181,192.168.xx.xxx:2181,192.168.xx.xxx:2181";
        CuratorFramework client =
                CuratorFrameworkFactory.newClient(connectstr, 8000, 8000, policy);
        client.start();
        // ExecutorService pool = Executors.newCachedThreadPool();
        //设置节点的cache

        TreeCache treeCache = new TreeCache(client, "/hello3");
        //设置监听器和处理过程
        treeCache.getListenable().addListener(new TreeCacheListener()
        {
            @Override
            public void childEvent(CuratorFramework client,
                                   TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if(data !=null){
                    switch (event.getType()) {
                        case NODE_ADDED:
                            System.out.println("NODE_ADDED : "+
                                    data.getPath() +" 数据:"+ new String(data.getData())+"监控到有新增节点");
                            break;
                        case NODE_REMOVED:
                            System.out.println("NODE_REMOVED : "+
                                    data.getPath() +" 数据:"+ new String(data.getData())+"监控到移除节点");
                            break;
                        case NODE_UPDATED:
                            System.out.println("NODE_UPDATED : "+
                                    data.getPath() +" 数据:"+ new String(data.getData())+"监控到有更新节点");
                            break;

                        default:
                            break;

                    }
                }else{
                    System.out.println( "data is null : "+
                            event.getType());
                }
            }
        });
        //开始监听
        treeCache.start();
        Thread.sleep(500000);
    }