zookeeper案例

该案例主要也是客户端监听原理，客户端监听服务器的上下线情况

先在集群上创建/servers 节点（用于存储连接的服务器的主机和该服务器的节点数）相当于zookeeper集群

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

服务端：

（1）先获取zookeeper连接

    创建类对象

该类为我们创建的服务端类：

        DistributeServer server = new DistributeServer();

    获取zookeeper连接：

自己创建连接方法：

    private void getconnect() throws IOException {
       zk = new ZooKeeper(connectstring, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            }
        });

    }

让后server对象在main函数中调用

（2）注册服务器到zookeeper集群：

注册是需要注册到zookeeper集群的/servers路径下，需要指定参数进行创建

private void regestServer(String hostname) throws InterruptedException, KeeperException {
zk.create(parentNode+"/"+hostname,hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    //  需要创建有序的临时节点所以-e（暂时） -s（有序）
        System.out.println("服务器"+hostname+"已注册连接");
    }

（3）业务逻辑（睡眠）：

    private void business() throws InterruptedException {
    Thread.sleep(Long.MAX_VALUE);
    }

服务端代码如下：

package com.tangxiaocong.case1;
import org.apache.zookeeper.*;
import java.io.IOException;
/**
 * @Date 2023/8/10 19:06
 * @Author 
 */
public class DistributeServer {
   private static String connectstring="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private static int sessionTimeout=2000;
    private ZooKeeper zk =null;
    private String parentNode = "/servers";
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //获取zk连接
        //创建
        DistributeServer server = new DistributeServer();
        server.getconnect();
        //注册服务器到zk集群
        //注册是需要在/servers节点下创建所开启的服务器的路径
        server.regestServer(args[0]);
        //业务逻辑（实际是延时让它睡觉---不然会注册完成就关闭）
        server.business();
    }
    private void business() throws InterruptedException {
    Thread.sleep(Long.MAX_VALUE);
    }
    private void regestServer(String hostname) throws InterruptedException, KeeperException {
zk.create(parentNode+"/"+hostname,hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    //  需要创建有序的临时节点所以-e（暂时） -s（有序）
        System.out.println("服务器"+hostname+"已注册连接");
    }
    private void getconnect() throws IOException {
       zk = new ZooKeeper(connectstring, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            }
        });
    }
}

客户端：

（1）获取zookeeper的连接：

    先创建客户端对象，在进行构建获取zookeeper连接的方法，本方法对process方法进行了重写，填写了再发生上下线的运行逻辑

 private void getConnect() throws IOException {
       zk= new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            try {
                getServerList();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    });
    }

（2）监听/servers下边的子节点的增减：

    构建方法client.getServerList()来进行监听：

代码逻辑就是通过getChildren()方法获取指定目录下的所有子目录并开启监听

再进行遍历，把遍历结果封装到一个集合中，最后进行输出

 private void getServerList() throws InterruptedException, KeeperException {
        List<String> children = zk.getChildren("/servers", true);
       //该方法会获取指定路径下的所有子节点
        //true 会走初始化中的watch 也可以自己创建watch
        //把所有的服务器都封装到一个集合
        ArrayList<String> list = new ArrayList<>();
        for (String child : children) {
            byte[] data = zk.getData("/servers" +"/"+ child, false, null);
            //上边已经便利到一个服务器对象，再进行添加
            list.add(new String(data));
        }
        System.out.println(list);
    }

（3）业务逻辑同服务端不在赘述。

客户端代码如下：

package com.tangxiaocong.case1;
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;
/**
 * @Date 2023/8/10 21:27
 * @Author 
 * 客户端的监听功能
 */
public class DistributeClient {
private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
   private int sessionTimeout=2000;
   private ZooKeeper zk=null;
   public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        //获取zk连接
        DistributeClient client = new DistributeClient();
        client.getConnect();
        //监听/servers下边的子节点的增减
        client.getServerList();
       //业务逻辑（睡眠）
       client.business();
    }
    private void business() throws InterruptedException {
   Thread.sleep(Long.MAX_VALUE);
   }
    private void getServerList() throws InterruptedException, KeeperException {
        List<String> children = zk.getChildren("/servers", true);
       //该方法会获取指定路径下的所有子节点
        //true 会走初始化中的watch 也可以自己创建watch
        //把所有的服务器都封装到一个集合
        ArrayList<String> list = new ArrayList<>();
        for (String child : children) {
            byte[] data = zk.getData("/servers" +"/"+ child, false, null);
            //上边已经便利到一个服务器对象，再进行添加
            list.add(new String(data));
        }
        System.out.println(list);
    }
    private void getConnect() throws IOException {
       zk= new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            try {
                getServerList();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    });
    }
}

案例二：ZooKeeper 分布式锁

分布式锁是什么？

日常使用计算机的时候，我们的电脑不会只开一个进程，但是当“进程1”在访问某些资源的时候，不能被其他进程所访问，它就会去获得锁，把她所访问的资源进行锁上，对该资源进行独占。"进程 1"用完该资源以后就将锁释放掉，让其 他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的 访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

锁的实现：

构造函数：

在该类中首先要实现构造方法，构造方法与类名相同，在该方法中需要获取连接，重写process方法，在该方法中实现释放CountDownLatch的类对象，有两种情况，正常连接释放一种，不是正常连接状态，则释放另一种。在构造方法中还要判断是否存在“/locks”路径，存在则正常退出，不存在则创建该路径。

加锁函数：

使用ZooKeeper对象进行创建节点（临时有序），让后获取“/locks”路径下的所有节点序号，对结果进行判断，如果返回的List集合只有一个节点，则直接返回，默认加锁，不用再做监听工作。如果不是只有一个节点，则对List集合进行排序，再获取他的节点名称，通过indexOf函数来获取该名称节点的下标。如果为-1，则数据异常，为0 则为最小节点，则直接退出，进行加锁不需要设置监听，结果为其他则需要设置监听，先设置监听字符串，当状态不发生改变会一致阻塞,只有上锁节点让位后会调用process方法进行释放。

解锁函数：

解锁就是直接删除节点即可

整体代码：

package com.tangxiaocong.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;
/**
 * @Date 2023/8/12 19:56
 * @Author 
 */
public class DistributedLock {
 final    private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";
    final  private int sessionTimeout=2000;
    final    private   ZooKeeper zk;
    private String waitPath;
    private String currentModu;
    //为了程序的健壮性，创建该对象   等待操作
    final   private CountDownLatch waitLach=new CountDownLatch(1);
    final   private CountDownLatch countDownLatch=new CountDownLatch(1);
  public DistributedLock() throws IOException, InterruptedException, KeeperException {
        //获取连接
      zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            //  connectLatch  如果正常连接zk  可以释放
                if (watchedEvent.getState()==Event.KeeperState.SyncConnected){
                    countDownLatch.countDown();
                }
                //检测到删除节点并且是前一个节点则释放waitlatch
                if (watchedEvent.getType()==Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath))
                {
                waitLach.countDown();
                }
            }
        });
      //等待是否正常连接  正常（已）连接会释放  否则阻塞
      countDownLatch.await();
        // 判断是否存在lock锁
        Stat stat = zk.exists("/locks", false);
        if (stat==null)
        {
            //创建该节点
            String s = zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);
        }
    }
    //对zk加锁
    public void zkLock()  {
        //创建临时的带序号的节点
        try {
            currentModu = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            List<String> children = zk.getChildren("/locks", false);
             //如果只有一个节点   则直接获取
            if(children.size()==1)
            {
                return;
            }
            else {
                //排序
                Collections.sort(children);
                //直接从s后边开始   开始的下标就是length的长度
                String substring = currentModu.substring("/locks/".length());
                //通过substring来获取在List集合中的下标位置
                int index = children.indexOf(substring);
                if (index==-1)
                {
                    System.out.println("数据异常");
                }
                else if (index==0)
                {
                    return;
                }
                else {
                    //  需要监听上一个节点
                    waitPath="/locks/"+children.get(index-1);
                    zk.getData(waitPath,true,new Stat());
                    //等待监听
                    waitLach.await();
                    return;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //判断创建的节点是否是最小序号的节点 如果是则获取锁  不是则监听他的前一个节点
    }
    //对zk解锁
    public void unzkLock()
    {
//删除节点
        try {
            //-1  是版本号
            zk.delete(this.currentModu,-1);
        } catch (InterruptedException  | KeeperException e) {
            e.printStackTrace();
        }
    }
}

测试类代码 ：

package com.tangxiaocong.case2;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
/**
 * @Date 2023/8/12 22:31
 * @Author 唐晓聪
 */
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.unzkLock();
                    System.out.println("线程1释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.zkLock();
                System.out.println("线程2启动获得锁");

                    Thread.sleep(5*1000);
                    lock2.unzkLock();
                    System.out.println("线程2释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

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

该案例是直接使用API进行实现分布式锁

实现步骤：

先创建分布式锁对象，new InterProcessMutex(),参数1为所要连接的客户端，参数2为监听路径

参数1传入的为getCuratorFramework()自定义函数，

该函数通过工厂类的方式进行建立连接，返回创建好的客户端，让后start启动客户端

创建完分布式锁对象后创建两个线程，在线程中进行获得锁，释放锁的操作。

代码如下：

package com.tangxiaocong.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;
/**
 * @Date 2023/8/13 20:07
 * @Author 
 */
public class CuratorLockTest {
    public static void main(String[] args) {
        //创建分布式锁1
        //参数1   所连接的客户端 参数2 监听路径
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");
        //创建分布式锁2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");
    //创建线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock1.acquire();
                    System.out.println("thread 1 acquire lock");
               lock1.acquire();
                    System.out.println("thread 1 again acquire lock");
                Thread.sleep(5*1000);
                lock1.release();
                    System.out.println("thread 1 relax lock");
                    lock1.release();
                    System.out.println("thread 1 again relax lock");
                    System.out.println();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.acquire();
                    System.out.println("thread 2 acquire lock");
                    lock2.acquire();
                    System.out.println("thread 2 again acquire lock");
                    Thread.sleep(5*1000);
                    lock2.release();
                    System.out.println("thread 2 relax lock");
                    lock2.release();
                    System.out.println("thread 2 again relax lock");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
    private static CuratorFramework getCuratorFramework() {
        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
        //通过工厂类的方式进行建立连接
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop102:2181,hadoop104:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy)//连接失败后  间隔多少秒下次间隔
                .build();
        client.start();
        System.out.println("zookeeper  success start  !!!!!");
        return client;
    }
}