zookeeper

zookeeper入门

概述

zookeeper是一个开源的分布式的，为分布式应用提供协调（zookkeeper：动物管理员）服务的Apache项目

zookeeper工作机制

zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架。负责储存和管理大家都关心的数据。然后接受观察者的注册，一旦这些数据发生变化，zookeeper就负责通知已经在zookeeper上注册的那些观察者。然后观察者作出相应的反应。
$zookeeper=文件系统+通知机制$

特点

1个leader和多个fllower组成的集群
集群中只要有半数以上的节点存活。zookeeper就能正常服务
全局数据一致：每个server保存一份相同的数据副本。Client无论连接到哪个server，数据都是一致的
更新请求顺序进行，来自同一个Client的更新请求按其发送顺序依次进行
数据更新原子性，一次数据更新要么成功，要么失败
实时性：在一定的时间范围内，Client能读到最新的数据。zk同步数据非常快（数据量小）

数据结构

整体上可以看作是一棵树。每个节点称作一个ZNode。每个ZNode默认存储1MB（数据量小）的数据。每个ZNode通过其路径唯一标识。

数据结构

应用场景

统一命名服务，统一配置管理，统一集群管理，服务器节点动态上下线，软负载均衡

统一命名服务

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

/
- /service
  - www.baidu.com
    - 192.168.22.13
    - 192.168.22.14
    - 192.168.22.15

统一配置管理

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

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

配置管理可由zookeeper实现

可将配置信息写入zookeeper上的一个ZNode
各个客户端服务器监听这个ZNode
一旦ZNode中的数据被修改，zookeeper将通知各个客户端服务器

统一集群管理

分布式环境中，实时掌握每个节点的状态是必要的。zk可以根据节点实时状态作出调整。
zk可以实现实时掌握监控节点状态变化
1. 可以将节点信息写入zk的znode
2. 监听这个znode可获取他的实时状态变化

服务器动态上下线

服务端启动时注册信息（创建临时节点）
客户端获取到当前在线服务器列表。并且注册监听
服务器节点下线
Zk：服务器节点上下线事件通知
process()重新获取服务器列表，并且注册监听

负载均衡

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

/
- /server
  - 注册登陆服务
    - 192.168.22.13：访问数60
    - 192.168.22.14：访问数50
    - 192.168.22.15：访问数65

zookeeper安装（standalone）

https://zookeeper.apache.org/

安装JDK
解压zookeeper

配置修改

#将conf这个路径下的zoo_sample.cfg修改为zoo.cfg
mv zoo_sample.cfg zoo.cfg

#打开zoo.cfg修改dataDir路径
dataDir=/home/yowai/Desktop/module/zookeeper/tmp

#创建tmp文件夹
mkdir tmp

操作zookeeper

#启动
bin/zkServer.sh start
#查看状态
bin/zkServer.sh status
#关闭
bin/zkServer.sh stop

#启动客户端
zkCli.sh
#退出客户端
quit

配置参数解读

#通信心跳数，zookeeper服务器与客户端心跳时间，单位毫秒。
tickTime=2000

#LF初始通信时限。集群中的Follower跟随者服务器与Leader领导者服务器之间初始连接时能容忍的最多心跳数（tickTime的数量），用它来限定集群中的Zookeeper服务器连接到Leader的时限。刚开始启动的时候比较耗时间，所以时间要长一点
initLimit=10

#集群中Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
syncLimit=5

#数据文件目录+数据持久化路径
dataDir=/home/yowai/Desktop/module/zookeeper/tmp

#客户端连接端口
clientPort=2181

zookeeper内部原理

选举机制

半数机制：集群中半数以上机器存活，集群可用。所以zookeeper适合安装奇数台服务器
虽然没有在配置文件中指定master和slave。但是其工作的时候，是有一个节点为leader，其它则为follower。leader是通过内部的选举机制临时产生的

用1个简单的例子说明此过程：假设有5台服务器组成的集群。它们的id从1-5。同时它们又是同时启动的，也就是说没有历史数据，在存放数据量这个维度上都是一样的

服务器1启动，此时只有1台服务器启动，它发出的报文没有任何响应，所以它的选举状态一直是looking
服务器2启动。它与最开始启动的服务器1进行通信，互相交换自己的选举结果。由于没有历史数据，所以Id较大的服务器2胜出。但由于没有达到超过半数以上的服务器都同意选举他（此例子中半数以上是3）。所以服务器1，2还是保持looking状态
服务器3启动，根据前面的理论分析，服务器3成为1，2，3中的老大。此时有3台服务器选举了它，所以它成为此次选举的leader
服务器4启动，因为已经有服务器3为leader，所以服务器4为follower
服务器5与4同理

节点类型

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

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

image-20200407014749647

持久化目录节点：客户端与zookeeper断开连接后，该节点仍然存在
持久化顺序编号目录节点：客户端断开后，该节点仍然存在，只是zookeeper给该节点名称进行顺序编号。分布式系统中，顺序号可以被用于为所有事件进行全局排序，这样客户端可以通过编号推断事件的顺序
临时目录节点：客户端断开连接后，该节点被删除
临时顺序编号目录节点：客户端断开后，该节点被删除，只是zookeeper给该节点名称进行顺序编号。

分布式安装（集群）

解压ZK到module文件夹

#将zoo_sample.cfg文件复制为zoo.cfg，然后添加内容
vim /home/yowai/Desktop/module/zookeeper/conf/zoo.cfg

#修改路径
dataDir=/home/yowai/Desktop/module/zookeeper/tmp

#使用和选举用不同的端口。其中1，2，3是第几台服务器
server.1=slave2:2888:3888
server.2=slave3:2888:3888
server.3=slave4:2888:3888

配置参数解读：

server.A=B:C:D

A是一个数字，表示这是第几台服务器。集群模式下配置一个文件myid。这个文件在DataDir目录下，文件里的数据就是A的值，zk启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server
B是这个服务器的IP地址
C是服务器启动与集群中的Leader服务器交换信息的端口
D是玩意集群中的Leader挂掉了，需要另一个端口重新进行选举出新的Leader。这个端口就是用来执行选举时服务器相互通信的端口

创建/home/yowai/Desktop/module/zookeeper/tmp文件

#在以上目录下添加1。注意：slave1 添加1；slave2 添加2；slave3 添加1；
echo 1 > myid

添加环境变量

export JAVA_HOME=/home/yowai/Desktop/module/jdk
export HADOOP_HOME=/home/yowai/Desktop/module/hadoop
export ZOOKEEPER_JOME=/home/yowai/Desktop/module/zookeeper

export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$ZOOKEEPER_HOME/bin

分发ZK

scp -r zookeeper/ slave3:`pwd`
scp -r zookeeper/ slave4:`pwd`

修改其他节点的信息

更改权重

#在/home/yowai/Desktop/module/zookeeper/tmp目录下
#slave3
echo 2 > myid

#slave4
echo 3 > myid

添加环境变量

节点都添加

export JAVA_HOME=/home/yowai/Desktop/module/jdk
export HADOOP_HOME=/home/yowai/Desktop/module/hadoop
export ZOOKEEPER_HOME=/home/yowai/Desktop/module/zookeeper

export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$ZOOKEEPER_HOME/bin

测试ZK

先启动权重为1的节点

#在bin目录下启动
./zkServer.sh start
#查看状态
./zkServer.sh status

#关闭
zkServer.sh stop

启动权重为2的节点

权重为1的节点信息

image-20200328031456502

权重为2的节点信息

image-20200328031402453

此时即使启动权重为3的节点，它也是「fllower」,因为集群中已经有一个「leader」了，没必要再选

监听器原理

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

image-20200419041644294

常见的监听

监听节点数据的变化：get path [watch]
监听子节点增减的变化：ls path [watch]

写数据流程

image-20200419035919642

Client向zookeeper的server1上写数据，发送一个写数据的请求
如果server1不是leader，那么server1会把接收的请求进一步转发给leader。leader把「写请求」广播给各个server。各个server写成功后会通知给leader
当leader收到半数以上server数据写成功，那么就说明数据写成功了。此时leader会告诉server1数据写成功了
server1会进一步通知client数据写成功了，这时就认为操作成功

zookeeper实战

客户端命令操作

命令	功能
`help`	显示所有操作命令
`ls path [watch]`	使用`ls`命令查看当前znode中所包含的内容
`ls2 path [watch]`	查看当前节点数据并能看到更新次数等信息
`create`	普通创建；`-s`含有序列；`-e`临时
`get path [watch]`	获得节点的值
`set`	设置节点的值
`stat`	查看节点状态
`delete`	删除节点
`rmr`	递归删除节点

常用API

maven依赖

<?xml version="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.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>zookeeper</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zook eeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
        </dependency>
    </dependencies>


</project>

历史服务配

Resources/log4j.properties

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

代码

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class TestZookeeper {


    private String connectString="slave2:2181,slave3:2181,slave4:2181";
    private int sessionTimeOut=2000;
    private ZooKeeper zkClient;


    //连接客户端
    @Before
    public void init() throws IOException {

        zkClient=new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
              public void process(WatchedEvent watchedEvent) {
//
//                System.out.println("--------start--------");
//                List<String> children = null;
//                try {
//                    children = zkClient.getChildren("/", true);
//                    for (String child:children){
//                        System.out.println(child);
//                    }
//                    System.out.println("--------end--------");
//
//                } catch (KeeperException e) {
//                    e.printStackTrace();
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//
//
//
            }
        });
    }


    //创建节点
    @Test
    public void createNode() throws KeeperException, InterruptedException {
        String path = zkClient.create("/atguigu", "dahaigezuishuai".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(path);
    }

    //获取节点并监听节点变化
    @Test
    public void getDataAndWatch() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/", true);
        System.out.println("-------------");
        for (String child:children){
            System.out.println(child);
        }
        //不让进程结束
        Thread.sleep(Long.MAX_VALUE);

    }


    //判断节点是否存在
    @Test
    public void exist() throws KeeperException, InterruptedException {
        Stat stat = zkClient.exists("/atguigu", false);
        System.out.println(stat==null? "not exitss":"exists");
    }



}

监听服务器节点动态上下线

Server

import org.apache.zookeeper.*;

import java.io.IOException;

public class Server {
    private String connectString="slave2:2181,slave3:2181,slave4:2181";
    private int sessionTimeout=2000;
    private ZooKeeper zkClient;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {

        Server server = new Server();
        //1，连接zk集群
        server.getConnect();
        
        //2，注册节点并写数据
        server.regist(args[0]);
        //3，业务逻辑，常常的睡，不让进程结束
        server.bussiness();
        
        
    }

    private void bussiness() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    //短暂节点；带序号的节点
    private void regist(String hostname) throws KeeperException, InterruptedException {
        String path = zkClient.create("/servers/server", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname+" is online");
    }

    private void getConnect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {

            }
        });
    }
}

Client

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 Client {

    private String connectString="slave2:2181,slave3:2181,slave4:2181";
    private int sessionTimeout=2000;
    private ZooKeeper zkClient;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        Client client = new Client();


        //1，获取zk集群连接
        client.getConnect();

        //2，注册监听
        client.getChildren();

        //3，业务逻辑
        client.business();

    }

    private void getChildren() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/servers", true);
        ArrayList<String> hosts = new ArrayList<String>();
        for (String child: children) {
            byte[] data = zkClient.getData("/servers/" + child, false, null);
            hosts.add(new String(data));

        }
        //将所有在线数据打印到控制台
        System.out.println(hosts);
    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);

    }

    private void getConnect() throws IOException {
       zkClient= new  ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                try {
                    getChildren();
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });
    }
}