首先创建一个Maven项目
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <groupId>groupId</groupId>
<artifactId>code</artifactId>
<version>1.0-SNAPSHOT</version> <dependencies>
<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.11</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.4-beta</version>
<type>pom</type>
</dependency> <!-- https://mvnrepository.com/artifact/junit/junit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency> </dependencies> </project>
pom.xml
package com.amber; import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException; /**
* zookeeper
* 连接zookeeper
* 创建znode
* 获取znode值
* 断开链接
*/
public class ZookeeperDemo {
private ZooKeeper zookeeper; /**
* 链接zookeeper
* @return
* @throws IOException
*/
public ZooKeeper zkConnect( ) throws IOException {
//zookeeper的ip:端口
String path = "127.0.0.1:2181";
//第二个参数是超时时间，第三个参数是设置观察者,现在可以先不管
zookeeper = new ZooKeeper(path, 20 * 1000, null);
return zookeeper;
} /**
* 创建znode节点
* @param path znode的路径
* @param value znode的值
* @param watcher
* @param node //创建node的模式
* @throws KeeperException
* @throws InterruptedException
*/
public void createZnode(String path, byte[] value, Watcher watcher, CreateMode node ) throws KeeperException, InterruptedException {
zookeeper.create(path, value, ZooDefs.Ids.OPEN_ACL_UNSAFE, node);
} /**
* 通过path获得znode的值
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public String getZnodeValue(String path ) throws KeeperException, InterruptedException {
//第二个值是代表是否开启监听，这里还是先不管.第三个参数就是结构体
byte[] data = zookeeper.getData(path, false, new Stat());
return new String(data);
} public void close() {
try {
if (zookeeper != null) {
zookeeper.close();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
} public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
ZookeeperDemo zookeeperDemo = new ZookeeperDemo();
//获取连接
ZooKeeper zooKeeper = zookeeperDemo.zkConnect();
//创建znode
zookeeperDemo.createZnode("/amber", "hahaha".getBytes(), null, CreateMode.PERSISTENT);
//获取znode的值
String znodeValue = zookeeperDemo.getZnodeValue("/amber");
System.out.println(znodeValue); zookeeperDemo.close(); }
}
ZookeeperDemo
通过上面的代码就可以实现通过java代码操控zookeeper.但是你可能有疑惑的是
- Create.PERSISTENT是什么
- watcher是什么
Znode的四种类型
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
// package org.apache.zookeeper; import org.apache.yetus.audience.InterfaceAudience.Public;
import org.apache.zookeeper.KeeperException.BadArgumentsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; @Public
public enum CreateMode {
PERSISTENT(0, false, false),
PERSISTENT_SEQUENTIAL(2, false, true),
EPHEMERAL(1, true, false),
EPHEMERAL_SEQUENTIAL(3, true, true); private static final Logger LOG = LoggerFactory.getLogger(CreateMode.class);
private boolean ephemeral;
private boolean sequential;
private int flag; private CreateMode(int flag, boolean ephemeral, boolean sequential) {
this.flag = flag;
this.ephemeral = ephemeral;
this.sequential = sequential;
} public boolean isEphemeral() {
return this.ephemeral;
} public boolean isSequential() {
return this.sequential;
} public int toFlag() {
return this.flag;
} public static CreateMode fromFlag(int flag) throws KeeperException {
switch(flag) {
case 0:
return PERSISTENT;
case 1:
return EPHEMERAL;
case 2:
return PERSISTENT_SEQUENTIAL;
case 3:
return EPHEMERAL_SEQUENTIAL;
default:
String errMsg = "Received an invalid flag value: " + flag + " to convert to a CreateMode";
LOG.error(errMsg);
throw new BadArgumentsException(errMsg);
}
}
}
CreateMode
znode分四种类型
PERSISTENT 持久节点 对应命令 create path valuePERSISTENT_SEQUENTIAL 顺序自动编号持久化节点，这种节点会根据当前已存在的节点数自动加 1 （有序持久节点）create -s path value
EPHEMERAL 临时节点， 客户端session超时这类节点就会被自动删除 create -e path valueEPHEMERAL_SEQUENTIAL 临时自动编号节点 (临时有序节点) create -s -e path valueCreateMode是一个枚举类型，里面有四个对象分别是.表示的就是Znode的类型
PERSISTENT(0, false, false), //持久节点
PERSISTENT_SEQUENTIAL(2, false, true), //有序节点
EPHEMERAL(1, true, false), //临时节点 只存在本次session中，当服务器重启后就会不见
EPHEMERAL_SEQUENTIAL(3, true, true); //有序临时节点 重启后数据不见
在创建持久节点(PERSISTENT)的时候，应该注意因为znode的path是不允许重复的，因此在创建持久节点之前，应先判断节点是否存在。但是持久有序节点（PERSISTENT_SEQUENTIAL）会自动在/path后面跟上dataVersion序号
if (zooKeeper.exists(path, false) == null) {
zookeeperWatchDemo.createZnode(path, value.getBytes(), null, CreateMode.PERSISTENT);
}
watcher
Zookeeper支持发布订阅功能，引入了watcher机制进行监听。当数据进行变动以后，可以及时通知客户端，数据进行了变动,并且把相应的时间通知给Watcher的Client。
watcher的特性:
- Watcher一次性触发器：只能监听一次，是一个一次性的动作,如果需要监听多次，那么应该递归
- 可以使用系统默认的watcher，也可以自定义Watcher.自定义Watcher必须实现
org.apache.zookeeper.Watcher接口
- Zookeeper的getData()，getChildren()，exists()都可以设置Watch选项。当watch为false的时候，或者watcher为null的时候代表不开启watch
new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
triggerWatch(path);
}
package com.amber; import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat; import java.io.IOException; public class ZookeeperWatchDemo {
private ZooKeeper zookeeper;
private String oldValue = "";
private String newValue = "";
public ZooKeeper zkConnect( ) throws IOException {
String path = "127.0.0.1:2181";
zookeeper = new ZooKeeper(path, 20 * 1000, null);
return zookeeper;
} public void createZnode(String path, byte[] value, Watcher watcher, CreateMode node ) throws KeeperException, InterruptedException {
zookeeper.create(path, value, ZooDefs.Ids.OPEN_ACL_UNSAFE, node);
} public String getZnodeValue(final String path ) throws KeeperException, InterruptedException {
byte[] data = zookeeper.getData(path, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
triggerWatch(path);
}
}, new Stat());
oldValue = new String(data);
return new String(data);
} public boolean triggerWatch (String path) {
byte[] data = new byte[0];
try {
data = zookeeper.getData(path, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
triggerWatch(path);
}
}, new Stat());
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
newValue = new String(data);
if (oldValue.equals(newValue)) {
System.out.println("on change");
return false;
} else {
System.out.println("oldvalue: " + oldValue + "new value: " + newValue);
oldValue = newValue;
return true;
}
} public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
//创建
ZookeeperWatchDemo zookeeperWatchDemo = new ZookeeperWatchDemo();
ZooKeeper zooKeeper = zookeeperWatchDemo.zkConnect();
String path = "/amberas";
String value = "hahahahaha";
if (zooKeeper.exists(path, false) == null) {
zookeeperWatchDemo.createZnode(path, value.getBytes(), null, CreateMode.PERSISTENT);
} String znodeValue = zookeeperWatchDemo.getZnodeValue(path);
System.out.println(znodeValue); Thread.sleep(1000 * 60 * 50);
}
}
ZookeeperWatchDemo