《快学BigData》--Zookeeper 总结（D）（24）

Zookeeper 的客户端的API的使用

1 -1 ) 、基本方法

功能 描述

create 在本地目录树中创建一个节点

delete 删除一个节点

exists 测试本地是否存在目标节点

get/set data 从目标节点上读取?/?写数据

get/set ACL 获取?/?设置目标节点访问控制列表信息

get children 检索一个子节点上的列表

sync 等待要被传送的数据

1-2 )、 增删改查znode数据

先在编辑器中导入zookeeper-3.4.5\lib的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);

}

}