• Zookeeper

    1 概述

    1.1 啥啊

            Zookeeper 是一个开源的分布式的，为分布式框架提供协调服务的 Apache 项目

            Zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。

    1.2 架构

    1）Zookeeper：一个领导者（Leader），多个跟随者（Follower）组成的集群。
    2）集群中只要有    半数以上节点存活，Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。
    3）全局数据一致：每个Server保存一份一模一样的数据副本，Client无论连接到哪个Server，数据都是一致的。
    4）更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行。
    5）数据更新原子性，一次数据更新要么成功，要么失败。
    6）实时性，在一定时间范围内，Client能读到最新数据。

    1.3 文件结构

            每个节点称做一个 ZNode。每一个 ZNode 默认能够存储 1MB 的数据，每个 ZNode 都可以通过其路径唯一标识。

    1.3.1 节点类型（持久/短暂/有序号/无序号）

    短暂（Ephemeral）：客户端和服务器端断开连接后，创建的节点自己删除

    持久（Persistent）：客户端和服务器端断开连接后，创建的节点不删除

    1.3.2 写数据流程

    给Leader

    给Follower 

     

    1.4 应用场景

            提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。 

    1.4.1统一命名服务

            在分布式环境下，经常需要对应用/服务进行统一命名，便于识别。例如：IP不容易记住，而域名容易记住。

    1.4.2 统一配置管理

    分布式环境下，配置文件同步非常常见。

    1. 一般要求一个集群中，所有节点的配置信息是一致的，比如 Kafka 集群。
    2. 对配置文件修改后，希望能够快速同步到各个节点上。

    配置管理可交由ZooKeeper实现。

    1. 可将配置信息写入ZooKeeper上的一个Znode。
    2. 各个客户端服务器监听这个Znode。
    3. 一旦Znode中的数据被修改，ZooKeeper将通知各个客户端服务器。

    1.4.3 统一集群管理

    ZooKeeper可以实现实时监控节点状态变化

    1. 可将节点信息写入ZooKeeper上的一个ZNode。
    2. 监听这个ZNode可获取它的实时状态变化

    1.4.4 服务器节点动态上下线

    客户端能实时洞察到服务器上下线的变化

    1.4.5 软负载均衡

    在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求

    1.5 安装部署

    1.5.1 流程

    学kafka的时候安装

    Apache ZooKeeper

    安装地址：https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz

    解压

    tar -zxvf kafka_2.12-3.0.0.tgz -C /export/server/

    改配置

    1. cd config
    2. cp zoo_sample.cfg zoo.cfg
    3. vim zoo.cfg 
    4.  
    5. # the directory where the snapshot is stored.
    6. # do not use /tmp for storage, /tmp here is just 
    7. # example sakes.
    8. dataDir=/export/server/zookeeper-3.5.7/zkdata
    9.  
    10.  
    11. # 这里和server.几 和下面的myid匹配上
    12. # 2888是正常leader端口，3888是备用端口
    13. server.1=hadoop1:2888:3888
    14. server.2=hadoop2:2888:3888
    15. server.3=hadoop3:2888:3888
    16.  
    17. :wq

    在设置的dataDir目录下创建一个名字叫myid的文件（必须叫myid！！！）

    写入一个数（几都行，只要三个集群别一样）

    [root@hadoop1 zkdata]# echo 1 > myid

    [root@hadoop2 zkdata]# echo 2 > myid

    [root@hadoop3 zkdata]# echo 3 > myid

    fenfa

    1.5.2 zoo.cfg配置解毒

    1. # The number of milliseconds of each tick
    2. # 每个滴答的毫秒数 心跳
    3. tickTime=2000
    4.  
    5. # The number of ticks that the initial 
    6. # synchronization phase can take 
    7. # 初始同步阶段可以占用的心跳数  到了没连上就失败
    8. initLimit=10
    9.  
    10. # The number of ticks that can pass between 
    11. # sending a request and getting an acknowledgement
    12. # 在发送请求和获得确认之间可以传递的心跳数
    13. syncLimit=5
    14.  
    15. # the directory where the snapshot is stored.
    16. # do not use /tmp for storage, /tmp here is just 
    17. # example sakes.
    18. # 快照所在的目录。不使用/ tmp为存储、/ tmp这里只是例子的缘故。
    19. dataDir=/export/server/zookeeper-3.5.7/zkdata
    20. # the port at which the clients will connect
    21. clientPort=2181
    22.  
    23. # the maximum number of client connections.
    24. # increase this if you need to handle more clients
    25. # 客户端连接的最大数量。如果需要处理更多客户端，请增加此值
    26. #maxClientCnxns=60
    27. #
    28. # Be sure to read the maintenance section of the 
    29. # administrator guide before turning on autopurge.
    30. # 在开启自动督促之前，请务必阅读管理员指南的维护部分。
    31. #
    32. # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
    33. #
    34. # The number of snapshots to retain in dataDir
    35. # dataDir中要保留的快照数量
    36. #autopurge.snapRetainCount=3
    37.  
    38. # Purge task interval in hours
    39. # Set to "0" to disable auto purge feature
    40. # 清除任务间隔(小时)  设置为“0”表示禁用自动清除功能
    41. #autopurge.purgeInterval=1

    1.5.3 原神启动

    hadoop1启动

    [root@hadoop1 zookeeper-3.5.7]# bin/zkServer.sh start
    ZooKeeper JMX enabled by default
    Using config: /export/server/zookeeper-3.5.7/bin/../conf/zoo.cfg
    Starting zookeeper ... STARTED

    [root@hadoop1 zookeeper-3.5.7]# bin/zkServer.sh status
    ZooKeeper JMX enabled by default
    Using config: /export/server/zookeeper-3.5.7/bin/../conf/zoo.cfg
    Client port found: 2181. Client address: localhost.
    Error contacting service. It is probably not running.
    一看状态居然失败了？因为另外两个没启动，不过半没法选出来leader

    hadoop2/3也启动

    bin/zkServer.sh start

    现在hadoop1状态

    [root@hadoop1 zookeeper-3.5.7]# bin/zkServer.sh status
    ZooKeeper JMX enabled by default
    Using config: /export/server/zookeeper-3.5.7/bin/../conf/zoo.cfg
    Client port found: 2181. Client address: localhost.
    Mode: follower


    [root@hadoop2 zookeeper-3.5.7]# bin/zkServer.sh status
    ZooKeeper JMX enabled by default
    Using config: /export/server/zookeeper-3.5.7/bin/../conf/zoo.cfg
    Client port found: 2181. Client address: localhost.
    Mode: leader（hadoop2成leader了）


    [root@hadoop3 zookeeper-3.5.7]# bin/zkServer.sh status
    ZooKeeper JMX enabled by default
    Using config: /export/server/zookeeper-3.5.7/bin/../conf/zoo.cfg
    Client port found: 2181. Client address: localhost.
    Mode: follower

    1.5.4 批量启动脚本

    1. #!/bin/bash
    2. case $1 in
    3. start)
    4.     for i in hadoop1 hadoop2 hadoop3
    5.     do
    6.         echo ----------------  zookeeper $i start ----------------
    7.         ssh $i "/export/server/zookeeper-3.5.7/bin/zkServer.sh start"
    8.         echo -----------------------------------------------------
    9.     done
    10.     ;;
    11. stop)
    12.     for i in hadoop1 hadoop2 hadoop3
    13.     do
    14.         echo ----------------  zookeeper $i stop ----------------
    15.         ssh $i "/export/server/zookeeper-3.5.7/bin/zkServer.sh stop"
    16.         echo -----------------------------------------------------
    17.     done
    18.     ;;
    19. status)
    20.     for i in hadoop1 hadoop2 hadoop3
    21.     do
    22.         echo ----------------  zookeeper $i status ----------------
    23.         ssh $i "/export/server/zookeeper-3.5.7/bin/zkServer.sh status"
    24.         echo ------------------------------------------------------
    25.     done
    26.     ;;
    27. *)
    28.     echo "Usage: $0 {start|stop|status}"
    29.     ;;
    30. esac
    31. echo done !!! \\\(\^\o\^\)\/\~ \\\(\^\o\^\)\/\~ \\\(\^\o\^\)\/\~

    chmod 777 zk

    1.5.5 IDEA幻景 

    保证 hadoop102、hadoop103、hadoop104 服务器上 Zookeeper 集群服务端启动。

    Maven添加pom文件 

    1.     <dependency>
    2.       <groupId>org.apache.logging.log4jgroupId>
    3.       <artifactId>log4j-coreartifactId>
    4.       <version>2.8.2version>
    5.     dependency>
    6.     <dependency>
    7.       <groupId>org.apache.zookeepergroupId>
    8.       <artifactId>zookeeperartifactId>
    9.       <version>3.5.7version>
    10.     dependency>

     resources下创建log4j.properties

    1. log4j.rootLogger=INFO, stdout
    2. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    3. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    4. log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
    5. log4j.appender.logfile=org.apache.log4j.FileAppender
    6. log4j.appender.logfile.File=target/spring.log
    7. log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
    8. log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

    1.6 客户端命令

    1.6.1 ls

    bin/zkCli.sh -server hadoop1:2181

    1.6.2 create 

    [zk: hadoop1:2181(CONNECTED) 2] create /yuanshen "keli"
    Created /yuanshen
    [zk: hadoop1:2181(CONNECTED) 3] create /yuanshen/daoqi "wanye"
    Created /yuanshen/daoqi
    [zk: hadoop1:2181(CONNECTED) 4] get /yuanshen（每个节点的属性值？）
    keli
    [zk: hadoop1:2181(CONNECTED) 5] get -s /yuanshen
    keli
    cZxid = 0x900000004
    ctime = Mon Oct 23 16:42:43 CST 2023
    mZxid = 0x900000004
    mtime = Mon Oct 23 16:42:43 CST 2023
    pZxid = 0x900000005
    cversion = 1
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x0
    dataLength = 4
    numChildren = 1

    [zk: hadoop1:2181(CONNECTED) 6] create -s /yuanshen/liyue "zhongli"
    Created /yuanshen/liyue0000000001
    带序号的可以重复名创建，没序号的不能创建重复（带序号的 -s，临时-e）

    [zk: hadoop1:2181(CONNECTED) 9] set /yuanshen/liyue0000000001 "shenhe"（改属性）

    1.6.3 delete 

    [zk: hadoop1:2181(CONNECTED) 15] delete /yuanshen/liyue0000000001
    [zk: hadoop1:2181(CONNECTED) 16] deleteall /yuanshen
    [zk: hadoop1:2181(CONNECTED) 17] ls /
    [kafka, zookeeper]

    1.7 监听器原理

            客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目录节点增加删除）时，ZooKeeper 会通知客户端。监听机制保证 ZooKeeper 保存的任何的数据的任何改变都能快速的响应到监听了该节点的应用程序。

    1）首先要有一个main()线程
    2）在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信（connet），一个负责监听（listener）。
    3）通过connect线程将注册的监听事件发送给Zookeeper。
    4）在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中。
    5）Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程。
    6）listener线程内部调用了process()方法。

    例子：

    都启动客户端，hadoop3上设置监听

    get -w /yuanshen

    hadoop2对属性进行改变，hadoop3显示

    WATCHER::

    WatchedEvent state:SyncConnected type:NodeDataChanged path:/yuanshen

    注意：在hadoop103再多次修改/sanguo的值，hadoop104上不会再收到监听。因为注册一次，只能监听一次。想再次监听，需要再次注册。 

    1.7.1  实例代码

    1. import org.apache.zookeeper.*;
    2. import org.apache.zookeeper.data.Stat;
    3. import org.junit.Before;
    4. import sun.net.idn.Punycode;
    5.  
    6. import java.io.IOException;
    7. import java.nio.charset.StandardCharsets;
    8. import java.util.List;
    9. import org.junit.Test;
    10.  
    11. public class zkClient {
    12.     private final String server = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
    13.     private final int sessionTimeout = 5000;
    14.     private ZooKeeper zkClient;
    15.  
    16.  
    17.     @Before
    18.     public void init() throws IOException {
    19.         zkClient = new ZooKeeper(server, sessionTimeout, new Watcher() {
    20.             @Override
    21.             public void process(WatchedEvent watchedEvent) {
    22.                 // 收到事件之后的逻辑
    23.                 List children = null;
    24.                 try {
    25.                     children = zkClient.getChildren("/", true);
    26.  
    27.                     for (String child:children){
    28.                         System.out.println(child);
    29.                     }
    30.                 } catch (Exception e) {
    31.                     e.printStackTrace();
    32.                 }
    33.  
    34.  
    35.             }
    36.         });
    37.     }
    38.  
    39.  
    40.     @Test
    41.     public void create() throws InterruptedException, KeeperException {
    42.         String s = zkClient.create("/yuanshen", "ye".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    43.  
    44.     }
    45.  
    46.  
    47.     @Test
    48.     public void getChildren() throws InterruptedException {
    49.         try {
    50.             List children = zkClient.getChildren("/", true);
    51.             for (String child: children){
    52.                 System.out.println(child);
    53.             }
    54.         } catch (Exception e) {
    55.             e.printStackTrace();
    56.         }
    57.         Thread.sleep(100*1000);
    58.     }
    59.     // 判断 znode 是否存在
    60.     @Test
    61.     public void exist() throws Exception {
    62.         Stat stat = zkClient.exists("/atguigu", false);
    63.  
    64.         System.out.println(stat == null ? "not exist" : "exist");
    65.     }
    66. }

            尚硅谷给ye讲迷了，这里每个监听语句都会在这里阻塞，true表示会循环监听，就这么一句话讲的弯弯绕绕的

    2 案例

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

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

    服务端

    1. package org.example.zk;
    2.  
    3. import org.apache.zookeeper.*;
    4.  
    5. import java.io.IOException;
    6.  
    7. public class Servers {
    8.     String servers = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
    9.     int SessionTimeout = 2000;
    10.     ZooKeeper zk = null;
    11.     String parentNode = "/server";
    12.  
    13.  
    14.     public void getConnect() throws IOException {
    15.         zk = new ZooKeeper(servers, SessionTimeout, new Watcher() {
    16.             @Override
    17.             public void process(WatchedEvent event) {
    18.  
    19.             }
    20.         });
    21.     }
    22.  
    23.     public void registServer(String hostname) throws InterruptedException, KeeperException {
    24.         String create = zk.create(parentNode+"/server", hostname.getBytes(), 
    25.                 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    26.         System.out.println(hostname +" is online "+ create);
    27.  
    28.     }
    29.  
    30.     public void business(String hostname) throws Exception{
    31.         System.out.println(hostname + " is working ...");
    32.  
    33.         Thread.sleep(Long.MAX_VALUE);
    34.     }
    35.  
    36.     public static void main(String[] args) throws Exception {
    37.  
    38.         // 1获取 zk连接
    39.         Servers server = new Servers();
    40.         server.getConnect();
    41.  
    42.         // 2 利用 zk连接注册服务器信息
    43.         server.registServer(args[0]);
    44.  
    45.         // 3 启动业务功能
    46.         server.business(args[0]);
    47.     }
    48. }

    客户端

    1. package org.example.zk;
    2.  
    3. import org.apache.zookeeper.WatchedEvent;
    4. import org.apache.zookeeper.Watcher;
    5. import org.apache.zookeeper.ZooKeeper;
    6.  
    7. import java.io.IOException;
    8. import java.util.ArrayList;
    9. import java.util.List;
    10.  
    11. public class Clients {
    12.     private static String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
    13.     private static int sessionTimeout = 2000;
    14.     private ZooKeeper zk = null;
    15.     private String parentNode = "/servers";
    16.  
    17.  
    18.     // 创建到 zk的客户端连接
    19.     public void getConnect() throws IOException {
    20.         zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    21.             @Override
    22.             public void process(WatchedEvent event) {
    23.                 // 再次启动监听
    24.                 try {
    25.                     getServerList();
    26.                 } catch (Exception e) {
    27.                     e.printStackTrace();
    28.                 }
    29.             }
    30.         });
    31.     }
    32.  
    33.  
    34.     // 获取服务器列表信息
    35.     public void getServerList() throws Exception {
    36.  
    37.         // 1获取服务器子节点信息，并且对父节点进行监听
    38.         List children = zk.getChildren(parentNode, true);
    39.  
    40.         // 2存储服务器信息列表
    41.         ArrayList servers = new ArrayList<>();
    42.  
    43.         // 3遍历所有节点，获取节点中的主机名称信息
    44.         for (String child : children) {
    45.             byte[] data = zk.getData(parentNode + "/" + child, false, null);
    46.             servers.add(new String(data));
    47.         }
    48.  
    49.         // 4打印服务器列表信息
    50.         System.out.println(servers);
    51.     }
    52.  
    53.     // 业务功能
    54.     public void business() throws Exception{
    55.  
    56.         System.out.println("client is working ...");
    57.         Thread.sleep(Long.MAX_VALUE);
    58.     }
    59.  
    60.     public static void main(String[] args) throws Exception {
    61.  
    62.         // 1获取 zk连接
    63.         Clients client = new Clients();
    64.         client.getConnect();
    65.  
    66.         // 2获取 servers的子节点信息，从中获取服务器信息列表
    67.         client.getServerList();
    68.  
    69.         // 3业务进程启动
    70.         client.business();
    71.     }
    72. }

    2.2 分布式锁

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

    1. package org.example.zk;
    2.  
    3. import org.apache.zookeeper.*;
    4. import org.apache.zookeeper.data.Stat;
    5.  
    6. import java.io.IOException;
    7. import java.util.Collections;
    8. import java.util.List;
    9. import java.util.concurrent.CountDownLatch;
    10.  
    11. public class Lock {
    12.     // zookeeper server 列表
    13.     private String connectString = "hadoop1:2181,hadoop2:2181,hadoop3:2181";
    14.     // 超时时间
    15.     private int sessionTimeout = 2000;
    16.  
    17.     private ZooKeeper zk;
    18.  
    19.     private String rootNode = "locks";
    20.     private String subNode = "seq-";
    21.     // 当前 client等待的子节点
    22.     private String waitPath;
    23.  
    24.     //ZooKeeper连接
    25.     private CountDownLatch connectLatch = new CountDownLatch(1);
    26.  
    27.     //ZooKeeper节点等待
    28.     private CountDownLatch waitLatch = new CountDownLatch(1);
    29.  
    30.     // 当前 client创建的子节点
    31.     private String currentNode;
    32.  
    33.     // 和 zk 服务建立连接，并创建根节点
    34.     public Lock() throws IOException,
    35.             InterruptedException, KeeperException {
    36.  
    37.         zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
    38.                     @Override
    39.                     public void process(WatchedEvent event) {
    40.                         // 连接建立时, 打开 latch, 唤醒 wait在该 latch 上的线程
    41.                         if (event.getState() == Event.KeeperState.SyncConnected){
    42.                             connectLatch.countDown();
    43.                         }
    44.  
    45.                         // 发生了 waitPath的删除事件
    46.                         if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)){
    47.                             waitLatch.countDown();
    48.                         }
    49.                     }
    50.                 });
    51.  
    52.         // 等待连接建立
    53.         connectLatch.await();
    54.  
    55.         //获取根节点状态
    56.         Stat stat = zk.exists("/" + rootNode, false);
    57.  
    58.         //如果根节点不存在，则创建根节点，根节点类型为永久节点
    59.         if (stat == null) {
    60.             System.out.println("根节点不存在");
    61.             zk.create("/" + rootNode, new byte[0],
    62.                     ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    63.         }
    64.     }
    65.  
    66.  
    67.     // 加锁方法
    68.     public void zkLock() {
    69.  
    70.         try {
    71.             //在根节点下创建临时顺序节点，返回值为创建的节点路径
    72.             currentNode = zk.create("/" + rootNode + "/" + subNode,
    73.                     null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
    74.                     CreateMode.EPHEMERAL_SEQUENTIAL);
    75.  
    76.             // wait一小会, 让结果更清晰一些
    77.             Thread.sleep(10);
    78.  
    79.             // 注意, 没有必要监听"/locks"的子节点的变化情况
    80.             List childrenNodes = zk.getChildren("/" +
    81.                     rootNode, false);
    82.  
    83.             // 列表中只有一个子节点, 那肯定就是 currentNode , 说明client获得锁
    84.             if (childrenNodes.size() == 1) {
    85.                 return;
    86.             } else {
    87.                 //对根节点下的所有临时顺序节点进行从小到大排序
    88.                 Collections.sort(childrenNodes);
    89.  
    90.                 //当前节点名称
    91.                 String thisNode = currentNode.substring(("/" +
    92.                         rootNode + "/").length());
    93.                 //获取当前节点的位置
    94.                 int index = childrenNodes.indexOf(thisNode);
    95.  
    96.                 if (index == -1) {
    97.                     System.out.println("数据异常");
    98.                 } else if (index == 0) {
    99.                     // index == 0, 说明 thisNode 在列表中最小, 当前client获得锁
    100.                     return;
    101.                 } else {
    102.                     // 获得排名比 currentNode 前 1位的节点
    103.                     this.waitPath = "/" + rootNode + "/" +
    104.                             childrenNodes.get(index - 1);
    105.  
    106.                     // 在 waitPath 上注册监听器, 当 waitPath 被删除时,zookeeper 会回调监听器的 process方法
    107.                     zk.getData(waitPath, true, new Stat());
    108.                     //进入等待锁状态
    109.                     waitLatch.await();
    110.  
    111.                     return;
    112.                 }
    113.             }
    114.         } catch (Exception e) {
    115.             e.printStackTrace();
    116.         }
    117.     }
    118.  
    119.     // 解锁方法
    120.     public void zkUnlock() {
    121.         try {
    122.             zk.delete(this.currentNode, -1);
    123.         } catch (InterruptedException | KeeperException e) {
    124.             e.printStackTrace();
    125.         }
    126.     }
    127.  
    128.  
    129. }
    130.  
    131.  
    132.  
    133.  
    134. package org.example.zk;
    135.  
    136. import org.apache.zookeeper.KeeperException;
    137.  
    138. import java.io.IOException;
    139.  
    140. public class LockTest {
    141.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    142.  
    143.         // 创建分布式锁 1
    144.         final Lock lock1 = new Lock();
    145.         // 创建分布式锁 2
    146.         final Lock lock2 = new Lock();
    147.  
    148.         new Thread(new Runnable() {
    149.             @Override
    150.             public void run() {
    151.                 // 获取锁对象
    152.                 try {
    153.                     lock1.zkLock();
    154.                     System.out.println("线程 1 获取锁");
    155.                     Thread.sleep(5 * 1000);
    156.  
    157.                     lock1.zkUnlock();
    158.                     System.out.println("线程 1 释放锁");
    159.                 } catch (Exception e) {
    160.                     e.printStackTrace();
    161.                 }
    162.             }
    163.         }).start();
    164.  
    165.         new Thread(new Runnable() {
    166.             @Override
    167.             public void run() {
    168.                 // 获取锁对象
    169.                 try {
    170.                     lock2.zkLock();
    171.                     System.out.println("线程 2 获取锁");
    172.                     Thread.sleep(5 * 1000);
    173.  
    174.                     lock2.zkUnlock();
    175.                     System.out.println("线程 2 释放锁");
    176.                 } catch (Exception e) {
    177.                     e.printStackTrace();
    178.                 }
    179.             }
    180.         }).start();
    181.     }
    182.  
    183. }
    184.  
    185.  
    186.

    2.3 Curator 框架实现分布式锁

    1. package org.example.zk;
    2.  
    3. import org.apache.curator.RetryPolicy;
    4. import org.apache.curator.framework.CuratorFramework;
    5. import org.apache.curator.framework.CuratorFrameworkFactory;
    6. import org.apache.curator.framework.recipes.locks.InterProcessLock;
    7. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
    8. import org.apache.curator.retry.ExponentialBackoffRetry;
    9.  
    10. public class Curator {
    11.     private String rootNode = "/locks";
    12.     // zookeeper server 列表 
    13.     private String connectString =
    14.             "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    15.  
    16.     // connection超时时间 
    17.     private int connectionTimeout = 2000;
    18.  
    19.     // session 超时时间 
    20.     private int sessionTimeout = 2000;
    21.  
    22.     public static void main(String[] args) {
    23.  
    24.         new Curator().test();
    25.     }
    26.  
    27.     // 测试 
    28.     private void test() {
    29.  
    30.         // 创建分布式锁 1 
    31.         final InterProcessLock lock1 = new InterProcessMutex(getCuratorFramework(), rootNode);
    32.  
    33.         // 创建分布式锁 2 
    34.         final InterProcessLock lock2 = new InterProcessMutex(getCuratorFramework(), rootNode);
    35.  
    36.         new Thread(new Runnable() {
    37.             @Override
    38.             public void run() {
    39.                 // 获取锁对象 
    40.                 try {
    41.                     lock1.acquire();
    42.                     System.out.println("线程 1 获取锁");
    43.                     // 测试锁重入 
    44.                     lock1.acquire();
    45.                     System.out.println("线程 1 再次获取锁");
    46.                     Thread.sleep(5 * 1000);
    47.                     lock1.release();
    48.                     System.out.println("线程 1 释放锁");
    49.                     lock1.release();
    50.                     System.out.println("线程 1 再次释放锁");
    51.                 } catch (Exception e) {
    52.                     e.printStackTrace();
    53.                 }
    54.             }
    55.         }).start();
    56.  
    57.         new Thread(new Runnable() {
    58.             @Override
    59.             public void run () {
    60.                 // 获取锁对象 
    61.                 try {
    62.                     lock2.acquire();
    63.                     System.out.println("线程 2 获取锁");
    64.                     // 测试锁重入 
    65.                     lock2.acquire();
    66.                     System.out.println("线程 2 再次获取锁");
    67.                     Thread.sleep(5 * 1000);
    68.                     lock2.release();
    69.                     System.out.println("线程 2 释放锁");
    70.                     lock2.release();
    71.                     System.out.println("线程 2 再次释放锁");
    72.                 } catch (Exception e) {
    73.                     e.printStackTrace();
    74.                 }
    75.             }
    76.         }).start();
    77.     }
    78.  
    79.     // 分布式锁初始化 
    80.     public CuratorFramework getCuratorFramework (){
    81.  
    82.         //重试策略，初试时间 3秒，重试 3 次 
    83.         RetryPolicy policy = new ExponentialBackoffRetry(3000, 3);
    84.  
    85.         //通过工厂创建 Curator 
    86.         CuratorFramework client = CuratorFrameworkFactory.builder()
    87.                         .connectString(connectString)
    88.                         .connectionTimeoutMs(connectionTimeout)
    89.                         .sessionTimeoutMs(sessionTimeout)
    90.                         .retryPolicy(policy).build();
    91.  
    92.         //开启连接 
    93.         client.start();
    94.         System.out.println("zookeeper 初始化完成...");
    95.         return client;
    96.     }
    97.  
    98. }
    99.         
    100.     
    101.  
    102.  
    103.  
    104.

    3 面试内容

    3.1 选举机制 

     半数机制，超过半数的投票通过，即通过。 
    （1）第一次启动选举规则： 
    投票过半数时，服务器 id 大的胜出 
    （2）第二次启动选举规则： 
    ①EPOCH 大的直接胜出 
    ②EPOCH 相同，事务 id 大的胜出 
    ③事务 id 相同，服务器 id 大的胜出 

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

    安装奇数台。 
    生产经验： 
    ⚫ 10 台服务器：3 台 zk； 
    ⚫ 20 台服务器：5 台 zk； 
    ⚫ 100 台服务器：11 台 zk； 
    ⚫ 200 台服务器：11 台 zk 

  • 相关阅读:
    Prometheus 采集snmp监控数据
    DeOldify 接口化改造 集成 Flask
    MPI实现矩阵向量乘法
    【FLASH存储器系列十】ONFI数据接口的时序参数与时序图
    接口--抽象方法
    新版TCGAbiolinks包学习03：差异分析
    Altium design 经验补充2
    java毕业设计社区微服务平台小程序服务器端Mybatis+系统+数据库+调试部署
    Socket说明
    数据挖掘实验：使用 Hadoop 实现 WordCount 应用
  • 原文地址：https://blog.csdn.net/qq_58551342/article/details/133992447