服务注册和发现-Zookeeper(二)

Zookeeper 概述

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

Zookeeper 工作机制

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

Zookeeper 特点

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

数据结构

  Zookeeper 数据模型的结构和Unix文件系统类似，整体上可以看成是一棵树，每个节点称为一个ZNode，而每个ZNode默认能够储存1MB的数据，每个ZNode都可以通过其路径唯一标识

Zookeeper 的应用场景

• 统一命名服务： 在分布式环境下，经常需要对服务/应用进行统一命名，方便识别
• 统一配置管理： 分布式环境下配置文件的同步非常常见，一般要求一个集群中，所有的节点配置信息是一致的，并且对配置文件进行修改后，能够快速同步到各个节点上；配置管理可交由Zookeeper 实现，可将配置信息写入Zookeeper 上的ZNode，各个客户端服务器监听这个ZNode，一旦ZNode中的数据被修改，Zookeeper 将通知各个客户端服务器进行更新。
• 统一集群管理： 分布式环境中，实时掌握每个节点的状态是必要的，可根据节点实时状态做出一些调整。Zookeeper 可以实现实时监控节点状态变化，可将节点信息写入Zookeeper 上的一个ZNode，监听这个ZNode可获取它的实时状态变化。
• 服务器动态上下线： 客户端能够实时洞察到服务器上下线的状态。
• 软负载均衡： 在Zookeeper 中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求

Zookeeper 安装

  Zookeeper 的官方网站：https://zookeeper.apache.org/ 下载地址：https://zookeeper.apache.org/releases.html 本文以3.8.0版本进行演示。

本地模式安装

  本地模式主要用于本地开发测试，生产环境的安装可以参考后文。Zookeeper 的运行需要JDK环境的支持，要求1.8以上，本次安装的环境采用jdk1.8，操作系统是Centos7
  将Zookeeper 下载到本地并上传到服务器进行解压

# 解压Zookeeper 
tar -zxvf apache-zookeeper-3.8.0-bin.tar.gz

# 进入到conf目录下将zoo_sample.cfg复制一份
cp zoo_sample.cfg zoo.cfg

# 修改配置文件
vim zoo.cfg
# 修改以下参数
# 1. 修改数据储存位置
dataDir=/home/zookeeper

# 启动服务端  start启动 | stop停止 | status状态 | restart重启
./bin/zkServer.sh start

# 启动客户端
./bin/zkCli.sh

# 退出客户端
quit

配置参数详解

# 服务端和客户端通信心跳时间，单位毫秒
tickTime=2000
# leader和follower初始连接时能容忍的最多心跳数(tickTime的数量)
initLimit=10
# leader和follower之间通信时间如果超过 syncLimit * tickTime ，leader认为follower死掉，从服务器列表中删除follower
syncLimit=5
# zookeeper数据保存路径
dataDir=/home/zookeeper
# 客户端连接端口
clientPort=2181
# 客户端最大连接数，默认60
maxClientCnxns=60
# 要在dataDir中保留的快照数，默认3
autopurge.snapRetainCount=3
# 清除任务间隔（小时），设置为0不清除任务，默认1
autopurge.purgeInterval=1

集群安装

# 在Zookeeper 的数据储存目录下创建 myid 文件，并在文件中添加服务器的编号，注意编号唯一不能重复
vim /home/zookeeper/myid
# 输入每个zookeeper服务器对应的编号，不能重复

# 在zookeeper的配置文件中增加集群配置
# 规则如下：
# server.A=B:C:D
# A 是一个数字，表示这是第几号服务器，就是配置在myid的数字
# B 服务器地址
# C 这个服务器Follower与集群的leader服务器交换信息的端口
# D 执行选举时服务器相互通信的端口
# 示例：
server.1=192.168.0.132:2888:3888
server.2=192.168.0.134:2888:3888
server.3=192.168.0.133:2888:3888

# 启动集群
./bin/zkServer.sh start

# 查看状态
./bin/zkServer.sh status

# 注意关闭防火墙或者开放端口

选举机制

  首先需要了解几个名词

• SID： 服务器ID，用来唯一标识一台Zookeeper集群中的机器，每台机器不能重复，和myid一致
• ZXID： 事务ID，ZXID是一个事务ID，用来标识一次服务器状态的变更。在某一时刻，集群中的每台机器的ZXID值不一定完全一样，这和Zookeeper服务器对于客户端更新请求的处理逻辑有关
• Epoch： 每个leader任期的代号，没有leader时同一轮投票过程中的逻辑时钟是相同的，每投完一次票这个数据就会增加
  
    第一次启动时的选举机制
    1. 服务器1启动时，发起一次选举。服务器1投自己一票，此时服务器1票数为1，不够半数以上，选举无法完成，服务器1状态保持为LOOKING
    2. 服务器2启动时，在发起一次选举，服务器1和2分别投自己一票并交换选票信息，此时服务器1发现服务器2的myid比自己目前投票推举的更大，更改选票为推举服务器2。 此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1、2状态保持LOOKING
    3. 服务器3启动时发起一次选举，根据myid的值，服务器1、2都会更改选票为服务器3，此时服务器1为0票，服务器2为0票，服务器3为3票超过了半数，选举服务器3为Leader。服务器1、2更改状态为FOLLOWING，服务器3更改状态为LEADING
    4. 服务器4启动时发起一次选举，此时服务器1、2、3的状态不是LOOKING不会更改选票信息，交换选票信息的结果为：服务器3为3票，服务器4为1票。此时服务器4需要服从多数更改选票信息为服务器3，并更改状态为FOLLOWING
    5. 服务器5启动时同服务器4一样，更改状态为FOLLOWING
    
    非第一次启动的选举机制
    1. 当Zookeeper集群中的一台服务器出现以下两种情况之一时，就会开始进入leader选举：
• 服务器初始化启动
• 服务器运行期间无法和leader保持连接
    2. 当一台机器进入leader选举流程时，当前集群也可能会处于以下两种状态
• 集群本来就已经存在一个leader。对于这种已经存在一个leader情况，机器视图去选举leader时，会被告知当前服务器的leader信息，对于该机器来说，仅仅需要和leader建立简介并进行状态同步即可
• 集群中确实不存在leader。对于这种情况选举leader的规则是：①Epoch大的直接胜出 ②Epoch相同的事务ID大的胜出 ③事务ID相同服务器ID大的胜出

Zookeeper启动停止脚本

  给大家提供一个Zookeeper集群启停的脚本

#!/bin/bash
case $1 in 
"start") { # 启动
	for i in 192.168.0.132 192.168.0.133 192.168.0.134
	do
		echo --------- zookeeper $i 启动 ---------
		ssh $i "/home/apache-zookeeper-3.8.0-bin/bin/zkServer.sh start"
	done
}
;;
"stop") { # 停止
	for i in 192.168.0.132 192.168.0.133 192.168.0.134
	do
		echo --------- zookeeper $i 停止 ---------
		ssh $i "/home/apache-zookeeper-3.8.0-bin/bin/zkServer.sh stop"
	done
}
;;
"status") { # 状态
	for i in 192.168.0.132 192.168.0.133 192.168.0.134
	do
		echo --------- zookeeper $i 状态 ---------
		ssh $i "/home/apache-zookeeper-3.8.0-bin/bin/zkServer.sh status"
	done
}
;;
esac

客户端命令行操作

1. 命令行语法

2. znode节点数据信息

启动客户端

# -server 连接指定的zookeeper服务器，默认连接本机
./bin/zkCli.sh -server 192.168.0.133:2181

查看节点信息

# 查看当前节点信息
# 可选参数 -s -w -R
ls -s /

3. 节点类型(持久/短暂/有序号/无序号)

  持久(Persistent)： 客户端和服务端断开连接后创建的节点不删除
  短暂(Ephemeral)： 客户端和服务端断开连接后创建的节点自己删除

• 持久化目录节点，客户端和zookeeper断开连接后该节点依旧存在
• 持久化顺序编号目录节点，客户端和zookeeper断开连接后，该节点依旧存在，只是zookeeper给该节点名称进行顺序编号
• 临时目录节点，客户端与zookeeper断开连接后，该节点被删除
• 临时顺序编号目录节点，客户端和zookeeper断开连接后，该节点被删除，只是 zookeeper给该节点名称进行顺序编号
    说明：创建znode节点是设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护
    注意：在分布式系统中，顺序号可以被用来为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序

示例：

# 创建不带序号的永久节点：create 节点名称 节点说明
create /testZnode "测试节点"

# 创建带序号的永久节点
create -s /test001 "test001"

# 创建不带序号的临时节点
create -e /test002 ""

# 创建带序号的临时节点
create -s -e /test003

# 获取节点信息
get -s /testZnode

# 修改节点值
set /testZnode "test demo"

4. 监听器原理

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

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

示例：

# 1. 监听节点值的变化
# 注册监听
get -w testZnode
# 在其他的客户端中修改节点的值
set /testZnode "qqqqqqqq"
# 可以看到在注册监听的那个控制台监听到节点数据的变化
# WatchedEvent state:SyncConnected type:NodeDataChanged path:/testZnode

# 2. 监听子节点数据的变化
# 注册监听
ls -w /testZnode
# 在其他客户端在/testZnode下创建子节点
create /testZnode/test "wwwwww"
# 在注册的控制台监听到节点的变化
# WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/testZnode

# 注意：注册一次，只能监听一次，想要再次监听就需要再次注册

5. 节点删除与查看

# 删除节点：delete 节点名
delete /test0010000000001

# 递归删除：deleteall 节点名
deleteall /testZnode

# 查看节点状态：stat 节点名
stat /zookeeper

Java操作zookeeper

  演示代码基于普通maven项目
1. 导入jar包

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.8.0</version>
</dependency>

2. 创建zookeeper客户端并操作zookeeper集群

public class ZookeeperClient {
    ZooKeeper zooKeeper;

    /**
     * 创建客户端
     */
    @Before
    public void createZooKeeperClient() throws IOException {
        // 创建zookeeper客户端
        zooKeeper = new ZooKeeper("192.168.0.132:2181,192.168.0.133:2181,192.168.0.134:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 监听节点事件，实现无限次监听
                // 获取子节点
                List<String> children = null;
                try {
                    children = zooKeeper.getChildren("/", true);
                } catch (KeeperException | InterruptedException ex) {
                    ex.printStackTrace();
                }
                for (String child : children) {
                    System.out.println(child);
                }
            }
        });
    }

    /**
     * 创建节点
     */
    @Test
    public void createNode() throws InterruptedException, KeeperException {
        // 创建子节点
        // 第三个参数代表权限：
        String node = zooKeeper.create("/test", "test demo".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * 获取子节点
     */
    @Test
    public void getChildren() throws InterruptedException, KeeperException {
        // 获取子节点
        List<String> children = zooKeeper.getChildren("/", true);
        for (String child : children) {
            System.out.println(child);
        }
        // 延迟
        Thread.sleep(Integer.MAX_VALUE);

        // 运行后在控制台增加或删除子节点可查看效果
    }

    /**
     * 判断节点是否存在
     */
    @Test
    public void exists() throws InterruptedException, KeeperException {
        Stat exists = zooKeeper.exists("/sdfasd", false);
        System.out.println(exists == null ? "节点不存在" : "节点存在");
    }

    /**
     * 删除节点
     */
    @Test
    public void deleteNode() throws InterruptedException, KeeperException {
        // 删除节点，-1代码不指定版本即删除所有版本，也可以指定版本
        zooKeeper.delete("/test", -1);
    }
}

客户端向服务端写数据流程

1. 写入请求直接发送给Leader节点

  客户端将数据写入到Leader后，Leader会将数据写入给其他的Follower，Follower在数据写入完成后通过ACK进行应答Leader；当超过半数的服务器写入成功后，Leader通过ACK应答客户端写入完成，同时给其他的服务器写入数据。

2. 写入请求发送给Follower节点

  客户端发起写入数据请求给Follower，Follower将写入请求转发给Leader，Leader再将数据写入给其他的Follower，Follower在数据写入完成后通过ACK进行应答Leader；当超过半数的服务器写入成功后，Leader通过ACK应答接收请求的Follower，该Follower则通过ACK应答客户端写入成功，同时Leader给其他的服务器写入数据。

服务器动态上下线

  在分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。
  服务端启动时，会在zookeeper集群中创建对应的节点，客户端获取当前在线的服务器列表并进行监听。当某个服务端宕机后，客户端就会收到服务端下线的通知并重新获取在线服务端列表进行监听

示例：
  1. 创建服务端节点，用于储存所有的服务端节点信息

create /servers "servers"

  2. 使用Java编写服务端，并注册到zookeeper，数据保存在 “/servers” 节点下

public class ServerDemo01 {
    public static void main(String[] args) throws InterruptedException, KeeperException {
        // 1. 获取zookeeper连接
        ZooKeeper zooKeeper = getConnection();
        // 2. 服务端注册到zookeeper集群,创建临时的带序列号的节点
        zooKeeper.create("/servers/ServerDemo01", "ServerDemo01".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println("ServerDemo01 上线了");
        // 3. 业务处理逻辑
        Thread.sleep(Long.MAX_VALUE);
    }

    public static ZooKeeper getConnection() {
        return getConnection("192.168.0.132:2181,192.168.0.133:2181,192.168.0.134:2181", 2000);
    }

    public static ZooKeeper getConnection(String connectString, int sessionTimeout) {
        try {
            return new ZooKeeper(connectString, sessionTimeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

  3. 使用Java编写客户端，监听 “/servers” 节点

public class ClientDemo01 {
    public static ZooKeeper zooKeeper;

    public static void main(String[] args) throws InterruptedException, KeeperException {
        // 1. 获取zookeeper连接
        zooKeeper = getConnection();
        // 2. 监听节点的变化
        List<String> children = zooKeeper.getChildren("/servers", true);
        List<String> list = new ArrayList<>();
        for (String child : children) {
            // 获取子节点数据
            byte[] data = zooKeeper.getData("/servers/"   child, false, null);
            String s = new String(data);
            list.add(s);
        }
        list.forEach(System.out::println);
        System.out.println("=========================================================");
        // 3. 业务逻辑
        Thread.sleep(Long.MAX_VALUE);
    }

    public static ZooKeeper getConnection() {
        return getConnection("192.168.0.132:2181,192.168.0.133:2181,192.168.0.134:2181", 2000);
    }

    public static ZooKeeper getConnection(String connectString, int sessionTimeout) {
        try {
            return new ZooKeeper(connectString, sessionTimeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    List<String> children = null;
                    try {
                        children = zooKeeper.getChildren("/servers", true);
                    } catch (KeeperException | InterruptedException e) {
                        e.printStackTrace();
                    }
                    List<String> list = new ArrayList<>();
                    assert children != null;
                    for (String child : children) {
                        // 获取子节点数据
                        byte[] data = new byte[0];
                        try {
                            data = zooKeeper.getData("/servers/"   child, false, null);
                        } catch (KeeperException | InterruptedException e) {
                            e.printStackTrace();
                        }
                        String s = new String(data);
                        list.add(s);
                    }
                    list.forEach(System.out::println);
                    System.out.println("=========================================================");
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

  4. 测试：先启动客户端，在启动服务端，当客户端监听到服务端注册到zookeeper时就会进行输出，当监听到服务端下线时也会进行输出。

Zookeeper 分布式锁

示例：
  1. Java编写加锁的示例demo

public class LockDemo {
    private final String NODE = "/locks";
    private final CountDownLatch latch = new CountDownLatch(1);
    private final CountDownLatch waitLatch = new CountDownLatch(1);
    private ZooKeeper zooKeeper;
    private String waitPath;
    private String currentNode;

    public LockDemo() throws InterruptedException, KeeperException, IOException {
        // 获取连接
        zooKeeper = new ZooKeeper("192.168.0.132:2181,192.168.0.133:2181,192.168.0.134:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 释放等待
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    latch.countDown();
                }
                if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });
        // 等待zookeeper正常连接后，才执行后面的代码
        latch.await();
        // 判断根节点是否存在
        Stat exists = zooKeeper.exists(NODE, false);
        if (exists == null) { // 不存在则创建根节点
            zooKeeper.create(NODE, NODE.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    /**
     * 加锁
     */
    public void lock() throws InterruptedException, KeeperException {
        // 创建对应的临时的带序号的节点
        currentNode = zooKeeper.create(NODE   "/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 判断节点是否是最小序号的节点，如果是则获取到锁，如果不是则监听前一个序号的节点
        List<String> children = zooKeeper.getChildren(NODE, false);

        // 如果只有一个节点，直接获取锁，如果有多个节点，判断是否是最小值
        if (children.size() == 1) {
        } else {
            Collections.sort(children);// 排序
            // 当前节点名称
            String thisNode = currentNode.substring(NODE.length()   1);
            // 当前节点的下标
            int index = children.indexOf(thisNode);
            if (index == -1) {
                System.out.println("数据异常");
            } else if (index == 0) {
                // 获取锁
            } else {
                // 监听前一个节点
                waitPath = NODE   "/"   children.get(index - 1);
                zooKeeper.getData(waitPath, true, null);
                // 等待监听完成
                waitLatch.await();
            }
        }
    }

    /**
     * 解锁
     */
    public void unLock() throws InterruptedException, KeeperException {
        // 删除当前节点
        zooKeeper.delete(currentNode, -1);
    }
}

  2. 测试，使用多线程异步创建多个线程

public class LockDemoTest {
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        for (int i = 0; i < 10; i  ) {
            LockDemo lockDemo = new LockDemo();
            int finalI = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        lockDemo.lock();// 加锁
                        System.out.println("线程"   (finalI   1)   "启动，获得锁");
                        Thread.sleep(1000);
                        lockDemo.unLock();//解锁
                        System.out.println("线程"   (finalI   1)   "释放锁");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

  上面是通过API手动去加锁，在实际开发中可能会存在以下的问题

• 会话连接是异步的，需要自己去处理，例如使用CountDownLatch
• Watch需要重复注册，不然就不能生效
• 开发的复杂性比较高
• 不支持多节点的删除和创建，需要自己去递归处理
    
    为了解决上面的问题，Apache推出了一个框架：Curator，解决了原生Java API开发分布式遇到的问题。接下来使用Curator来测试zookeeper的分布式功能

  1. 导入jar包

<!-- 分布式锁框架依赖 -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.1</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.1</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-client</artifactId>
    <version>5.2.1</version>
</dependency>

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.8.0</version>
</dependency>

  2. 代码实现

public class CuratorTest {
    public static CuratorFramework getClient() {
        // 创建连接
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.0.132:2181,192.168.0.133:2181,192.168.0.134:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(new ExponentialBackoffRetry(3000, 3)).build();
        // 启动客户端
        client.start();
        System.out.println("zookeeper 启动成功");
        return client;
    }

    public static void main(String[] args) {
        for (int i = 0; i < 10; i  ) {
            // 创建分布式锁
            InterProcessMutex lock = new InterProcessMutex(getClient(), "/locks");
            int finalI = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        lock.acquire();// 获取锁
                        System.out.println("线程"   finalI   "获取到锁");
                        Thread.sleep(2000);
                        lock.release();// 释放锁
                        System.out.println("线程"   finalI   "释放锁");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

源码地址：https://gitee.com/peachtec/hxz-study

这篇好文章是转载于：学新通技术网