Zookeeper学习---3、服务器动态上下线监听案例、ZooKeeper 分布式锁案例、企业面试真题-阿里云开发者社区

1、服务器动态上下线监听案例

1、需求

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

2、需求分析

3、具体实现

（1）先在集群上创建/servers 节点

（2）在IDEA中创建包名：org.example.zkcase1

（3）服务端向Zookeeper注册代码

package org.example.zkcase1;
import org.apache.zookeeper.*;
import java.io.IOException;
/**
 * @ClassName DistributeServer
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/25 9:12
 * @Version 1.0
 */
public class DistributeServer {
    private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private int sessionTimeout=2000;
    ZooKeeper zk=null;
    private String parentNode="/servers";
    //创建到zk的客户端连接
    public void getConnect() throws IOException {
        zk=new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            }
        });
    }
    //注册服务器
    public void registServer(String hostname) throws InterruptedException, KeeperException {
        String create = zk.create(parentNode + "/servers", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname+"is online"+create);
    }
    //业务功能
    public void business(String hostname) throws InterruptedException {
        System.out.println(hostname+" is working...");
        Thread.sleep(Long.MAX_VALUE);
    }
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //1、获取zk连接
        DistributeServer server=new DistributeServer();
        server.getConnect();
        //2、利用zk连接注册服务器信息
        server.registServer(args[0]);
        //3、启动业务功能
        server.business(args[0]);
    }
}

（4）客户端代码

package org.example.zkcase1;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
/**
 * @ClassName DistributeClient
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/25 9:23
 * @Version 1.0
 */
public class DistributeClient {
    private String connectionString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private int sessionTimeout=2000;
    private ZooKeeper zk;
    private String parentNode="/servers";
    //创建到zk的客户端连接
    public void getConnection() throws IOException {
        zk=new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                //再次启动监听
                try {
                    getServerList();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (KeeperException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }
    //获取服务器列表信息
    public void getServerList() throws InterruptedException, KeeperException {
        //1、获取服务器子节点信息，并且对父节点进行监听
        List<String> children = zk.getChildren(parentNode, true);
        //2、储存服务器信息列表
        ArrayList<String> servers=new ArrayList<>();
        //3、遍历所有节点，获取节点中的主机名称信息
        for (String child : children) {
            byte[] data = zk.getData(parentNode + "/" + child, false, null);
            servers.add(new String(data));
        }
        //4、打印服务器列表信息
        System.out.println(servers);
    }
    //业务功能
    public void business() throws InterruptedException {
        System.out.println("client is working...");
        Thread.sleep(Long.MAX_VALUE);
    }
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //1、获取zk连接
        DistributeClient client=new DistributeClient();
        client.getConnection();
        //2、获取servers的子节点信息，从中获取服务器信息列表
        client.getServerList();
        //3、业务进程启动
        client.business();
    }
}

4、测试

（1）在Linux命令行上操作增加减少服务器

（a）启动 DistributeClient 客户端

（b）在 hadoop102 上 zk 的客户端/servers 目录上创建临时带序号节点

（c）观察IDEA控制台变化

（d）执行删除操作

（e）观察IDEA控制台变化

（2）在IDEA上操作增加和减少服务器

（a）启动 DistributeClient 客户端（如果已经启动过，不需要重启）

（b）启动 DistributeServer 服务

①点击 Edit Configurations…

②在弹出的窗口中（Program arguments）输入想启动的主机，例如，hadoop102

③回到 DistributeServer 的 main 方法，右键，在弹出的窗口中点击 Run“DistributeServer.main()

④观察 DistributeServer 控制台，提示 hadoop102 is working

⑤观察 DistributeClient 控制台，提示 hadoop102 已经上线

2、ZooKeeper 分布式锁案例

什么加做分布式锁呢？

比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

2.1 原生 Zookeeper 实现分布式锁案例

1、分布式锁实现

package org.example.lock;
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;
/**
 * @ClassName DistributedLock
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/25 14:51
 * @Version 1.0
 */
public class DistributedLock {
    //Zookeeper server列表
    private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    //超时时间
    private int sessionTimeout=2000;
    private ZooKeeper zk;
    private String rootNode="locks";
    private String subNode="seq-";
    //当前client等待的子节点
    private String waitPath;
    //Zookeeper连接
    private CountDownLatch connectLatch=new CountDownLatch(1);
    //Zookeeper节点等待
    private CountDownLatch waitLatch=new CountDownLatch(1);
    //当前client创建的子节点
    private String currentNode;
    //和zk服务建立连接，并创建根节点
    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        zk=new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                //连接建立时，打开latch，唤醒wait在该latch上的线程
                if (watchedEvent.getState()==Event.KeeperState.SyncConnected){
                    try {
                        connectLatch.countDown();
                    } catch (Exception e){
                        e.printStackTrace();
                    }
                }
                //发生了waitPath的删除事件
                if (watchedEvent.getType()==Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });
        //等待连接建立
        connectLatch.await();
        //获取根节点状态
        Stat stat = zk.exists("/" + rootNode, false);
        if (stat==null){
            System.out.println("根节点不存在");
            zk.create("/"+rootNode,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        }
    }
    //加锁方法
    public void zkLock(){
        try {
            //在根节点下创建临时顺序节点，返回值为创建的节点路径
            currentNode=zk.create("/"+rootNode+"/"+subNode,null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            //wait一会儿，让结果更加清晰一些
            Thread.sleep(10);
            //注意，没有必要监听"/locks"的子节点的变化情况
            List<String> childrenNodes = zk.getChildren("/" + rootNode, false);
            //列表中只有一个子节点，那肯定就是currentNode，说明client获得锁
            if (childrenNodes.size()==1){
                return;
            }
            else {
                //对根节点下的所有临时顺序节点从小到大排序
                Collections.sort(childrenNodes);
                //当前节点名称
                String thisNode = currentNode.substring(("/" + rootNode + "/").length());
                int index = childrenNodes.indexOf(thisNode);
                if (index==-1){
                    System.out.println("数据异常");
                } else if (index==0) {
                    //index==0,说明thisNode在列表中最小，当前client获得锁
                    return;
                }else {
                    //获得排名比current前一位的节点
                    this.waitPath="/"+rootNode+"/"+childrenNodes.get(index-1);
                    //在waitPath上注册监听器，当waitPath被删除时，Zookeeper会回调监听器的process方法
                    zk.getData(waitPath,true,new Stat());
                    //进入等待锁状态
                    waitLatch.await();
                    return;
                }
            }
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    //解锁方法
    public void zkUnlock(){
        try {
            zk.delete(this.currentNode,-1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        }
    }
}

2、分布式锁测试

package org.example.lock;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
/**
 * @ClassName DistributedLockTest
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/25 15:27
 * @Version 1.0
 */
public class DistributedLockTest {
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //创建分布式锁
        final DistributedLock lock1=new DistributedLock();
        final DistributedLock lock2=new DistributedLock();
        new Thread(new Runnable() {
            @Override
            public void run() {
                //获取锁对象
                try {
                    lock1.zkLock();
                    System.out.println("线程1获取锁");
                    Thread.sleep(5*1000);
                    lock1.zkUnlock();
                    System.out.println("线程1释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                //获取锁对象
                try {
                    lock2.zkLock();
                    System.out.println("线程2获取锁");
                    Thread.sleep(5*1000);
                    lock2.zkUnlock();
                    System.out.println("线程2释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

（2）观察控制台变化

2.2 Curator框架实现分布式锁案例

1、原生的 Java API 开发存在的问题

（1）会话连接是异步的，需要自己去处理。比如使用 CountDownLatch

（2）Watch 需要重复注册，不然就不能生效

（3）开发的复杂性还是比较高的

（4）不支持多节点删除和创建。需要自己去递归

2、Curator 是一个专门解决分布式锁的框架，解决了原生 JavaAPI 开发分布式遇到的问题。

3、Curator 案例实操

（1）添加依赖

<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>

（2）代码实现

package org.example.zkCase2;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
/**
 * @ClassName CuratorLockTest
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/25 15:52
 * @Version 1.0
 */
public class CuratorLockTest {
    private String rootNode="/locks";
    //zookeeper server列表
    private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    //connection超时时间
    private  int connectionTimeOut=2000;
    //session超时时间
    private int sessionTimeOut=2000;
    public static void main(String[] args) {
        new CuratorLockTest().test();
    }
    //测试
    private  void test(){
        //创建分布式锁1
        final InterProcessLock lock1=new InterProcessMutex(getCuratorFramework(),rootNode);
        //创建分布式锁2
        final InterProcessLock lock2=new InterProcessMutex(getCuratorFramework(),rootNode);
        new Thread(new Runnable() {
            @Override
            public 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) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public 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) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
    }
    //分布式锁初始化
    private CuratorFramework getCuratorFramework() {
        //重试策略，初始时间3秒，重试3次
        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
        //通过工厂创建Curator
        CuratorFramework client= CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .connectionTimeoutMs(connectionTimeOut)
                .sessionTimeoutMs(sessionTimeOut)
                .retryPolicy(policy)
                .build();
        //开启连接
        client.start();
        System.out.println("zookeeper 初始化完成...");
        return client;
    }
}

（3）观察控制台变化

3、企业面试真题

3.1 选举机制

半数机制·，超过半数的投票通过，即通过。

1、第一次启动选举机制：

投票过半数时，服务器id大的胜出

2、第二次启动选举规则：

（1）EPOCH大的直接胜出

（2）EPOCH相同，事务id大的胜出

（3）事务id相同，服务器id大的胜出

3.2 生产集群安装多少zk合适

安装奇数台。

生产经验：

10台服务器：3台zk
20台服务器：5台zk
100台服务器：11台zk
200台服务器：11台zk
服务器台数多：
（1）好处：提高可靠性
（2）坏处：提高通信延时

3.3 常用命令

ls、get、create、delete

Zookeeper学习---3、服务器动态上下线监听案例、ZooKeeper 分布式锁案例、企业面试真题

1、服务器动态上下线监听案例

2、ZooKeeper 分布式锁案例

2.1 原生 Zookeeper 实现分布式锁案例

2.2 Curator框架实现分布式锁案例

3、企业面试真题

3.1 选举机制

3.2 生产集群安装多少zk合适

3.3 常用命令

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Zookeeper学习---3、服务器动态上下线监听案例、ZooKeeper 分布式锁案例、企业面试真题

1、服务器动态上下线监听案例

2、ZooKeeper 分布式锁案例

2.1 原生 Zookeeper 实现分布式锁案例

2.2 Curator框架实现分布式锁案例

3、企业面试真题

3.1 选举机制

3.2 生产集群安装多少zk合适

3.3 常用命令

热门文章

最新文章

相关课程

相关电子书

相关实验场景