1. 基本使用
org.apache.zookeeper.Zookeeper是客户端入口主类,负责建立与server的会话。它提供了表1所示几类主要方法:
功能 | 描述 |
create | 在本地目录树中创建一个节点 |
delete | 删除一个节点 |
exists | 测试本地是否存在目标节点 |
get/set data | 从目标节点上读取/写数据 |
get/set ACL | 获取/设置目标节点访问控制列表信息 |
get children | 检索一个子节点上的列表 |
sync | 等待要被传送的数据 |
表1:ZooKeeper API描述
2.pom.xml配置依赖包
在本篇文章中共用一个pom.xml
<dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.10</version> </dependency> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.6</version> </dependency> <dependency> <groupId>jline</groupId> <artifactId>jline</artifactId> <version>0.9.94</version> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <dependency> <groupId>io.netty</groupId> <artifactId>netty</artifactId> <version>3.9.4.Final</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.6.1</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.6.1</version> </dependency> </dependencies>
3. 增删改查demo
zookeeper上的znode节点增删改查demo
package cn.itcast.zk; import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import org.junit.Before; import org.junit.Test; import java.util.List; /** * @author y15079 * @create 2018-03-16 16:53 * @desc 简单的增删改查demo **/ public class SimpleZkClient { private static final String connectString = "mimi1:2181,mimi2:2181,mini3:2181"; private static final int sessionTimeout = 2000; ZooKeeper zkClient = null; @Before public void init() throws Exception { zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() { public void process(WatchedEvent watchedEvent) { //收到事件通知后的回调函数(应该是我们自己的事件处理逻辑) System.out.println(watchedEvent.getType() + "---" + watchedEvent.getPath()); try { zkClient.getChildren("/", true); } catch (Exception e) { } } }); } /** * 数据的增删改查 */ //创建数据节点到zk中 @Test public void testCreate(String[] args) throws Exception{ //参数1,要创建的节点的路径 参数2:节点数据 参数3:节点的权限 参数4:节点的类型 String node = zkClient.create("/idea", "hellozk".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); //上传的数据可以是任何类型,但都要转成byte[] } //判断znode是否存在 public void testExist() throws Exception{ //节点元数据 Stat stat = zkClient.exists("/idea", false); System.out.println(stat == null ? "not exist": "exist"); } //获取子节点 @Test public void getChildren() throws Exception{ List<String> children = zkClient.getChildren("/", true); for (String child: children){ System.out.println(child); } Thread.sleep(Long.MAX_VALUE);//真正运行时可以注释 } //获取znode的数据 @Test public void getData() throws Exception{ byte[] data = zkClient.getData("/idea",false, null); System.out.println(new String(data)); } //删除znode的数据 @Test public void deleteZnode() throws Exception{ //参数2:指定要删除的版本,-1表示删除所有版本 zkClient.delete("/idea", -1); } //设置znode @Test public void setData() throws Exception{ zkClient.setData("/app1","hello".getBytes(), -1); byte[] data = zkClient.getData("/app1", false, null); System.out.println(new String(data)); } }
3. Zookeeper 应用案例(分布式共享锁 || 分布式应用HA)
3.1 分布式共享锁的简单实现
package cn.itcast.zkclock; import org.apache.zookeeper.*; import java.util.Collections; import java.util.List; import java.util.Random; /** * @author y15079 * @create 2018-03-17 18:40 * @desc 分布式共享锁 单线程的 **/ public class DistributedClientLock { //会话超时 private static final int SESSION_TIMEOUT = 5000; //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; //记录client创建的子节点路径 private volatile String thisPath; /** * 连接zk * @throws Exception */ public void connectZookeeper() throws Exception{ zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() { public void process(WatchedEvent watchedEvent) { try { //判断事件类型,此处只处理子节点变化事件 if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged && watchedEvent.getPath().equals("/" + groupNode)){ //获取子节点,并对父节点进行监听, 开始监听,监听锁是否有变化 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); String thisNode = thisPath.substring(("/" + groupNode + "/").length()); //获取父目录下的子节点路径,不带父目录 //去比较是否自己是最小id, 约定id最小最先拿到锁 Collections.sort(childrenNodes); if (childrenNodes.indexOf(thisNode) == 0){ //indexOf是索引的意思,查看子节点路径在所有子节点中的索引是否为0,为0则证明子节点id最小,最先获取锁 //访问共享资源处理业务,并且在处理完成之后删除锁 doSomething(); //删除锁,监听到,暂时不会调用watcher里面的process方法 //重新注册一把新的锁,不懂为什么注册新的锁 thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //在上面创建好锁后才调用watcher里面的process方法 } } } catch (Exception e) { } } }); //程序一进来就先注册一把锁到zk上,即创建子节点,还没有开始监听,zk watcher里面的process方法不会执行 thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //wait一小会,便于观察 Thread.sleep(new Random().nextInt(1000)); //从zk的锁目录下,获取所有子节点,并且监听子节点的变化, 开始监听,监听锁是否有变化 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); //列表中只有一个子节点,那肯定就是thisPath,说明client获得锁 if (childrenNodes.size() == 1){ doSomething(); //删除锁,监听到,暂时不会调用watcher里面的process方法 //为什么还创建锁?为了下一次client获取锁做好准备 thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //在上面创建好锁后才调用watcher里面的process方法 } } /** * 共享资源的访问逻辑写在这个方法中 * 处理业务逻辑,并释放 * @throws Exception */ private void doSomething() throws Exception{ try { System.out.println("gain lock: " + thisPath); Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); }finally { System.out.println("finished: " + thisPath ); //将thisPath删除,监听thisPath的client将获得通知 //相当于释放锁 zk.delete(this.thisPath, -1); } } public static void main(String[] args) throws Exception{ DistributedClientLock dcl = new DistributedClientLock(); dcl.connectZookeeper(); Thread.sleep(Long.MAX_VALUE); } }
3.2 分布式HA应用的简单实现
客户端实现:
package cn.itcast.zkdist; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import java.util.ArrayList; import java.util.List; /** * @author y15079 * @create 2018-03-17 15:20 * @desc 分布式HA应用 客户端 **/ public class DistributedClient { private static final String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; //注意:加volatile的意义何在?保持一致性 private volatile List<String> serverList; private ZooKeeper zk = null; /** * 创建到zk的客户端连接 * @throws Exception */ public void getConnect() throws Exception{ zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { public void process(WatchedEvent watchedEvent) { //收到事件通知后的回调函数(应该是我们自己的事件处理逻辑) try { //重新更新服务器列表,并且注册了监听 getServerList(); } catch (Exception e) { } } }); } /** * 获取服务器信息列表 * @throws Exception */ public void getServerList() throws Exception{ //获取服务器子节点信息,并且对父节点进行监听 List<String> children = zk.getChildren(parentNode, true); //先创建一个局部的list来存服务器信息 List<String> servers = new ArrayList<String>(); for (String child: children){ //child只是子节点的节点名 byte[] data = zk.getData(parentNode + "/" + child, false, null); servers.add(new String(data)); } //把servers赋值给成员变量serverList,已提供给各业务线程使用 serverList = servers; //打印服务器列表 System.out.println(serverList); } /** * 业务功能 * @throws Exception */ public void handleBusiness() throws Exception{ System.out.println("client start working......"); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { //获取zk连接 DistributedClient client = new DistributedClient(); client.getConnect(); //获取servers的子节点信息(并监听),从中获取服务器信息列表 client.getServerList(); //启动业务功能 client.handleBusiness(); } }
服务端实现:
package cn.itcast.zkdist; import org.apache.zookeeper.*; /** * @author y15079 * @create 2018-03-17 1:12 * @desc 分布式HA应用 服务端 **/ public class DistributedServer { private static final String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; private ZooKeeper zk = null; /** * 创建到zk的客户端连接 * @throws Exception */ public void getConnect() throws Exception{ zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { public void process(WatchedEvent event) { //收到事件通知后的回调函数(应该是我们自己的事件处理逻辑) System.out.println(event.getType() + "---" + event.getPath()); try { zk.getChildren("/",true); } catch (Exception e) { } } }); } /** * 向zk集群注册服务器信息 * @param hostname * @throws Exception */ public void registerServer(String hostname) throws Exception{ String create = zk.create(parentNode+"server", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(hostname + " is online..." + create); } /** * 业务功能 * @throws Exception */ public void handleBusiness(String hostname) throws Exception{ System.out.println(hostname + " start working......"); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { //获取zk连接 DistributedServer server = new DistributedServer(); server.getConnect(); //利用zk连接注册服务器信息 server.registerServer(args[0]); //启动业务功能 server.handleBusiness(args[0]); } }
github地址:
https://github.com/qiushangwenyue/Zookeeper-Java-API-Demo