1、zookeeper事件监听（watcher）

1.1、watcher概念 ：

• zookeeper提供了数据的发布/订阅功能，多个订阅者可同时监听某一特定主题对象，当该主题对象的自身状态发生变化时例如节点内容改变、节点下的子节点列表改变等，会实时、主动通知所有订阅者。

• zookeeper采用了 Watcher机制实现数据的发布订阅功能。该机制在被订阅对象发生变化时会异步通知客户端，因此客户端不必在 Watcher注册后轮询阻塞，从而减轻了客户端压力

• watcher机制事件上与观察者模式类似，也可看作是一种观察者模式在分布式场景下的实现方式。

1.1.1、watcher架构：

watcher实现由三个部分组成

• zookeeper服务端

• zookeeper客户端

• 客户端的ZKWatchManager对象

客户端首先将 Watcher注册到服务端，同时将 Watcher对象保存到客户端的watch管理器中。当Zookeeper服务端监听的数据状态发生变化时，服务端会主动通知客户端，接着客户端的 Watch管理器会触发相关 Watcher来回调相应处理逻辑，从而完成整体的数据 发布/订阅流程

1.1.2、watcher特性：

1.1.3、watcher接口设计 ：

Watcher是一个接口，任何实现了Watcher接口的类就算一个新的Watcher。Watcher内部包含了两个枚举类：KeeperState、EventType

1.2、Watcher事件类型(EventType)：

EventType是数据节点发生变化时对应的通知类型 （节点的增删改查）。EventType变化时KeeperState永远处于SyncConnected通知状态下；当keeperState发生变化时，EventType永远为None。

1.3、Watcher通知状态(KeeperState)：

KeeperState是客户端与服务端连接状态发生变化时对应的通知类型。

1.4、捕获相应的事件：

以node-x节点为例，说明调用的注册方法和可用监听事件间的关系：

1.5、注册watcher的方法

事件类型为：None

public class ZkConnectionWatcher implements Watcher {//zookeeper链接地址private static final String IP = "192.168.1.7:2181";//计数器对象，阻塞主线程，等待其他线程private static CountDownLatch countDownLatch = new CountDownLatch(1);@Override//当服务端有事件消息时，通过watchedEvent，传递给客户端public void process(WatchedEvent watchedEvent) {Event.KeeperState state = watchedEvent.getState();if(state == Event.KeeperState.SyncConnected){// 正常System.out.println("正常连接");}else if (state == Event.KeeperState.Disconnected){// 可以用Windows断开虚拟机网卡的方式模拟// 当会话断开会出现，断开连接不代表不能重连，在会话超时时间内重连可以恢复正常System.out.println("断开连接");}else if (state == Event.KeeperState.Expired){// 如果会话超时了，那么就需要重新创建zookeeper链接System.out.println("会话超时");}else if (state == Event.KeeperState.AuthFailed){// 在操作的时候权限不够会出现System.out.println("授权失败");}countDownLatch.countDown();}public static void main(String[] args) throws Exception {// 5000为会话超时时间ZooKeeper zooKeeper = new ZooKeeper(IP, 5000, new ZkConnectionWatcher());countDownLatch.await();// 模拟授权失败zooKeeper.addAuthInfo("digest1","itcast1:123451".getBytes());byte[] data = zooKeeper.getData("/hadoop", false, null);System.out.println(new String(data));TimeUnit.SECONDS.sleep(50);}
}

1.6、watcher机制exists

//是否使用连接对象的监听器
exists(String path, boolean b)
//使用自定义监听器
exists(String path, Watcher w)//对节点操作后，返回的状态。
NodeCreated：**节点**创建
NodeDeleted：**节点**删除
NodeDataChanged：**节点**内容

public class ZkExistsWathcer {private static ZooKeeper zookeeper ;private static  CountDownLatch cou ;@Beforepublic void before()throws Exception{//计数器对象cou=new CountDownLatch(1);zookeeper = new ZooKeeper("192.168.1.7", 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(event.getState()==Event.KeeperState.SyncConnected){System.out.println("链接创建成功！");//释放阻塞cou.countDown();}System.out.println("节点地址-->"+event.getPath());System.out.println("节点发生变化类型-->"+event.getType());}});//阻塞主线程，等待链接对象创建成功cou.await();}@Afterpublic void after() throws Exception {zookeeper.close();}/***使用连接对象的监视器。new ZooKeeper时，第三个参数new Watcher()内部方法* 测试：开启服务，然后在服务器上，对wathcer节点进行增删改的操作。这里就能监听到。* 注意：每次监听，只能监听一次操作。下次操作必须重启，再操作还能监听到。*/@Testpublic void test0() throws Exception {zookeeper.exists("/wathcer",true);Thread.sleep(50000);}/*** 自定义监听器。* 测试：同上* 注意：同上*/@Testpublic void test1() throws Exception {zookeeper.exists("/wathcer1", new Watcher() {@Overridepublic void process(WatchedEvent event) {if(event.getState()==Event.KeeperState.SyncConnected){System.out.println("链接创建成功！");//释放阻塞cou.countDown();}System.out.println("节点地址-->"+event.getPath());System.out.println("节点发生变化类型-->"+event.getType());}});Thread.sleep(50000);}/*** wathcer本身机制是一次性的。可以采用通知完后，再调用zookeeper.exists（）方法。来回复用*/@Testpublic void test2() throws Exception {Watcher watcher = new Watcher() {@Overridepublic void process(WatchedEvent event) {try {System.out.println("节点地址-->" + event.getPath());System.out.println("节点发生变化类型-->" + event.getType());//通知完成后，在调用zookeeper.exists方法，重新对该节点监听Stat exists = zookeeper.exists("/wathcer1", this);} catch (Exception e) {e.printStackTrace();}}};zookeeper.exists("/wathcer1", watcher);Thread.sleep(50000);}/*** 一个节点有多个监听器对象。只要对节点进行操作，多个监听器都会执行。*/@Testpublic void test3() throws Exception {zookeeper.exists("/wathcer1", new Watcher() {@Overridepublic void process(WatchedEvent event) {System.out.println("监听器1");}});zookeeper.exists("/wathcer1", new Watcher() {@Overridepublic void process(WatchedEvent event) {System.out.println("监听器2");}});Thread.sleep(50000);}
}

1.7、watcher机制getData

//是否使用连接对象的监听器
getData(String path, boolean b, Stat stat)
//自定义监听器
getData(String path, Watcher w, Stat stat)//对节点操作反正的状态值
NodeDeleted：**节点**删除
NodeDataChange：**节点**内容发生变化

测试：仿照上面exists的代码。只需将 zooKeeper.exists（path,boolean）换成 zooKeeper.getData(path,boolean,stat)即可。此处测试不做过多赘述。

1.8、watcher机制getChildren

//是否使用连接对象的监听器
getChildren(String path, boolean b)
//使用自定义监听器
getChildren(String path, Watcher w)//对节点的子节点操作后，返回的状态。
NodeChildrenChanged：子节点发生变化
NodeDeleted：子节点删除

测试：仿照上面exists的代码。只需将 zooKeeper.exists（path,boolean）换成 zooKeeper.getChildren(path,boolean)即可。此处测试不做过多赘述。

2.配置中心案例

2.1、配置中心思路

若数据库的用户名和密码改变时候，还需要重新加载媛存，比较麻烦，通过 Zookeeper可以轻松完成,当数据库发生变化时自动完成缓存同步，使用事件监听机制可以做出一个简单的配置中心

设计思路

1. 连接zookeeper服务器

2. 读取zookeeper中的配置信息，注册watcher监听器，存入本地变量

3. 当zookeeper中的配置信息发生变化时，通过watcher的回调方法捕获数据变化事件

4. 重新获取配置信息

2.2、初始化信息

create /config "Mysql config"
create /config/url "192.168.1.7:3306"
create /config/username "root"
create /config/password "123456"

2.3、编写配置中心案例代码

/*** 配置中心案例*/
public class ZkWatcherConfig  implements Watcher {//zookeeper链接地址private static final String IP = "192.168.1.7:2181";//计数器对象，阻塞主线程，等待其他线程private CountDownLatch countDownLatch = new CountDownLatch(1);//zookeeper对象private static  ZooKeeper zookeeper ;//数据库连接（提供get/set方法）private String url;private String username;private String password;//构造方法，对象创建的时候，就初始化数据链接的信息public ZkWatcherConfig(){initValue();}@Overridepublic void process(WatchedEvent event) {try {Event.KeeperState state = event.getState();if(event.getType()== Event.EventType.None) {if (state == Event.KeeperState.SyncConnected) {System.out.println("正常连接");countDownLatch.countDown();} else if (state == Event.KeeperState.Disconnected) {System.out.println("断开连接");} else if (state == Event.KeeperState.Expired) {System.out.println("会话超时");} else if (state == Event.KeeperState.AuthFailed) {System.out.println("授权失败");}}else if(event.getType()== Event.EventType.NodeDataChanged){//如果配置信息发生变化，那么重新加载initMethod方法initValue();}else if(event.getType()==Event.EventType.NodeDeleted){System.out.println("如果谁删了配置信息，请揪出来祭天！！！");}}catch (Exception e){e.printStackTrace();}}public void initValue() {try{//连接zookeeper//一点要判断zookeeper是否连接，如果连接，就不用再连接了。// 你如果连接特别慢。那么下面的测试就失败，因为每次都连接的话，非常耗时。还没重新连接，测试的循环又开始下一轮了。if(zookeeper==null)zookeeper=new ZooKeeper(IP,5000,this);//阻塞线程等待连接创建成功countDownLatch.await();this.url=new String(zookeeper.getData("/config/url", true, null));this.username=new String(zookeeper.getData("/config/username", true, null));this.password=new String(zookeeper.getData("/config/password", true, null));System.out.println("配置信息，加载一次");}catch (Exception e){e.printStackTrace();}}public static void main(String[] args) throws InterruptedException {ZkWatcherConfig zkConfigWatcher = new ZkWatcherConfig();for(int i=1;i<=10;i++){Thread.sleep(2000);System.out.println("当前url："+zkConfigWatcher.getUrl());System.out.println("当前user："+zkConfigWatcher.getUsername());System.out.println("当前password："+zkConfigWatcher.getPassword());System.out.println("--------------");}}public String getUrl() {return url;}public void setUrl(String url) {this.url = url;}public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}
}

3.全局唯一ID

问题：

在过去的单库单表型系统中，通常第可以使用数据库字段自带的auto_ increment属性来自动为每条记录生成个唯一的ID。但是分库分表后，就无法在依靠数据库的auto_ increment属性来唯一标识一条记录了。此时我们就可以用zookeeper在分布式环境下生成全局唯一ID

public class ZKUniqueIDWatcher implements Watcher {private CountDownLatch countDownLatch=new CountDownLatch(1);//zookeeper信息private static final String IP="192.168.1.7:2181";private static ZooKeeper zooKeeper;//唯一IDprivate String uniqueId="/id";public ZKUniqueIDWatcher(){try{//连接zookeeperif(zooKeeper==null)zooKeeper=new ZooKeeper(IP,5000,this);//阻塞线程等待连接创建成功countDownLatch.await();}catch (Exception e){e.printStackTrace();}}@Overridepublic void process(WatchedEvent event) {try{if(event.getType()== Event.EventType.None) {if (event.getState() == Event.KeeperState.SyncConnected) {System.out.println("连接创建成功");countDownLatch.countDown();} else if (event.getState() == Event.KeeperState.Disconnected)System.out.println("断开连接");else if (event.getState() == Event.KeeperState.Expired) {System.out.println("会话超时");zooKeeper = new ZooKeeper(IP, 5000, new ZKUniqueIDWatcher());} else if (event.getState() == Event.KeeperState.AuthFailed) {System.out.println("认证失败");}}}catch (Exception e){e.printStackTrace();}}private String getUniqueId(){try{//创建临时有序结点/*** arg1:创建的节点* agr2:节点的数据信息（这里只生成唯一ID，不需要数据）* arg3:创建节点的权限列表* arg4:CreateMode.EPHEMERAL_SEQUENTIAL 生成临时有序ID*/return zooKeeper.create(uniqueId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);}catch (Exception e){e.printStackTrace();}return null;}public static void main(String[] args) throws InterruptedException {ZKUniqueIDWatcher zkUniqueIDWatcher= new ZKUniqueIDWatcher();for(int i=1;i<=10;i++){System.out.println(zkUniqueIDWatcher.getUniqueId().substring(3));}}
}

4、zookeeper分布式锁

设计思路 ：

1. 每个客户端往/Locks下创建临时有序节点/Locks/Lock_，创建成功后/Locks下面会有每个客户端对应的节点，如/Locks/Lock_000000001

2. 客户端取得/Locks下子节点，并进行排序，判断排在前面的是否为自己，如果自己的锁节点在第一位，代表获取锁成功

3. 如果自己的锁节点不在第一位，则监听自己前一位的锁节点。例如，自己锁节点Lock_000000002，那么则监听Lock_000000001

4. 当前一位锁节点(Lock_000000001)对应的客户端执行完成，释放了锁，将会触发监听客户端(Lock_000000002)的逻辑

5. 监听客户端重新执行第2步逻辑，判断自己是否获得了锁

代码实现：

/*** zookeeper分布式锁* 1、准备工作* 2、在构造方法中，创建zookeeper连接* 3、创建锁* 4、获取锁* 5、释放锁* 6、内部类测试分布式锁*/
public class MyLock {//zookeeper连接String IP="192.168.1.7:2181";//程序计数器，用于zookeeper创建时，阻塞主线程等待zookeeper创建完成CountDownLatch downLatch=new CountDownLatch(1);//zookeeper配置信息ZooKeeper zooKeeper;//父节点private static final String LOCK_ROOT_PATH="/Locks";//规定所有子节点格式private static final String LOCK_NODE_PATH="Lock_";//生成的最终路径private String lockpath;//在构造方法中，打开zookeeper连接public MyLock(){try {zooKeeper=new ZooKeeper(IP, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(event.getType()==Event.EventType.None){if(event.getState()==Event.KeeperState.SyncConnected){System.out.println("zookeeper连接成功");//计数器减一，释放阻塞downLatch.countDown();}}}});//阻塞主线程，等待zookeeper创建成功downLatch.await();}catch (Exception e){e.printStackTrace();}}//获取锁public void acquireLock() throws Exception {//创建锁节点createLock();//尝试获取锁attempLock();}//创建锁private void createLock()throws Exception {//判断Locks是否存在，不存在就创建Stat stat = zooKeeper.exists(LOCK_ROOT_PATH, false);if(stat==null){//创建/Locks为持久化节点zooKeeper.create(LOCK_ROOT_PATH,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}//创建临时有序子节点lockpath=zooKeeper.create(LOCK_ROOT_PATH+"/"+LOCK_NODE_PATH,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println("临时有序子节点创建成功："+lockpath);}//监听上一个节点是否被删除private  Watcher watcher=new Watcher() {@Overridepublic void process(WatchedEvent event) {//10、监听节点删除事件if(event.getType()==Event.EventType.NodeDeleted){//11、监听上一个节点释放锁。synchronized (this){//12、释放当前节点的线程watcher.notifyAll();}}}};/*** 尝试获取锁* 序号大致表现线程运行步骤。*/private void attempLock() throws Exception  {//1、获取Locks节点下的所有子节点List<String> children = zooKeeper.getChildren(LOCK_ROOT_PATH, false);//2、对子节点进行排序Collections.sort(children);//3、获取当前节点的位置  /Locks/Lock_0000000000,截取后，int index=children.indexOf(lockpath.substring(LOCK_ROOT_PATH.length()+1));//4、如果index是0，那么当前节点在集合中就是第一个，就能获取锁if(index==0){System.out.println("获取锁成功");return;}else{//5、else说明当前节点前面还有锁，那么就监听当前节点的前一个节点的删除事件。String prePath=children.get(index-1);Stat stat = zooKeeper.exists(LOCK_ROOT_PATH + "/" + prePath, watcher);if(stat==null){//6、如果stat为null，说明前一个节点刚好释放锁，当前节点只需重新走一遍当前方法。获取锁attempLock();}else {//7、如果不为null，那么前一个节点，不是占着锁，就是还没获取到锁。那么就让当前线程等待，等待前一个节点释放锁//8、同步块传递当前的监视器对象。synchronized (watcher){//9、监视器等待。阻塞当前线程。（10-12步骤，请看上面wathcer方法）watcher.wait();}//13、走到这里，前一个节点释放，当前线程也在监视器方法中释放。那么就当前节点重新获取一遍锁。attempLock();}}}//释放锁public void releaseLock() throws Exception {//删除临时有序节点。-1代表不指定版本号zooKeeper.delete(lockpath,-1);zooKeeper.close();System.out.println("锁释放");}}//测试分布式锁类。
class TicketSeller {private void sell(){System.out.println("开始售票");int sleepMills=5000;try{//模拟复杂逻辑Thread.sleep(sleepMills);}catch (Exception e){e.printStackTrace();}System.out.println("售票结束");}private void sellWithLock() throws Exception{MyLock zkLock=new MyLock();zkLock.acquireLock();sell();zkLock.releaseLock();}public static void main(String[] args) throws Exception {TicketSeller seller = new TicketSeller();for(int i=0;i<10;i++){seller.sellWithLock();}}
}