学习笔记：快速入门ZooKeeper技术

本博客介绍了ZooKeeper的相关概念、ZooKeeper的安装与配置、ZooKeeper的常用命令操作、ZooKeeper的ACL权限控制、ZooKeeper的JavaAPI的使用、ZooKeeper的集群搭建。

YanZhaoHanWei_555

1925人浏览 · 2022-04-20 22:05:34

YanZhaoHanWei_555 · 2022-04-20 22:05:34 发布

学习视频：黑马程序员 ZooKeeper 视频教程，快速入门 ZooKeeper 技术
学习资料：黑马程序员 公众号提供的文档资料链接 | 提取码：dor4）

本文最后更新于 2022-04-25，若内容或图片失效，请留言反馈。部分素材来自网络，若不小心影响到您的利益，请联系博主删除。

ReadMe
- 大体上是照搬了公众号提供的资料的。
- 写博客的原因：写写笔记，做做记录，巩固知识，方便自己在线阅读。
- 环境：Win10、JDK1.8、IntelliJ IDEA 2019.3.3 x64、VMWare16、CentOS7、apache-zookeeper-3.5.6
- 相关前置博客：《CentOS7下载安装》《CentOS7卸载OpenJDK，安装JDK》《Dubbo快速入门》
- 本博客介绍了 ZooKeeper 的相关概念、ZooKeeper 的安装与配置、ZooKeeper 的常用命令操作、ZooKeeper 的 ACL 权限控制、ZooKeeper 的 JavaAPI 的使用和 Zookeeper 的集群搭建。

文章目录

1.ZooKeeper的概念
2.ZooKeeper的安装配置
- 2.1.ZooKeeper的下载安装
- 2.1.ZooKeeper的配置启动
3.ZooKeeper命令操作
4.Zookeeper的ACL权限控制
5.ZooKeeper JavaAPI操作
6.ZooKeeper集群搭建

1.ZooKeeper的概念

Zookeeper 是 Apache Hadoop 项目下的一个子项目，是一个树形目录服务。
Zookeeper 翻译过来就是动物园管理员，他是用来管 Hadoop（大象）、Hive（蜜蜂）、Pig（小猪）的管理员。简称 zk。
Zookeeper 是一个分布式的、开源的分布式应用程序的协调服务。
Zookeeper 提供的主要功能包括：配置管理、分布式锁、集群管理。

在这里插入图片描述

ZooKeeper的数据结构
- zookeeper 的数据结点可以视为树状结构（或者目录），树中各节点成为 znode ，一个 znode 可以有多个子结点。
- zookeeper 结点在结构上表现为树状，使用路径来定位某个 znode。
- znode 兼具文件和目录两种特点
  - 既像文件一样维护着数据、元信息、ACL、时间戳等数据结构
  - 又像目录一样可以作为路径标识的一部分。
- znode 大体上分为三部分（使用 get 命令查看）
  - 结点的数据
  - 结点的子结点
  - 结点的状态
- 结点类型（在创建时被确定且不能更改）
  - 临时结点：生命周期依赖于创建它的会话，会话结束，结点将被自动删除。临时结点不允许拥有子结点。
  - 持久化结点：生命周期不依赖于会话，只有在客户端显式执行删除操作时才被删除。

2.ZooKeeper的安装配置

2.1.ZooKeeper的下载安装

1.环境准备

ZooKeeper 服务器是用 Java 创建的，它运行在 JVM 之上。需要安装 JDK 7 或更高版本。

文档中所提供的是 apache-zookeeper-3.5.6，所需 JDK 版本是1.8。

另外，本文这里是在 CentOS7 上安装 JDK ，此时需先卸载其自带的 openJDK。

具体操作可见我之前写的博客：《Centos 7 下载 + 安装 + 更换国内源》《CentOS7卸载openJDK，安装jdk1.8》

2.上传

这里可以使用工具（如 putty、SecureCRT 、MobaXterm 等软件）将压缩包传入虚拟机内。

注意：这里不要直接拖拽到虚拟机。直接拖拽文件至虚拟机内，可能会对文件造成一定损失，导致安装启动失败。

3.解压安装

将上传的的 ZooKeeper 放到 /opt/ZooKeeper 目录下，在该目录下解压。

# 创建 zooKeeper 目录
mkdir /opt/zooKeeper

# 将 zookeeper 安装包移动到 /opt/zooKeeper 目录下
mv apache-zookeeper-3.5.6-bin.tar.gz /opt/zookeeper/

# 解压
tar -zxvf apache-ZooKeeper-3.5.6-bin.tar.gz

2.1.ZooKeeper的配置启动

1.配置 zoo.cfg

进入到 conf 目录

cd /opt/zooKeeper/apache-zooKeeper-3.5.6-bin/conf/

在 conf 目录下，拷贝 zoo_sample.cfg 文件，重命名为 zoo.cfg

cp zoo_sample.cfg zoo.cfg

使用 vim 修改 zoo.cfg 文件

vim /opt/zooKeeper/apache-zooKeeper-3.5.6-bin/conf/zoo.cfg

在 zoo.cfg 文件内添加如下两行数据

dataDir=/opt/zooKeeper/zkdata # 配置 ZK 的数据目录
dataLogDir=/opt/zooKeeper/zkdatalog # 配置 ZK 的日志目录

进入 /opt/zooKeeper/ 目录

cd /opt/zooKeeper/

在 /opt/zooKeeper/ 目录下创建两个文件夹：zkdata、zkdatalog

mkdir zkdata

mkdir zkdatalog

2.启动、停止 zookeeper

进入 zookeeper 的 bin 目录，启动服务命令

./zkServer.sh start

停止服务命令

./zkServer.sh stop

查看服务状态：

./zkServer.sh status

3.ZooKeeper命令操作

3.1.ZooKeeper数据模型

ZooKeeper 是一个树形目录服务,其数据模型和 Unix 的文件系统目录树很类似，拥有一个层次化结构。
这里面的每一个节点都被称为： ZNode，每个节点上都会保存自己的数据和节点信息。
节点可以拥有子节点，同时也允许少量（1MB）数据存储在该节点之下。
节点可以分为四大类：
- PERSISTENT 持久化节点
- EPHEMERAL 临时节点：-e
- PERSISTENT_SEQUENTIAL 持久化顺序节点：-s
- EPHEMERAL_SEQUENTIAL 临时顺序节点：-es

在这里插入图片描述

3.2.ZooKeeper服务端常用命令

启动 ZooKeeper 服务

./zkServer.sh start

查看 ZooKeeper 服务状态:

./zkServer.sh status

停止 ZooKeeper 服务:

./zkServer.sh stop

重启 ZooKeeper 服务

./zkServer.sh restart

下图是在已经连接上 CentOS7 的 MobaXterm 界面上的操作

在这里插入图片描述

3.3.ZooKeeper客户端常用命令

ZooKeeper 的基本的 CRUD 操作、创建临时循序节点

3.3.1.基本的CRUD

连接 ZooKeeper 服务端

./zkCli.sh -server ip:port

断开连接

quit

显示指定目录下的节点

ls 目录

查看命令帮助

help

创建节点

create /节点path value

获取节点值

get /节点path # 使用 get 查看结点的属性和数据

stat /节点path # 使用 stat 只返回属性 没有数据

设置节点值

set /节点path value

删除单个节点

delete /节点path

删除带有子节点的节点

deleteall /节点path

这里是 bin 目录下的文件

在这里插入图片描述

3.3.2.监听器（维护配置信息）

 # 使用此命令注册的监听器能够在节点内容发生改变的时候，向客户端发出通知。
 # 需要注意的是：Zookeeper 的触发器是一次性的（one-time trigger），即触发一次就会立即失效
get path watch

# 使用此命令注册的监听器能够在节点状态发生改变时，向客户端发出通知
stat path watch

# 能够监听该节点下所有子节点的增加和删除操作
ls/ls2 path watch # 可以监听到子结点

在这里插入图片描述

3.3.3.创建临时循序节点

创建临时节点

create -e /节点path value

创建顺序节点

create -s /节点path value

查询节点详细信息

ls -s /节点path

节点信息介绍

czxid：节点被创建的事务 ID
ctime：创建时间
mzxid：最后一次被更新的事务 ID
mtime：修改时间
pzxid：子节点列表最后一次被更新的事务 ID
cversion：子节点的版本号
dataversion：数据版本号
aclversion：权限版本号
ephemeralOwner：用于临时节点，代表临时节点的事务 ID，如果为持久节点则为 0
dataLength：节点存储的数据的长度
numChildren：当前节点的子节点个数

4.Zookeeper的ACL权限控制

1.概述

Zookeeper 类似文件系统，client 可以创建节点、更新节点、删除节点。
Zookeeper 的 access control list 访问控制列表可以控制节点的权限。
ACL权限控制，使用 scheme：id：permission 来标识，主要涵盖三个方面：
- 权限模式（scheme）：授权的策略
- 授权的对象（id）：授权的对象
- 权限（permission）：授予的权限

2.特性

Zookeeper 的权限控制是基于每个 ZNode 节点，需要对每个节点设置权限
每个 ZNode 支持设置多种权限控制方案和多个权限
子节点不会继承父节点的权限，客户端无权访问某节点，但可能可以访问他的子节点

3.scheme：权限模式

描述	方案
world	只有一个用户：anyone，代表登录 Zookeeper 的所有人（默认）
ip	对客户端使用 IP 地址认证
auth	使用已添加认证的用户认证
digest	使用用户名 : 密码方式认证

4.id：授权对象

授权对象 id 是指：权限赋予的实体
例如：IP地址或用户

5.permission：授予的权限

权限	ACL 简写	描述
create	c	可以创建子节点
delete	d	可以删除子节点（即下一级节点）
read	r	可以读取节点数据及显示子节点列表
write	w	可以设置节点数据
admin	a	可以设置节点访问控制列表权限

增、删、查、改、管理权限可以简写为 cdrwa
注意：这5中权限中，delete 是指对子节点的删除权限，其他四种权限指对自身节点的操作权限

6.授权的相关命令

命令	使用方式	描述
getAcl	getAcl	读取ACL权限
setAcl	setAcl	设置ACL权限
addauth	addauth	添加认证用户

7.相关命令

world 授权模式

setAcl <path> world:anyone:<acl>

ip 授权模式

setAcl  <path> ip:<ip>:<acl>

Auth 授权模式

addauth digest <user>:<password> # 添加认证用户

setAcl <path> auth:<user>:<acl> # 授权

Digest 授权模式

setAcl <path> digest:<user>:<password>:<acl>

上文中的密码是经过 SHA1 即 base64 处理的密文，在 SHELL 中可以通过以下命令计算：

echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64

5.ZooKeeper JavaAPI操作

5.1.Curator介绍

Curator 是 Apache ZooKeeper 的 Java 客户端库。
常见的 ZooKeeper Java API ：
- 原生 Java API
- ZkClient
- Curator
Curator 项目的目标是简化 ZooKeeper 客户端的使用。
Curator 最初是 Netfix 研发的,后来捐献了 Apache 基金会,目前是 Apache 的顶级项目。
官网：https://curator.apache.org/

5.2.使用IDEA创建项目（Curator）

建立如下的项目结构

在这里插入图片描述

pom.xml

<?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>com.itheima</groupId>
    <artifactId>curator-zk</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
            <scope>test</scope>
        </dependency>

        <!-- curator -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>

        <!-- 日志 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.21</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.21</version>
        </dependency>

    </dependencies>


    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    
</project>

log4j.properties

log4j.rootLogger=off,stdout

log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%d{yyyy-MM-dd HH/:mm/:ss}]%-5p %c(line/:%L) %x-%m%n

5.3.Curator API的CRUD操作

建立连接、添加节点、删除节点、修改节点、查询节点

5.3.1.建立连接

1.CuratorTest.java文件（第一种方式建立连接）

package com.itheima.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
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.util.List;

public class CuratorTest {
    private CuratorFramework client;

    /**
     * 建立连接
     */
    @Test
    public void testConnect() {
        /*
         * @param connectString       连接字符串。zk server 地址和端口 "192.168.2.212:2181,192.168.2.212:2181"
         * @param sessionTimeoutMs    会话超时时间 单位ms
         * @param connectionTimeoutMs 连接超时时间 单位ms
         * @param retryPolicy         重试策略
         */
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        //创建连接，构造客户端对象_第一种方式
        client = CuratorFrameworkFactory.newClient(
                "192.168.2.212:2181", 60 * 1000,
                15 * 1000, retryPolicy);

        //开启连接
        client.start();
    }
}

2.CuratorTest.java文件（第二种方式建立连接）

package com.itheima.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
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.util.List;

public class CuratorTest {
    private CuratorFramework client;

    @Test
    public void testConnect() {
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        //创建连接构造客户端对象_第二种方式：链式编程的方式
        client = CuratorFrameworkFactory.builder()
                .connectString("192.168.2.212:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("itheima")//指定名称空间后，系统默认之后所做的操作的根目录就是该目录;若该目录下无子节点，过段时间就会自动删除该目录
                .build();

        //开启连接
        client.start();
    }
}

5.3.2.创建节点

CuratorTest.java文件

package com.itheima.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.After;
import org.junit.Before;
import org.junit.Test;

public class CuratorTest {
    private CuratorFramework client;

    /**
     * 建立连接
     */
    @Before
    public void testConnect() {
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
        
        //创建连接构造客户端对象_第二种方式：链式编程的方式
        client = CuratorFrameworkFactory.builder()
                .connectString("192.168.2.212:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("itheima")
                .build();

        //开启连接
        client.start();
    }

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

    /* 2.创建节点 带有数据 */
    @Test
    public void testCreate_2() throws Exception {
        String path = client.create().forPath("/app2", "hehe".getBytes());
    }

    /* 3.设置节点的类型 */
    @Test
    public void testCreate_3() throws Exception {
        //默认类型：持久化
        /**
         * PERSITENT 持久化节点
         * EPHEMERAL 临时节点：-e
         * PERSISTENT_SEQUENTIAL ：持久化顺序节点: -s
         * EPHEMERAL_SEQUENTIAL: 临时顺序节点:-es
         */
        String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
    }

    /* 4.创建多级节点 /app1/p1 */
    @Test
    public void testCreate_4() throws Exception {
        //creatingParentsIfNeeded()：如果父节点不存在，则创建父节点
        String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");
    }

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

5.3.3.查询节点

CuratorTest.java文件中添加的部分

/**
 * 查询节点：
 * 1. 查询数据：get: getData().forPath()
 * 2. 查询子节点： ls: getChildren().forPath()
 * 3. 查询节点状态信息：ls -s:getData().storingStatIn(状态对象).forPath()
 */

@Test
public void testGet_1() throws Exception {
    //1.查询数据：get
    byte[] data = client.getData().forPath("/app1");
}

@Test
public void testGet_2() throws Exception {
    // 2.查询子节点： ls
    List<String> path = client.getChildren().forPath("/");
}

@Test
public void testGet_3() throws Exception {
    Stat status = new Stat();
    //3.查询节点状态信息：ls -s
    client.getData().storingStatIn(status).forPath("/app1");
}

5.3.4.修改节点

CuratorTest.java文件中添加的部分

/**
 * 修改数据
 * 1. 基本修改数据：setData().forPath()
 * 2. 根据版本修改: setData().withVersion().forPath()
 * * version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。
 *
 * @throws Exception
 */
@Test
public void testSet() throws Exception {
    client.setData().forPath("/app1", "itcast".getBytes());
}

@Test
public void testSetForVersion() throws Exception {
    Stat status = new Stat();
    //查询节点状态信息：ls -s
    client.getData().storingStatIn(status).forPath("/app1");

    int version = status.getVersion();//查询出来的版本信息
    
    client.setData().withVersion(version).forPath("/app1", "heheda".getBytes());
}

5.3.5.删除节点

CuratorTest.java文件中添加的部分

/**
 * 删除节点： delete deleteall
 * 1. 删除单个节点:delete().forPath("/app1");
 * 2. 删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");
 * 3. 必须成功的删除:为了防止网络抖动。本质就是重试。  client.delete().guaranteed().forPath("/app2");
 * 4. 回调：inBackground
 *
 * @throws Exception
 */

@Test
public void testDelete_1() throws Exception {
    // 1.删除单个节点
    client.delete().forPath("/app1");
}

@Test
public void testDelete_2() throws Exception {
    //2.删除带有子节点的节点
    client.delete().deletingChildrenIfNeeded().forPath("/app4");
}

@Test
public void testDelete_3() throws Exception {
    //3.必须成功的删除
    client.delete().guaranteed().forPath("/app2");
}

@Test
public void testDelete4() throws Exception {
    //4.回调
    client.delete().guaranteed().inBackground(new BackgroundCallback() {
        @Override
        public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
            System.out.println(event);
        }
    }).forPath("/app1");
}

5.4.Watch事件监听

5.4.1.Watch监听的概念

Curator API 常用操作 - Watch 事件监听

ZooKeeper 允许用户在指定节点上注册一些 Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。
ZooKeeper 中引入了 Watcher 机制来实现了发布/订阅功能能,能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。
ZooKeeper 原生支持通过注册 Watcher 来进行事件监听，但是其使用并不是特别方便需要开发人员自己反复注册 Watcher，比较繁琐。
Curator 引入了 Cache 来实现对 ZooKeeper 服务端事件的监听。
ZooKeeper 提供了三种 Watcher：
- NodeCache：只是监听某一个特定的节点
- PathChildrenCache：监控—个 ZNode 的子节点.
- TreeCache：可以监控整个树上的所有节点，类似于 PathChildrenCache 和 NodeCache 的组合

5.4.2.NodeCache

CuratorWatcherTest.java

package com.itheima.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class CuratorWatcherTest {
    private CuratorFramework client;

    /**
     * 建立连接
     */
    @Before
    public void testConnect() {
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        client = CuratorFrameworkFactory.builder()
                .connectString("192.168.2.212:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .namespace("itheima")
                .build();

        //开启连接
        client.start();
    }

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

    /**
     * 演示 NodeCache：给指定一个节点注册监听器
     */

    @Test
    public void testNodeCache() throws Exception {
        //1.创建 NodeCache 对象
        final NodeCache nodeCache = new NodeCache(client, "/app1");
        //2.注册监听
        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));
            }
        });

        //3.开启监听.如果设置为 true，则开启监听 是，加载缓冲数据
        nodeCache.start(true);

        while (true) {

        }
    }
}

5.4.3.PathChildrenCache

CuratorWatcherTest.java

/**
 * 演示 PathChildrenCache：监听某个节点的所有子节点们
 */

@Test
public void testPathChildrenCache() throws Exception {
    //1.创建监听对象
    PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/app2", true);

    //2. 绑定监听器
    pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            System.out.println("子节点变化了~");
            System.out.println(event);
            //监听子节点的数据变更，并且拿到变更后的数据
            //1.获取类型
            PathChildrenCacheEvent.Type type = event.getType();
            //2.判断类型是否是update
            if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                System.out.println("数据变了！！！");
                byte[] data = event.getData().getData();
                System.out.println(new String(data));
            }
        }
    });
    //3. 开启
    pathChildrenCache.start();

    while (true) {

    }
}

5.4.4.TreeCache

CuratorWatcherTest.java

/**
 * 演示 TreeCache：监听某个节点自己和所有子节点们
 */

@Test
public void testTreeCache() throws Exception {
    //1. 创建监听器
    TreeCache treeCache = new TreeCache(client, "/app2");

    //2. 注册监听
    treeCache.getListenable().addListener(new TreeCacheListener() {
        @Override
        public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
            System.out.println("节点变化了");
            System.out.println(event);
        }
    });

    //3. 开启
    treeCache.start();

    while (true) {

    }
}

5.5.分布式锁

分布式锁

在我们进行单机应用开发，涉及并发同步的时候，我们往往采用 synchronized 或者 Lock 的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个 JVM 之下，没有任何问题。
但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨 JVM 之间已经无法通过多线程的锁解决同步问题。
那么就需要一种更加高级的锁机制，来处理种跨机器的进程之间的数据同步问题——这就是分布式锁。

在这里插入图片描述

分布式锁原理

核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。
客户端获取锁时，在 lock 节点下创建临时顺序节点。
然后获取 lock 下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。
如果发现自己创建的节点并非 lock 所有子节点中最小的，说明自己还没有获取到锁。此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。
如果发现比自己小的那个节点被删除，则客户端的 Watcher 会收到相应通知，此时再次判断自己创建的节点：
- 是否是 lock 子节点中序号最小的，如果是则获取到了锁；
- 如果不是，则重复以上步骤，继续获取到比自己小的一个节点，并注册监听。

在这里插入图片描述

5.6.案例：模拟12306售票

Curator 实现分布式锁 API

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

在这里插入图片描述

接下来以线程的方式来模拟卖票过程

Ticket12306.java

package com.itheima.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 Ticket12306 implements Runnable{
    private int tickets = 10;//数据库的票数

    private InterProcessMutex lock ;

    public Ticket12306(){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);

        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.2.212:2181")
                .sessionTimeoutMs(60 * 1000)
                .connectionTimeoutMs(15 * 1000)
                .retryPolicy(retryPolicy)
                .build();

        //开启连接
        client.start();

        lock = new InterProcessMutex(client,"/lock");
    }

    @Override
    public void run() {
        while(true){
            //获取锁
            try {
                lock.acquire(3, TimeUnit.SECONDS);
                if(tickets > 0){
                    System.out.println(Thread.currentThread()+":"+tickets);
                    Thread.sleep(100);
                    tickets--;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                //释放锁
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

LockTest.java

package com.itheima.curator;

public class LockTest {
    public static void main(String[] args) {
        Ticket12306 ticket12306 = new Ticket12306();

        //创建客户端
        Thread t1 = new Thread(ticket12306,"携程");
        Thread t2 = new Thread(ticket12306,"飞猪");

        t1.start();
        t2.start();
    }

}

6.ZooKeeper集群搭建

6.1.Leader选举

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

在这里插入图片描述

6.2.搭建要求

真实的集群是需要部署在不同的服务器上的，但是在我们测试时同时启动很多个虚拟机内存会吃不消。

所以我们通常会搭建 伪集群，也就是把所有的服务都搭建在一台虚拟机上，用端口进行区分。

我们这里要求搭建一个三个节点的 Zookeeper 集群（伪集群）。

6.3.准备工作

重新部署一台虚拟机作为我们搭建集群的测试服务器。

1.安装 JDK，详情请见：《CentOS7卸载openJDK，安装jdk1.8》
2.Zookeeper 压缩包上传到服务器
3.将 Zookeeper 解压，建立 /usr/local/zookeeper-cluster 目录，将解压后的 Zookeeper 复制到三个目录

[root@localhost ~]# mkdir /usr/local/zookeeper-cluster

[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-1

[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-2

[root@localhost ~]# cp -r  apache-zookeeper-3.5.6-bin /usr/local/zookeeper-cluster/zookeeper-3

4.创建 data 目录，并且将 conf 下 zoo_sample.cfg 文件改名为 zoo.cfg

mkdir /usr/local/zookeeper-cluster/zookeeper-1/data

mkdir /usr/local/zookeeper-cluster/zookeeper-2/data

mkdir /usr/local/zookeeper-cluster/zookeeper-3/data

mv /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

mv /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

mv /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo_sample.cfg  /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

5.配置每一个 Zookeeper 的 dataDir、dataLogDir 和 clientPort

修改/usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-1/conf/zoo.cfg

clientPort=2181
dataDir=/usr/local/zookeeper-cluster/zookeeper-1/data
dataLogDir=/usr/local/zookeeper-cluster/zookeeper-1/datalog

修改/usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-2/conf/zoo.cfg

clientPort=2182
dataDir=/usr/local/zookeeper-cluster/zookeeper-2/data
dataLogDir=/usr/local/zookeeper-cluster/zookeeper-2/datalog

修改/usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

vim /usr/local/zookeeper-cluster/zookeeper-3/conf/zoo.cfg

clientPort=2183
dataDir=/usr/local/zookeeper-cluster/zookeeper-3/data
dataLogDir=/usr/local/zookeeper-cluster/zookeeper-3/datalog

6.4.配置集群

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

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

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

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

在每一个 zookeeper 的 zoo.cfg 配置客户端访问端口（clientPort）和集群服务器 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

集群服务器IP列表如下

server.1=192.168.2.212:2881:3881
server.2=192.168.2.212:2882:3882
server.3=192.168.2.212:2883:3883

解释：server.服务器ID = 服务器IP地址 : 服务器之间通信端口 : 服务器之间投票选举端口

6.5.启动集群

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

/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

启动后我们查询一下每个实例的运行状态

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

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

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

在这里插入图片描述

Mode 为 follower 表示是 跟随者（从）

Mode 为 leader 表示是 领导者（主）

6.6.模拟集群异常

1.首先我们先测试：如果是从服务器挂掉，会怎么样？把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个节点的集群，从服务器挂掉，集群正常

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

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

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

在这里插入图片描述

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

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

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

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

在这里插入图片描述

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

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

/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 产生了。

由此我们得出结论：当集群中的主服务器挂了，集群中的其他服务器会自动进行选举状态，然后产生新得 leader

5.我们再次测试，当我们把2号服务器重新启动起来启动后，会发生什么？2号服务器会再次成为新的领导吗？我们看结果

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

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

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

在这里插入图片描述

我们会发现，2号服务器启动后依然是跟随者（从服务器），3号服务器依然是领导者（主服务器），没有撼动3号服务器的领导地位。

由此我们得出结论：当领导者产生后，再次有新服务器加入集群，不会影响到现任领导者。

6.7.ZooKeeper集群角色

在ZooKeeper集群服中务中有三个角色
- Leader 领导者：
  - 处理事务请求
  - 集群内部各服务器的调度者
- Follower 跟随者：
  - 处理客户端非事务请求，转发事务请求给 Leader 服务器
  - 参与 Leader 选举投票
- Observer 观察者：
  - 处理客户端非事务请求，转发事务请求给 Leader 服务器

在这里插入图片描述

向您推荐>>Eolink开发者社区

权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区

更多推荐

深入理解 Mocha 测试框架：从零实现一个 Mocha

前言什么是自动化测试自动化测试在很多团队中都是Devops环节中很难执行起来的一个环节，主要原因在于测试代码的编写工作很难抽象，99%的场景都需要和业务强绑定，而且写测试代码的编写工作量往往比编写实际业务代码的工作量更多。在一些很多业务场景中投入产出比很低，适合写自动化测试的应该是那些中长期业务以及一些诸如组件一样的基础库。自动化测试是个比较大的概念，其中分类也比较多，比如单元测试，端对端测试，集

云原生

ELK实现containerd的容器日志采集展示【基于logging的全栈监测】

企业级ELK Stack构建介绍

云原生

(20200916 Solved)docker-compose up创建容器自动退出

问题描述如题，创建容器后自动退出了。并且docker start container无效解决方案原因是缺失了控制终端的配置，需要在docker-compose.yml中增加tty:true ，有时候这样也不行，需要再增加一个command:/bin/bash，命令不一定是这个，需要是一个不会退出的命令，然后用-d后台启动容器。Referencesdocker-compose启动容器后自动退出...