Zookeeper

Zookeeper是一个开源的分布式，为分布式框架提供协调服务（并不处理业务实现，而是辅助的）Apache项目，使用Java编写，支持Java和C两种编程语言

1. 在微服务做为注册中心
2. 搭建Hadoop、HBase集群，使用ZK作为集群管理者
3. ZK实现分布式锁：Redis也可以实现

ZK内存数据模型（树状）

1. 每个子目录如/node1都被称作一个znode(节点)。这个znode是被它所在的路径唯一标识
2. znode可以有子节点目录,并且每个znode可以存储数据
3. znode是有版本的。每个 znode中存储的数据可以有多个版本，也就是一个访问路径中可以存储多份数据。
4. znode可以被监控，包括这个目录节点中存储的数据的修改，子节点目录的变化等，一旦变化可以通知设置监控的客户端

节点的分类

持久结点（PERSISTENT）

在节点创建后，就一直存在，知道有删除操作来主动删除这个节点——不会因为创建该节点的客户端会话消失而消失

持久顺序结点（PERSISTENT_SEQUENTIAL）

与持久结点一致，额外特征是在ZK中，每个父节点为第一级结点维护一份时序，会记录每个结点创建的先后顺序。在创建子节点的时候，可以设置这个属性。在创建节点过程中，ZK会自动给节点名加上一个数字后缀作为新节点名，这个数字后缀是整数的最大值

临时结点(EPHEMERAL)

会随着客户端会话失效而清除。指会话失效而非连接断开。临时结点不能创建子节点

临时顺序结点(EPHEMERAL_SEQUENTIAL)

父节点为第一级子节点维护时序

安装JDK

1.官网下载JDK8-jdk-8u351-linux-x64.rpm

解压：rpm -ivh jdk-8u351-linux-x64.rpm

jdk默认安装路径：/usr/java

2.配置环境变量：vim /etc/profile ->G跳到末尾

追加:export JAVA_HOME=/usr/java/jdk-8u351-linux-x64

export PATH=$PATH:$JAVA_HOME/bin

3.重新加载

source /etc/profile

安装Zookeeper

1. 下载官网安装包并解压（下载编译后的安装包-带bin的）
2. 将zoo_sample.cfg改名mv zoo_sample.cfg zoo.cfg
3. 修改配置vim ~/zookeeper/conf/zoo.cfg

tickTime=2000 ：zk集群节点间每2秒心跳

initLimit=10：同步10次，一次2秒心跳。集群初始搭建时，集群节点同步超时时间10*2=20秒

syncLimit=5：集群在运行过程中同步数据超时时间5*2=10秒

dataDir：默认数据的存储位置（保证存储位置存在）

clientPort：zk服务监听的端口号

maxClinetCnxns：线程池线程数量默认60个(Redis=150)

autopurge.snapRetainCount=3：没生成3个快照合并快照

Autopurge.purgeInterval=1：1小时3个快照合并，0则不合并

4.开启zk ./zkServer.sh start ~/zookeeper/conf/zoo.cfg

5.启动客户端连接到zk

./bin/zkCli.sh (server 192.........:2181)

 安装并启动成功

客户端基本命令(有些已过时，具体看zookeeper提示)

 

节点监听机制watch：对节点目录、数据监听

客户端可以监测znode节点的变化，变化触发相应相应的事件，然后清楚对该节点的监测。当监测一个znode节点时，zookeeper会发送通知给监听节点。一个watch事件是一个一次性的触发器（触发过一次就失效），当呗设置watch的数据或目录发生改变，会将改变通知给watch的客户端

对节点目录监听：ls -w path

对节点数据监听：get -w path

 

 

zookeeper为什么可以作为注册中心？

利用zookeeper的监听机制

当服务连接上zookeeper时，会自动创建一个临时节点，数据为服务ip端口。集群的话，先创建持久节点，下面多个临时节点。当A其中一服务发生宕机时，zookeeper会清除其ip，同时通知B服务，将对用ip删除

Java操作Zookeeper

"1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>
 
    <groupId>groupIdgroupId>
    <artifactId>zookeeper2artifactId>
    <version>1.0-SNAPSHOTversion>
 
    <properties>
        <maven.compiler.source>8maven.compiler.source>
        <maven.compiler.target>8maven.compiler.target>
    properties>
 
    <dependencies>
        <dependency>
            <groupId>com.101tecgroupId>
            <artifactId>zkclientartifactId>
            <version>0.11version>
        dependency>
 
        <dependency>
            <groupId>junitgroupId>
            <artifactId>junitartifactId>
            <version>4.13.1version>
            <scope>compilescope>
        dependency>
 
    dependencies>
    
project>

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
 
import java.io.IOException;
import java.util.List;
 
public class Main {
    private ZkClient zkClient;
    @Before
    public void before(){
        //获取连接，初始化客户端对象
        //1.ip地址 2.会话超时时间 3.连接超时时间 4.序列化方式
        zkClient  = new ZkClient("192.168.235.135:2181",60000*30,60000,new SerializableSerializer());
    }
 
    //创建节点
    @Test
    public void testCreateNode(){
        //1.持久节点
        zkClient.create("/node1","p", CreateMode.PERSISTENT);
        zkClient.create("/node1/node1_1","p", CreateMode.PERSISTENT);
        //1.持久顺序节点
        zkClient.create("/node2","ps", CreateMode.PERSISTENT_SEQUENTIAL);
        //1.临时节点
        zkClient.create("/node3","e", CreateMode.EPHEMERAL);
        //1.临时顺序节点
        zkClient.create("/node4","es", CreateMode.EPHEMERAL_SEQUENTIAL);
    }
 
    //删除节点
    @Test
    public void testDeleteNode(){
        zkClient.delete("/node1");//删除没有子节点的节点
 
        zkClient.deleteRecursive("/node1");//删除有子节点的节点，使用递归删除
    }
 
    //查看节点子节点
    @Test
    public void testFindNode(){
        List children = zkClient.getChildren("/");
        for(String c: children)
            System.out.println(c);
    }
 
    //修改，查看数据
    @Test
    public void testGetNode(){
        //通过Java客户端操纵，需要保证节点存储的数据 和 获取节点时数据的序列化必须一致
        //所以不要通过客户端设置值，java取值
        zkClient.writeData("/node1","a");
        Object data = zkClient.readData("/node1");
        System.out.println(data);
    }
 
    //获取节点状态和数据
    @Test
    public void testFindNodeStat(){
        Stat stat = new Stat();
        Object readData = zkClient.readData("/node1",stat);
        System.out.println("readData:"+readData);
        System.out.println("stat:"+stat);
        System.out.println(stat.getCtime());
        System.out.println(stat.getAversion());
        System.out.println(stat.getCversion());
        System.out.println(stat.getCzxid());
        System.out.println(stat.getDataLength());
        System.out.println(stat.getEphemeralOwner());
        System.out.println(stat.getMtime());
        System.out.println(stat.getMzxid());
    }
 
    //监听节点（数据）：永久监听
    @Test
    public void  WatchData() throws IOException {
        zkClient.subscribeDataChanges("/node1", new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                //当前节点数据变化触发
                System.out.println("路径："+s);
                System.out.println("修改后数据："+o);
            }
 
            @Override
            public void handleDataDeleted(String s) throws Exception {
                //节点删除触发
                System.out.println("删除的节点："+s);
            }
        });
        System.in.read();//阻塞，保持一直运行
    }
 
    @Test
    public void  WatchChilds() throws IOException {
        //监听目录
        zkClient.subscribeChildChanges("node1", new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List list) throws Exception {
                //目录改变触发
                System.out.println("父节点"+s);
                for(String l:list)
                    System.out.println("发生变更的孩子节点："+l);
            }
        });
        System.in.read();//阻塞，保持一直运行
    }
 
    @After
    public void after(){
        //释放资源
        zkClient.close();//因为调用close，所以创建的临时节点直接失效
    }
}

 

ZK的集群

集群cluster：集合同一种软件服务的多个节点同时提供服务

所有节点配置必须保持一致，后选举leader节点（至少三台服务器，最好奇数）

集群解决的问题：

1. 单节点的并发访问的压力问题
2. 单节点故障问题（如硬件老化，自然灾害等）

客户端连接集群中任意节点

如果数据不一致：zk利用原子广播协议zab，维持zookeeper数据高度一致性

原理：对集群中一个节点操作写操作时，会向leader节点进行确认，如果leader确认后会向所有节点原子广播写入当前数据，如果有节点写入失败，全部节点撤销写操作回滚

搭建ZK集群

搭建集群3个节点（因为在一个服务器上，三个不同的端口）

在根目录下分别创建节点存放数据的目录

mkdir zkdata1 zkdata2 zkdata3

在各目录下新建文件myid用于区别其他服务的唯一标识id

touch zkdata1/myid zkdata2/myid zkdata3/myid

echo "1" >> zkdata1/myid

echo "2" >> zkdata2/myid

echo "3" >> zkdata3/myid

分别编辑他们的配置，修改dataDir和端口clientPort

vim ./zkdata1/zoo.cfg

vim ./zkdata2/zoo.cfg

vim ./zkdata3/zoo.cfg

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/root/zkdata2

clientPort=4001

server.1=192.168.235.135:3002:3003

server.2=192.168.235.135:4002:4003

server.3=192.168.235.135:5002:5003

分别集群

./zookeeper/bin/zkServer.sh start ./zkdata1/zoo.cfg

./zookeeper/bin/zkServer.sh start ./zkdata2/zoo.cfg

./zookeeper/bin/zkServer.sh start ./zkdata3/zoo.cfg

分别开启三个会话，分别连接

./zookeeper/bin/zkCli.sh server 192.168.235.135:3001

./zookeeper/bin/zkCli.sh server 192.168.235.135:4001

./zookeeper/bin/zkCli.sh server 192.168.235.135:5001

可以使用命令查看谁是leader、follows：

./zookeeper/bin/zkServer.sh status ./zkdata1/zoo.cfg

./zookeeper/bin/zkServer.sh status ./zkdata2/zoo.cfg

./zookeeper/bin/zkServer.sh status ./zkdata3/zoo.cfg

需要注意的是：java操作集群和操作单节点一致，虽然集群数据保持高度一致，但是初始化客户端对象时，zkServer最好全部写上，假设只写一个，如果该zkServer宕机，则无法用java操作集群

@Before
    public void before(){
        //获取连接，初始化客户端对象
        //1.ip地址 2.会话超时时间 3.连接超时时间 4.序列化方式
        zkClient  = new ZkClient("192.168.235.135:3001,192.168.235.135:4001,192.168.235.135:5001",60000*30,60000,new SerializableSerializer());
    }