视频地址：【尚硅谷】大数据技术之Zookeeper 3.5.7版本教程_哔哩哔哩_bilibili

1. 尚硅谷大数据技术Zookeeper教程-笔记01【Zookeeper(入门、本地安装、集群操作)】
2. 尚硅谷大数据技术Zookeeper教程-笔记02【服务器动态上下线监听案例、ZooKeeper分布式锁案例、企业面试真题】
3. 尚硅谷大数据技术Zookeeper教程-笔记03【源码解析-算法基础】
4. 尚硅谷大数据技术Zookeeper教程-笔记04【源码解析-源码详解】

目录

第04章-服务器动态上下线监听案例

P020【020_尚硅谷_zk_案例_服务器动态上下线_需求分析】04:41

P021【021_尚硅谷_zk_案例_服务器动态上下线_服务器注册】10:44

P022【022_尚硅谷_zk_案例_服务器动态上下线_客户端监听】10:01

P023【023_尚硅谷_zk_案例_服务器动态上下线_测试】06:55

第05章-ZooKeeper分布式锁案例

P024【024_尚硅谷_zk_案例_分布式锁_需求分析】05:18

P025【025_尚硅谷_zk_案例_分布式锁_代码实现（上）】09:57

P026【026_尚硅谷_zk_案例_分布式锁_代码实现（下）】16:19

P027【027_尚硅谷_zk_案例_分布式锁_测试】05:39

P028【028_尚硅谷_zk_案例_分布式锁_成熟框架curator】09:44

第06章-企业面试真题（面试重点）

P029【029_尚硅谷_zk_企业面试真题】03:32

---

第04章-服务器动态上下线监听案例

P020【020_尚硅谷_zk_案例_服务器动态上下线_需求分析】04:41

第 4 章 服务器动态上下线监听案例

4.1 需求

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

4.2 需求分析

P021【021_尚硅谷_zk_案例_服务器动态上下线_服务器注册】10:44

package com.atguigu.case1;

import org.apache.zookeeper.*;

import java.io.IOException;

public class DistributeServer {
    private String connectString = "node1:2181,node2:2181,node3:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        DistributeServer server = new DistributeServer();
        //1、获取zk连接
        server.getConnect();

        //2、注册服务器到zk集群
        server.regist(args[0]);

        //3、启动业务逻辑（睡觉）
        server.business();
    }

    //1、获取zk连接
    private void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            }
        });
    }

    //2、注册服务器到zk集群
    private void regist(String hostname) throws KeeperException, InterruptedException {
        String create = zk.create("/servers/" + hostname, hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname + " is online");
    }

    //3、启动业务逻辑（睡觉）
    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }
}

P022【022_尚硅谷_zk_案例_服务器动态上下线_客户端监听】10:01

package com.atguigu.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;

public class DistributeClient {
    private String connectString = "node1:2181,node2:2181,node3:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        DistributeClient client = new DistributeClient();
        //1、获取zk连接
        client.getConnect();

        //2、监听/servers下面子节点的增加和删除
        client.getServerList();

        //3、业务逻辑（睡觉）
        client.business();
    }

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

    //2、监听/servers下面子节点的增加和删除
    private void getServerList() throws KeeperException, InterruptedException {
        List<String> children = zk.getChildren("/servers", true);
        ArrayList<String> servers = new ArrayList<>();
        for (String child : children) {
            byte[] data = zk.getData("/servers/" + child, false, null);
            servers.add(new String(data));
        }
        System.out.println(servers);//打印
    }

    //3、业务逻辑（睡觉）
    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }
}

P023【023_尚硅谷_zk_案例_服务器动态上下线_测试】06:55

第05章-ZooKeeper分布式锁案例

P024【024_尚硅谷_zk_案例_分布式锁_需求分析】05:18

第 5 章 ZooKeeper分布式锁案例

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

分布式锁案例分析

P025【025_尚硅谷_zk_案例_分布式锁_代码实现（上）】09:57

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

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 {
    private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentMode;

    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) {
                    connectLatch.countDown();
                }

                //waitLatch需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });

        //等待zk正常连接后，往下走程序
        connectLatch.await();

        //判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);

        if (stat == null) {
            //创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    //对zk加锁
    public void zklock() {
        //创建对应的临时带序号节点
        //判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点
    }

    //解锁
    public void unZkLock() {
        //删除节点
    }
}

P026【026_尚硅谷_zk_案例_分布式锁_代码实现（下）】16:19

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 {
    private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentMode;

    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) {
                    connectLatch.countDown();
                }

                //waitLatch需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });

        //等待zk正常连接后，往下走程序
        connectLatch.await();

        //判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);

        if (stat == null) {
            //创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    //对zk加锁
    public void zklock() {
        //创建对应的临时带序号节点
        try {
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            //wait一小会, 让结果更清晰一些
            Thread.sleep(10);

            //判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点

            List<String> children = zk.getChildren("/locks", false);

            //如果children 只有一个值，那就直接获取锁；如果有多个节点，需要判断谁最小
            if (children.size() == 1) {
                return;
            } else {
                Collections.sort(children);
                //获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                //通过seq-00000000获取该节点在children集合的位置
                int index = children.indexOf(thisNode);

                //判断
                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    //就一个节点，可以获取锁了
                    return;
                } else {
                    //需要监听他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath, true, new Stat());
                    //等待监听
                    waitLatch.await();
                    return;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //解锁
    public void unZkLock() {
        //删除节点
        try {
            zk.delete(this.currentMode, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}

P027【027_尚硅谷_zk_案例_分布式锁_测试】05:39

package com.atguigu.case2;

import org.apache.zookeeper.KeeperException;

import java.io.IOException;

public class DistributedLockTest {
    public static void main(String[] args) throws InterruptedException, IOException, 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 (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.unZkLock();
                    System.out.println("线程2 释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

P028【028_尚硅谷_zk_案例_分布式锁_成熟框架curator】09:44

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

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

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

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

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

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

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

详情请查看官方文档：https://curator.apache.org/index.html

3）Curator 案例实操

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，再次获取到锁。");

                    Thread.sleep(5 * 1000);

                    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，再次获取到锁。");

                    Thread.sleep(5 * 1000);

                    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("node1:2181,node2:2181,node3:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy).build();

        client.start();//启动客户端

        System.out.println("zookeeper启动成功。");
        return client;
    }
}

第06章-企业面试真题（面试重点）

P029【029_尚硅谷_zk_企业面试真题】03:32

6.1 选举机制

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

（1）第一次启动选举规则：投票过半数时，服务器 id 大的胜出

（2）第二次启动选举规则：①EPOCH 大的直接胜出、②EPOCH 相同，事务 id 大的胜出、③事务 id 相同，服务器 id 大的胜出

6.2 生产集群安装多少zk合适？

安装奇数台。

生产经验：10 台服务器：3 台 zk；20 台服务器：5 台 zk；100 台服务器：11 台 zk；200 台服务器：11 台 zk。

服务器台数多：好处，提高可靠性；坏处：提高通信延时

6.3 常用命令

ls、get、create、delete