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() {
            @Override
            public 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() {
            @Override
            public 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() {
            @Override
            public 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(){
            @Override
            public 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(){
            @Override
            public 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) {
        //创建分布式锁1
        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("线程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(){
            @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) {
                    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;
    }
}