Zookeeper 的客户端的API的使用
1 -1 ) 、基本方法
功能 描述
create 在本地目录树中创建一个节点
delete 删除一个节点
exists 测试本地是否存在目标节点
get/set data 从目标节点上读取?/?写数据
get/set ACL 获取?/?设置目标节点访问控制列表信息
get children 检索一个子节点上的列表
sync 等待要被传送的数据
1-2 )、 增删改查znode数据
先在编辑器中导入zookeeper-3.4.5lib的JAR包，还有一个是zookeeper根目录下的zookeeper-3.4.5.jar
package zookeeperTest;
import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
/**
* zookeeper的基本的操作
*/
public class ZookeeperSimple {
// 会话超时时间，设置为与系统默认时间一致
private static final int SESSION_TIMEOUT = 30000;
// 创建 ZooKeeper 实例
ZooKeeper zk;
// 创建 Watcher 实例
Watcher wh = new Watcher() {
public void process(org.apache.zookeeper.WatchedEvent event) {
System.out.println(event.toString());
}
};
/**
* 初始化zookeeper的操作
*/
private void createZKInstance() throws IOException {
zk = new ZooKeeper("hadoop1:2181", ZookeeperSimple.SESSION_TIMEOUT,
this.wh);
}
/**
* zookeeper的基本的操作
*
* @throws IOException
* @throws InterruptedException
* @throws KeeperException
*/
private void ZKOperations() throws IOException, InterruptedException,
KeeperException {
// 创建节点
zk.create("/testZookeeper", "mytestZookeeper".getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
// 删除节点
System.out
.println(new String(zk.getData("/testZookeeper", false, null)));
// 修改节点
zk.setData("/testZookeeper", "zookeeperModify".getBytes(), -1);
// 获取修改节点的状态
System.out
.println(new String(zk.getData("/testZookeeper", false, null)));
// 删除节点
zk.delete("/testZookeeper", -1);
// 查看删除节点的状态
System.out.println("节点状态：" zk.exists("/testZookeeper", false));
}
/**
* 关闭zookeeper的链接
*/
private void ZKClose() throws InterruptedException {
zk.close();
}
public static void main(String[] args) throws IOException,
InterruptedException, KeeperException {
ZookeeperSimple dm = new ZookeeperSimple();
dm.createZKInstance();
dm.ZKOperations();
dm.ZKClose();
}
}
1-3）、zookeeper服务器上下线动态感知
A）、概述
在分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。
B）、设计思路
需求让客户端动态感知在线的服务器的运行状态。
1-1）、服务器端
获取链接 ( process )
利用zookeeper链接注册服务器的信息
启动业务的逻辑实现
服务器端不需要实现监听
1-2）、客户端
获取链接(process)
获取server上的子节点的信息，并从中获服务器的信息列表
业务启动线程
C）、代码实现
1-1）、服务端代码实现
package zookeeperPerception;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
/**
* 动态感知Server
*/
public class ZookeeperServer {
private String groupNode = "sgroup";
private String subNode = "sub";
/**
* 连接zookeeper
*
* @param address
* server的地址
*/
public void connectZookeeper(String address) throws Exception {
ZooKeeper zk = new ZooKeeper("hadoop1:2181,hadoop2:2181,hadoop3:2181",
5000, new Watcher() {
public void process(WatchedEvent event) {
// 不做处理
}
});
// 在"/sgroup"下创建子节点
// 子节点的类型设置为EPHEMERAL_SEQUENTIAL, 表明这是一个临时节点, 且在子节点的名称后面加上一串数字后缀
// 将server的地址数据关联到新创建的子节点上
String createdPath = zk.create("/" groupNode,
address.getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("create: " createdPath);
}
/**
* server的工作逻辑写在这个方法中 此处不做任何处理, 只让server sleep
*/
public void handle() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
ZookeeperServer as = new ZookeeperServer();
as.connectZookeeper("server1");
as.handle();
}
}
1-2）、客户端代码实现
package zookeeperPerception;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
/**
* 动态感知Client
*/
public class ZookeeperClient {
private String groupNode = "sgroup";
private ZooKeeper zk;
private Stat stat = new Stat();
private volatile List<String> serverList;
/**
* 连接zookeeper
*/
public void connectZookeeper() throws Exception {
zk = new ZooKeeper("hadoop1:2181,hadoop2:2181,hadoop3:2181", 5000,
new Watcher() {
public void process(WatchedEvent event) {
// 如果发生了"/sgroup"节点下的子节点变化事件, 更新server列表, 并重新注册监听
if (event.getType() == EventType.NodeChildrenChanged
&& ("/" groupNode).equals(event.getPath())) {
try {
updateServerList();
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
updateServerList();
}
/**
* 更新server列表
*/
private void updateServerList() throws Exception {
List<String> newServerList = new ArrayList<String>();
// 获取并监听groupNode的子节点变化
// watch参数为true, 表示监听子节点变化事件.
// 每次都需要重新注册监听, 因为一次注册, 只能监听一次事件, 如果还想继续保持监听, 必须重新注册
List<String> subList = zk.getChildren("/" "sgroup0000000039", true);
for (String subNode : subList) {
// 获取每个子节点下关联的server地址
byte[] data = zk.getData("/" "sgroup0000000039", false, stat);
newServerList.add(new String(data, "utf-8"));
}
// 替换server列表
serverList = newServerList;
System.out.println("server list updated: " serverList.size());
}
/**
* client的工作逻辑写在这个方法中 此处不做任何处理, 只让client sleep
*/
public void handle() throws InterruptedException {
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
ZookeeperClient ac = new ZookeeperClient();
ac.connectZookeeper();
ac.handle();
}
}
D）、Alibaba实现动态感知的代码实现
https://github.com/alibaba/taokeeper/tree/master/taokeeper-research
1-4） 、分布式共享锁
A）、概述
在我们自己的分布式业务系统中，可能会存在某种资源，需要被整个系统的各台服务器共享访问，但是只允许一台服务器同时访问
B）、设计思路
先查看所有机器上的锁，查看自己的锁是不是最小的，如果是最小的则会访问最大的那把锁，并把数据同步下来，访问完后把自己的锁删掉后，再会重新注册一把新锁。
C）、代码实现
package lock;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class DistributedClientLock {
// 会话超时
private static final int SESSION_TIMEOUT = 2000;
// zookeeper集群地址
private String hosts = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
private String groupNode = "locks";
private String subNode = "sub";
private boolean haveLock = false;
private ZooKeeper zk;
// 记录自己创建的子节点路径
private volatile String thisPath;
/**
* 连接zookeeper
*/
public void connectZookeeper() throws Exception {
zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
public void process(WatchedEvent event) {
try {
// 判断事件类型，此处只处理子节点变化事件
if (event.getType() == EventType.NodeChildrenChanged
&& event.getPath().equals("/" groupNode)) {
// 获取子节点，并对父节点进行监听
List<String> childrenNodes = zk.getChildren("/"
groupNode, true);
String thisNode = thisPath
.substring(("/" groupNode "/").length());
// 去比较是否自己是最小id
Collections.sort(childrenNodes);
if (childrenNodes.indexOf(thisNode) == 0) {
// 访问共享资源处理业务，并且在处理完成之后删除锁
doSomething();
// 重新注册一把新的锁
thisPath = zk.create("/" groupNode "/"
subNode, null, Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
// 1、程序一进来就先注册一把锁到zk上
thisPath = zk.create("/" groupNode "/" subNode, null,
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// wait一小会，便于观察
Thread.sleep(new Random().nextInt(1000));
// 从zk的锁父目录下，获取所有子节点，并且注册对父节点的监听
List<String> childrenNodes = zk.getChildren("/" groupNode, true);
// 如果争抢资源的程序就只有自己，则可以直接去访问共享资源
if (childrenNodes.size() == 1) {
doSomething();
thisPath = zk.create("/" groupNode "/" subNode, null,
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
}
}
/**
* 处理业务逻辑，并且在最后释放锁
*/
private void doSomething() throws Exception {
try {
System.out.println("gain lock: " thisPath);
Thread.sleep(2000);
// do something
} finally {
System.out.println("finished: " thisPath);
//
// 访问完毕后，需要手动去删除之前的锁节点,-1代表删除所有的版本的 记录信息。
zk.delete(this.thisPath, -1);
}
}
public static void main(String[] args) throws Exception {
DistributedClientLock dl = new DistributedClientLock();
dl.connectZookeeper();
Thread.sleep(Long.MAX_VALUE);
}
}
