zookeeper(2) 服务器动态上下线监听和分布式锁案例

案例一：服务器动态上下线监听

某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。

1.服务端代码

package com.atguigu.case1;import org.apache.zookeeper.*;import java.io.IOException;
import java.nio.charset.StandardCharsets;public class DistributeServer {String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";int sessionTimeout = 20000;private ZooKeeper zooKeeper;public static void main(String[] args) throws IOException, InterruptedException, KeeperException {DistributeServer server = new DistributeServer();server.getConnect();server.regist(args[0]);server.business();}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void regist(String hostname) throws InterruptedException, KeeperException {String s = zooKeeper.create("/servers/", hostname.getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println(hostname + "isonline");}private void getConnect() throws IOException {zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {}});}
}

2.客户端代码

package com.atguigu.case1;import org.apache.zookeeper.*;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;public class DistributeClient {String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";int sessionTimeout = 20000000;private ZooKeeper zooKeeper;public static void main(String[] args) throws IOException, InterruptedException, KeeperException {DistributeClient client = new DistributeClient();client.getConnect();client.getServerList();client.business();}private void getServerList() throws InterruptedException, KeeperException {List<String> children = zooKeeper.getChildren("/servers", true);ArrayList<String> servers = new ArrayList<>();for (int i = 0; i < children.size(); i++) {byte[] data = zooKeeper.getData("/servers/" + children.get(i), false, null);servers.add(new String(data));}System.out.println(servers);}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void regist(String hostname) throws InterruptedException, KeeperException {String s = zooKeeper.create("/servers/"+hostname, hostname.getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println(hostname + "isonline");}private void getConnect() throws IOException {zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {try {getServerList();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}});}
}

案例二：分布式锁案例原生方法

package com.atguigu.case2;import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class DistributedLock {String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";int sessionTimeout = 2000;private final ZooKeeper zooKeeper;private CountDownLatch connectLatch  = new CountDownLatch(1);private CountDownLatch waitLatch  = new CountDownLatch(1);private String waitPath;private String currentMode;public DistributedLock() throws IOException, InterruptedException, KeeperException {//获取连接zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {//监听到前一个节点已经下线//connectLatch 如果连上需要释放if(watchedEvent.getState() == Event.KeeperState.SyncConnected) {connectLatch.countDown();}//waitLatch 需要释放if(watchedEvent.getType() == Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)){waitLatch.countDown();}}});//等待zk正常连接后，才往下走connectLatch.await();//判断根节点/locks是否存在Stat stat = zooKeeper.exists("/locks", false);if(stat == null) {//创建根节点zooKeeper.create("/locks","locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}}//加锁public void zklock(){//创建对应的临时的带序号的节点try {currentMode = zooKeeper.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);//判断创建的节点是否是最小的节点，如果是获取到锁；不是，监听前一个序号最小的节点List<String> children = zooKeeper.getChildren("/locks", false);//如果children只有一个值，那就直接获取锁；如果有多个节点，需要判断谁最小if(children.size() == 1) {return;} else {//排序Collections.sort(children);//获取当前节点名称String thisNode = currentMode.substring("/locks/".length());//通过thisNode获取在children集合中的位置int index = children.indexOf(thisNode);if(index == -1) {System.out.println("数据异常");}else if(index == 0) {//就一个节点数据，就可以获取锁了return;}else {//需要监听 前一个节点变化waitPath = "/locks/"+children.get(index-1);zooKeeper.getData(waitPath,true,null);//等待监听waitLatch.await();//接收到之后return;}}} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}public void unZklock(){//删除节点try {zooKeeper.delete(currentMode,-1);} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}
}

测试

package com.atguigu.case2;import org.apache.zookeeper.KeeperException;import java.io.IOException;public class DistributedLockTest {public static void main(String[] args) throws IOException, InterruptedException, KeeperException {DistributedLock lock1 = new DistributedLock();DistributedLock lock2 = new DistributedLock();new Thread(new Runnable(){@Overridepublic void run() {try {lock1.zklock();;System.out.println("线程1启动获取到锁");Thread.sleep(5000);lock1.unZklock();System.out.println("线程1释放锁");} catch (InterruptedException e) {e.printStackTrace();}}}).start();new Thread(new Runnable(){@Overridepublic void run() {try {lock2.zklock();System.out.println("线程2启动获取到锁");Thread.sleep(5000);lock2.unZklock();System.out.println("线程2释放锁");} catch (InterruptedException e) {e.printStackTrace();}}}).start();}
}

案例三：分布式锁案例框架方法

引入包

        <dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.3.0</version></dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-client</artifactId><version>4.3.0</version></dependency>

package com.atguigu.case3;import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;public class CuratorLockTest {public static void main(String[] args) {//创建分布式锁1InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");//创建分布式锁2InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");new Thread(new Runnable(){@Overridepublic void run() {try {lock1.acquire();System.out.println("线程1获取到锁");lock1.acquire();System.out.println("线程1再次获取到锁");lock1.release();System.out.println("线程1释放锁");lock1.release();System.out.println("线程1再次释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();new Thread(new Runnable(){@Overridepublic void run() {try {lock2.acquire();System.out.println("线程2获取到锁");lock2.acquire();System.out.println("线程2再次获取到锁");lock2.release();System.out.println("线程2释放锁");lock2.release();System.out.println("线程2再次释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();}private static CuratorFramework getCuratorFramework() {ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop100:2181,hadoop101:2181,hadoop102:2181").connectionTimeoutMs(2000).sessionTimeoutMs(2000).retryPolicy(policy).build();client.start();System.out.println("zookeeper启动成功");return client;}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/660043.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！