Zookeeper-环境搭建

下载

https://archive.apache.org/dist/zookeeper/

选择`3.5.7` 稳定下载,上传`linux`

也可以在 linux 上使用wget 下载 wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz

本地模式安装

安装JDK

`JDK` 选择

# wget 进行 JDK 下载
wget https://cdn.azul.com/zulu/bin/zulu8.64.0.15-ca-jdk8.0.342-linux_x64.tar.gz
# 创建目录
mkdir -p /opt/model/jdk8/
# 解压到指定目录下
tar -zxvf zulu8.64.0.15-ca-jdk8.0.342-linux_x64.tar.gz -C /opt/model/jdk8/
# 进入目录

复制(淡黄色背景的路径)如下路径，用于配置环境变量

# 配置环境变量
vim /etc/profile
 export JAVA_HOME= 路径
 export JRE_HOME=${JAVA_HOME}/jre
 export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
 export PATH=${JAVA_HOME}/bin:$PATH
 
# 保存退出
# 使其生效
source /etc/profile

环境检测

1	java -version

如图信息则已正确安装`JDK`

拷贝apache-zookeeper-3.5.7-bin.tar.gz 到Linux 系统下

解压到指定目录

# 创建目录
mkdir -p zookeeper-3.5.7/
# 解压
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/model/zookeeper-3.5.7/
# 修改名称
 mv apache-zookeeper-3.5.7-bin/ zookeeper3.5.7

配置

修改数据文件存储路径

cd /opt/model/zookeeper-3.5.7/zookeeper3.5.7/conf/
# 创建数据文件存储目录
mkdir zookeeperData
# 获取完整路径信息
pwd

修改前	修改后

修改配置文件名称

原因

启动时文件名称报错

修改配置文件名称

# 方法一
cp zoo_sample.cfg zoo.cfg
# 方法二
mv zoo_sample.cfg zoo.cfg

启动测试

成功启动服务	通过客户端连接服务端

Zookeeper基本使用命令

如下路径的如下文件

启动服务端

1	./zkServer.sh start

停止服务端

1	./zkServer.sh stop

状态查看

1	./zkServer.sh status

启动客户端

1	./zkCli.sh

退出客户端

quit

重启

1 2	# 执行了两步: 先停止后启动 ./zkServer.sh restart

数据模型

Zookeeper 是一个树形目录服务,其数据模型和Unix 的文件系统目录树很类似,拥有一个层次化结构

以节点分布

这里的每一个节点都被称为Znode 每个节点上都会保存自己的数据和节点信息
节点可以拥有子节点，同时也允许少量(1MB)数据存储在该节点之下
节点可以分为四大类
- PERSISTENT 持久化节点
- EPHEMERAL 临时节点:-e
- PERSISTENT_SEQUENTIAL 持久化顺序节点： -s
- EPHEMERAL_SEQUENTIAL 临时顺序节点: -es

客户端命令

通过客户端连接服务端

./zkCli.sh -server localhost:2181

# 连接本机可以简写
./zkCli.sh

配置参数解读(`zoo_sample.cfg`)

tickTime=2000

通信心跳时间，Zookeeper 服务器与客户端心跳的时间,单位毫秒
initLimit=10

LF: 初始通信时限

Leader 和Follower 初连接接时间时能容忍的最多心跳数tickTime的数量
syncLimit=5

LF: 同步通信时限

Loader 和Follower 之间通信时间如果超过syncLimit * tickTime,Leader 认为Follower 死掉,从服务器列表删除Follwer
dataDir

保存Zookeeper 中的数据

注意: 默认是temp 目录,容易被Linux 系统定期删除,所以一般不用默认地tmp 目录
clientPort=2181

客户端连接端口,通常不做修改

客户端操作

可操作命令

addauth scheme auth
       close
       config [-c] [-w] [-s]
       connect host:port
       create [-s] [-e] [-c] [-t ttl] path [data] [acl]
       delete [-v version] path
       deleteall path
       delquota [-n|-b] path
       get [-s] [-w] path
       getAcl [-s] path
       history
       listquota path
       ls [-s] [-w] [-R] path
       ls2 path [watch]
       printwatches on|off
       quit
       reconfig [-s] [-v version] [[-file path] | [-members serverID=host:port1:port2;port3[,...]*]] | [-add serverId=host:port1:port2;port3[,...]]* [-remove serverId[,...]*]
       redo cmdno
       removewatches path [-c|-d|-a] [-l]
       rmr path
       set [-s] [-v version] path data
       setAcl [-s] [-v version] [-R] path acl
       setquota -n|-b val path
       stat [-w] path
       sync path

查看节点信息

1 2	# 查看根目录 ls /

查看zookeeper 节点下的信息

1	ls /zookeeper

创建节点

# 指定在那个节点下面创建
# 创建在根节点下
create /coderitl
# 创建节点并存储数据(存在的节点不能直接覆盖)
create /coderitl 我是数据

获取数据

1	get /coderitl

存储数据

1	set /coderitl

删除节点

delete /coderitl

# 创建一个嵌套节点
create /coderitl/test
# 删除带有子节点的节点
deleteall /coderitl

创建临时节点

1 2	# 会话关闭则自动删除 create -e /userinfo

创建顺序节点

1	create -s /app1

创建持久化的顺序节点

1	create -es /app2

查看详细信息

ls -s /

mZxid 最后一次被更新的事务id
mtime =修改时间
pZxid =子节点列表最后一次被更新的事务id
cversion 子节点的版本号
dataVersion 数据版本号
aclVersion 权限版本号
ephemeralOwner 用于临时节点，代表临时节点的事务id，如果为持久节点则为0
dataLength 节点存储的数据长度
numChildren 当前节点的子节点个数

网络重启失败解决方案

报错

服务重启失败

解决方案

systemctl stop NetworkManager
systemctl disable NetworkManager
# 重启network
systemctl start network.service

查看

成功解决

ZooKeeper JavaAPI 操作

建立连接
添加节点
删除节点
修改节点
查询节点
Watch 事件监听
分布式锁实现

版本问题

Zookeeper 版本为3.5.x 以上需要使用Curator 4.x.x+
创建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>curator-zk</artifactId>
  <version>1.0-SNAPSHOT</version>
  <packaging>war</packaging>


  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <!-- curator -->
    <dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-recipes</artifactId>
      <version>4.0.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-framework</artifactId>
      <version>4.0.0</version>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-api</artifactId>
      <version>1.7.36</version>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.36</version>
    </dependency>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter</artifactId>
      <version>RELEASE</version>
      <scope>compile</scope>
    </dependency>
  </dependencies>


</project>

测试连接

// 第一种连接方式
@Test
public void testCuratorConnection1() {
    /**
         * String connectString, 连接字符串: zookeeperIP:port 集群使用逗号隔开
         * int sessionTimeoutMs, 会话超时时间，默认: 60*1000ms
         * int connectionTimeoutMs, 连接超时时间: 15*1000
         * RetryPolicy retryPolicy 重试策略
         */
    // 重试策略
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
    // 第一种连接方式
    CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.247.130:2181", retryPolicy);
    // 开启连接
    client.start();
}

连接方式二

package com.coderitl.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

public class TestCurator {
    private CuratorFramework client;

    // 第二种连接方式
    @BeforeEach
    public void testCuratorConnection2() {
        // 重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        client = CuratorFrameworkFactory.builder().connectString("192.168.247.130").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).namespace("coderitl").build();
        // 开启连接
        client.start();
        System.out.println("testCuratorConnection2 " + client);
    }

    @AfterEach
    public void close() {
        if (client != null) {
            client.close();
        }
    }

    /**
     * 创建节点: create 持久 临时 顺序 数据
     * 1. 基本创建
     * 2. 创建节点 带有数据
     * 3. 设置节点的类型
     * 4. 创建多级节点 /app1/p1 同时存在
     */
    @Test
    public void testCreateNode() throws Exception {
        System.out.println("client = " + client);
        // 1. 如果创建节点,没有指定数据,则默认将当前客户端的 ip 作为数据存储
        String forPath = client.create().forPath("/app");
        System.out.println("forPath = " + forPath);
    }

    @Test
    public void testCreateNodeAndData() throws Exception {
        System.out.println("client = " + client);
        // 2. 创建节点并带有数据
        String forPath = client.create().forPath("/app2", "data info".getBytes());
        System.out.println("forPath = " + forPath);
    }

    @Test
    public void testCreateNodeEphemeal() throws Exception {
        System.out.println("client = " + client);
        // 3. 设置节点的类型
        // 默认类型: 持久化
        String forPath = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
        System.out.println("forPath = " + forPath);
    }

    @Test
    public void testCreateNode4() throws Exception {
        System.out.println("client = " + client);
        // 4. 创建多级节点
        // creatingParentsIfNeeded: 如果父节点不存在,则创建父节点
        String forPath = client.create().creatingParentsIfNeeded().forPath("/app3/app2");
        System.out.println("forPath = " + forPath);
    }
}

创建节点并带有数据

创建节点并带有数据

创建带有类型的节点

创建带有类型的节点

虚拟机中Zookeeper 数据结果

结果

查询

/**
     * 查询节点:
     * 1. 查询数据: get
     * 2. 查询子节点 ls
     * 3. 查询节点状态信息 ls -s
     */

    @Test
    public void testSelectNode1() throws Exception {
        System.out.println("client = " + client);
        // 查询数据 get
        byte[] data = client.getData().forPath("/app");
        System.out.println(new String(data));
    }

    @Test
    public void testSelectNode2() throws Exception {
        System.out.println("client = " + client);
        // 查询子节点
        List<String> strings = client.getChildren().forPath("/app3");
        System.out.println(strings);
    }

    @Test
    public void testSelectNode3() throws Exception {
        System.out.println("client = " + client);
        Stat status = new Stat();
        System.out.println("status = " + status);
        // 查询节点的状态信息: ls -s xxx
        client.getData().storingStatIn(status).forPath("/app3");
        System.out.println("status = " + status);
    }

查询多级节点

查询多级节点

基本修改

1	client.setData().forPath("/app3","set data".getBytes());

修改数据(根据版本修改数据)

/**
     * 修改数据
     */
@Test
public void testSetData() throws Exception {
    Stat status = new Stat();
    // 查询节点的状态信息: ls -s
    client.getData().storingStatIn(status).forPath("/app3");
    int version = status.getVersion(); // 查询的 version
    client.setData().withVersion(version).forPath("/app3","set data222".getBytes());
}

删除节点

删除单个节点

@Test
public void testDeleteOneNode() throws Exception {
    client.delete().forPath("/app");
}

删除带有子节点的节点

@Test
public void testDeleteAllOneNode() throws Exception {
    client.delete().deletingChildrenIfNeeded().forPath("/app3");
}

必须成功的删除: 为了防止网络抖动,本质就是重试

@Test
public void testDeleteNode() throws Exception {
    client.delete().guaranteed().forPath("/app2");
}

回调

@Test
public void testDeleteNodeCall() throws Exception {
    client.delete().guaranteed().inBackground(new BackgroundCallback() {
        @Override
        public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
            System.out.println("我被删除了..........");
        }
    }).forPath("/app2");
}

watch 事件监听

Zookeeper 允许用户在指定节点上注册一些Watcher 并且在一些特定事件触发的时候,Zookerper 服务端会将事件通知到感兴趣的客户端上去,该机制是Zookeeper 实现分布式协调服务的重要特性
Zookeeper 中引入了Watcher 机制来实现发布订阅功能,能够让多个订阅者同时监听某一个对象,当一个对象自身状态变化时,会通知所有订阅者

zookeeper 提供了三种Watcher

NodeCache 只是监听某一个特定的节点

/**
    * 给指定的一个节点注册监听器
    *
    * @throws Exception
    */
   @Test
   public void testNodeCache() throws Exception {
       // 创建 NodeCache 对象
       NodeCache nodeCache = new NodeCache(client, "/app1");
       // 注册监听
       nodeCache.getListenable().addListener(new NodeCacheListener() {
           @Override
           public void nodeChanged() throws Exception {
               System.out.println("节点变化了.....");
               // 获取修改后的数据
               byte[] data = nodeCache.getCurrentData().getData();
               System.out.println(new String(data));
           }
       });
       // 开启监听,如果设置为 true,则开启监听 加载缓冲数据
       nodeCache.start(true);
       while (true) {
       }
   }

PathChildrenCache 监控一个ZNode 的子节点

/**
    * PathChildrenCache: 监听某个节点的所有子节点们
    *
    * @throws Exception
    */
   @Test
   public void testPathChildrenNode() throws Exception {
       // 创建监听对象
       PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);
       // 绑定监听器
       pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
           @Override
           public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
               System.out.println("子节点变化了.....");
               System.out.println(event);
               // 监听子节点的数据变更,并且拿到变更后的数据
               // 获取类型
               PathChildrenCacheEvent.Type type = event.getType();
               // 判断类型是否是 update
               if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                   byte[] data = event.getData().getData();
                   System.out.println(new String(data));
               }
           }
       });
       // 开启
       pathChildrenCache.start();
       while (true) {
       }
   }

TreeCache 可以监控整个树上的所有节点,类似于PathChildrenCache 和NodeCache 的组合

/**
    * PathChildrenCache: 监听某个节点自己和所有子节点们
    *
    * @throws Exception
    */
   @Test
   public void testTreeCache() throws Exception {
       TreeCache treeCache = new TreeCache(client, "/app2");
       treeCache.getListenable().addListener(new TreeCacheListener() {
           @Override
           public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
               System.out.println();
               // 监听子节点的数据变更,并且拿到变更后的数据
               // 获取类型
               
           }
       });
   }

分布式锁原理

核心思想: 当客户端要获取锁,则创建节点，使用完锁,则删除该节点

原理

客户端获取锁时,在lock 节点下创建临时顺序节点
然后获取lock 下面的所有子节点,客户端获取到所有的子节点之后,如果发现自己创建的子节点序号最小,那么就认为该客户端获取到了锁,使用完锁后,将该节点删除
如果发现自己创建的节点并非lock 所有子结点中最小的,说明自己还没有获取到锁,此时客户端需要找到比自己小的那个节点,同时对其注册事件监听器,监听删除事件

如果发现比自己小的那个节点被删除,则客户端的Watcher 会收到相应的通知,此时再次判断自己创建的节点是否是lock 子节点中序号最小的,如果时则获取到了锁,如果不是则重复以上步骤继续获取比自己小的一个节点并注册监听

分布式锁

在Curator 中有五种锁方案
- InterProcessSemaphoreMutex:分布式排他锁(非可重入锁)
- InterProcessMutex：分布式可重入排他锁
- InterProcessReadWriteLock：分布式读写锁
- InterProcessMultiLock:将多个锁作为单个实体管理的容器
- InterProcessSemaphoreV2:共享信号量

12306 售票

分析

多线程模拟

public class Tick12306 implements Runnable {
    // 多线程
    private int ticket = 10; // 被共享的数据=> 票

    @Override
    public void run() {
        // 实现买票
        while (true) {
            if (ticket > 0) {
                System.out.println(Thread.currentThread() + ": " + ticket);
                // 票数减少
                ticket--;
            }
        }
    }
}

public class ClientTicket {
    public static void main(String[] args) {
        Tick12306 tick12306 = new Tick12306();
        // 代理商卖票
        Thread t1 = new Thread(tick12306,"携程");
        Thread t2 = new Thread(tick12306,"飞猪");
        // 启动线程
        t1.start();
        t2.start();
    }
}

资源错误

解决方案=> 分布式锁(syncxx将不适用)

package com.coderitl.curator;

import org.apache.curator.RetryPolicy;
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;

import java.util.concurrent.TimeUnit;

public class Tick12306 implements Runnable {
    // 多线程
    private int ticket = 10; // 被共享的数据=> 票
    private InterProcessMutex lock;

    public Tick12306() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.247.130")
                .sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy).build();
        // 开启连接
        client.start();
        // 初始化 lock
        lock = new InterProcessMutex(client, "/lock");
    }

    @Override
    public void run() {
        // 实现买票
        while (true) {
            try {
                // 获取锁
                lock.acquire(3, TimeUnit.SECONDS);
                if (ticket > 0) {
                    System.out.println(Thread.currentThread() + ": " + ticket);
                    // 票数减少
                    ticket--;
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                // 释放锁
                try {
                    lock.release();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

使用分布式锁解决超卖

集群

集群

Leader 的选举
- Serverid:服务器ID,比如有三台服务器,编号分别是1,2,3 编号越大在选择算法中的权重越大
- Zxid：数据ID，服务器中存放的最大数据ID，值越大说明数据越新，在选举算法中数据越新权重越大
- 在Leader 选举的过程中,如果某台Zookeeper 获得了超过半数的选票,则此Zookeeper 就可以成为Leader 了

伪集群环境搭建(以端口号区分)

在虚拟机新建目录,不干扰之前的Zookeeper

# 创建存放目录
mkdir -p /usr/local/zookeeper-cluster
# 进入该目录
cd /usr/local/zookeeper-cluster
# 下载 zookeeper
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz
# 解压
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /usr/local/zookeeper-cluster/

操作

1
2
3

cp -r apache-zookeeper-3.5.7-bin zookeeper-1
cp -r apache-zookeeper-3.5.7-bin zookeeper-2
cp -r apache-zookeeper-3.5.7-bin zookeeper-3

拷贝三份

创建数据存储目录

mkdir zookeeper-1-data
mkdir zookeeper-2-data
mkdir zookeeper-3-data

修改如下

修改数据文件路径和对应端口号

配置集群

在每个zookeeper 的data 目录下创建一个myid 文件,内容分别是1,2,3 这个文件就是记录每个服务器的ID

1
2
3

echo 1 >/usr/local/zookeeper-cluster/zookeeper-1/zookeeper-1-data/myid
echo 2 >/usr/local/zookeeper-cluster/zookeeper-2/zookeeper-2-data/myid
echo 3 >/usr/local/zookeeper-cluster/zookeeper-3/zookeeper-3-data/myid

在每一个zookeeper 的zoo.cfg 配置客户端访问端口clientPort 和集群服务器IP 列表

# 集群 ip
vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg
vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

# 每个 zoo.cfg  添加的都是三条
server.1=192.168.247.130:28812:3881
server.2=192.168.247.130:28813:3882
server.3=192.168.247.130:28814:3883

# server。服务器id=服务器ip地址:服务器之间通信端口:服务器之间投票选举端口

启动集群

启动集群就是分别启动每个实例

1
2
3

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh start
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh start

启动集群

集群后的状态

集群后的状态	单机角色信息

故障测试

首先我们先测试如果是从服务器挂掉,会怎么样

把3 号服务器停掉,观察1 号和2 号,发现状态并没有变化

/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh stop

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

# 3个节点的集群，从服务器挂掉,集群正常

我们再把1 号服务器(从服务器)也停掉,查看2 号（主服务器）的状态,发现已经停止运行了

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh stop

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

# 由此得出结论,3 个节点的集群,2个从服务器都挂掉,主服务器也无法运行,因为可运行的机器没有超过集群总数量的半数

我们再次把1 服务器启动起来,发现2 号服务器又开始正常，而且依然是领导者

1
2
3

/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh start

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh status

我们把3 号服务器也启动起来,把2 号服务器停掉,停掉后观察1 、3 号服务器的状态

/usr/local/zookeeper-cluster/zookeeper-2/bin/zkServer.sh stop


/usr/local/zookeeper-cluster/zookeeper-1/bin/zkServer.sh status
/usr/local/zookeeper-cluster/zookeeper-3/bin/zkServer.sh status

# 结果是=> 产生新的 leader

Zookeeper 集群角色
1. Leader 领导者
  - 处理事务请求
  - 集群内部个服务器的调度者
2. Follower 跟随着
  - 处理客户端非事务请求,转发给事务请求给Leader 服务器
  - 参与Leader 选取投票
3. Observer 观察者
  - 处理客户端非事务请求,转发事务请求给Leader 服务器