分布式--ZooKeeper 分布式服务协同-阿里云开发者社区

1. ZooKeeper官方

1). 简介

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。
ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。ZooKeeper包含一个简单的原语集，提供Java和C的接口。
ZooKeeper代码版本中，提供了分布式独享锁、选举、队列的接口，代码在zookeeper-3.4.8\src\recipes。其中分布锁和队列有Java和C两个版本，选举只有Java版本。

2). ZooKeeper下载

图1.png

3). 分布式应用

图2.png

4). 优点及挑战

分布式应用的优点
- 可靠性 - 单个或几个系统的故障不会使整个系统出现故障。
- 可扩展性 - 可以在需要时增加性能，通过添加更多机器，在应用程序配置中进行微小的更改，而不会有停机时间。
- 透明性 - 隐藏系统的复杂性，并将其显示为单个实体/应用程序。
分布式应用的挑战
- 竞争条件 - 两个或多个机器尝试执行特定任务，实际上只需在任意给定时间由单个机器完成。例如，共享资源只能在任意给定时间由单个机器修改。
- 死锁 - 两个或多个操作等待彼此无限期完成。
- 不一致 - 数据的部分失败。

5). ZooKeeper描述

Apache ZooKeeper是由集群（节点组）使用的一种服务，用于在自身之间协调，并通过稳健的同步技术维护共享数据。ZooKeeper本身是一个分布式应用程序，为写入分布式应用程序提供服务。
ZooKeeper提供的常见服务如下 :

命名服务 - 按名称标识集群中的节点。它类似于DNS，但仅对于节点。
配置管理 - 加入节点的最近的和最新的系统配置信息。
集群管理 - 实时地在集群和节点状态中加入/离开节点。
选举算法 - 选举一个节点作为协调目的的leader。
锁定和同步服务 - 在修改数据的同时锁定数据。此机制可帮助你在连接其他分布式应用程序（如Apache HBase）时进行自动故障恢复。
高度可靠的数据注册表 - 即使在一个或几个节点关闭时也可以获得数据。

6). 工作流

一旦ZooKeeper集合启动，它将等待客户端连接。客户端将连接到ZooKeeper集合中的一个节点。它可以是leader或follower节点。一旦客户端被连接，节点将向特定客户端分配会话ID并向该客户端发送确认。如果客户端没有收到确认，它将尝试连接ZooKeeper集合中的另一个节点。一旦连接到节点，客户端将以有规律的间隔向节点发送心跳，以确保连接不会丢失。

如果客户端想要读取特定的znode，它将会向具有znode路径的节点发送读取请求，并且节点通过从其自己的数据库获取来返回所请求的znode。为此，在ZooKeeper集合中读取速度很快。
如果客户端想要将数据存储在ZooKeeper集合中，则会将znode路径和数据发送到服务器。连接的服务器将该请求转发给leader，然后leader将向所有的follower重新发出写入请求。如果只有大部分节点成功响应，而写入请求成功，则成功返回代码将被发送到客户端。否则，写入请求失败。绝大多数节点被称为 Quorum 。

7). 节点

让我们分析在ZooKeeper集合中拥有不同数量的节点的效果。

如果我们有单个节点，则当该节点故障时，ZooKeeper集合将故障。它有助于“单点故障"，不建议在生产环境中使用。
如果我们有两个节点而一个节点故障，我们没有占多数，因为两个中的一个不是多数。
如果我们有三个节点而一个节点故障，那么我们有大多数，因此，这是最低要求。ZooKeeper集合在实际生产环境中必须至少有三个节点。
如果我们有四个节点而两个节点故障，它将再次故障。类似于有三个节点，额外节点不用于任何目的，因此，最好添加奇数的节点，例如3，5，7。

图3.png

组件	描述
写入（write）	写入过程由leader节点处理。leader将写入请求转发到所有znode，并等待znode的回复。如果一半的znode回复，则写入过程完成。
读取（read）	读取由特定连接的znode在内部执行，因此不需要与集群进行交互。
复制数据库（replicated database）	它用于在zookeeper中存储数据。每个znode都有自己的数据库，每个znode在一致性的帮助下每次都有相同的数据。
Leader	Leader是负责处理写入请求的Znode。
Follower	follower从客户端接收写入请求，并将它们转发到leader znode。
请求处理器（request processor）	只存在于leader节点。它管理来自follower节点的写入请求。
原子广播（atomic broadcasts）	负责广播从leader节点到follower节点的变化。

2. 安装

1). 修改/conf/zoo_sample.cfg配置文件

I. 修改文件名zoo_sample.cfg为zoo.cfg.

图4.png

II. 修改zoo.cfg文件内容

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=D:\\zookeeper-3.4.10.tar\\zookeeper-3.4.10\\data
dataLogDir=D:\\zookeeper-3.4.10.tar\\zookeeper-3.4.10\\log
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

III. 配置文件解析

tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。
dataDir：顾名思义就是 Zookeeper 保存数据的目录，默认情况下，Zookeeper 将写数据的日志文件也保存在这个目录里。
dataLogDir：顾名思义就是 Zookeeper 保存日志文件的目录
clientPort：这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。

2). 启动

进入到bin目录，并且启动zkServer.cmd，这时会启动一个java进程

zkServer.cmd

图5.png

查看进程

jps -l -v

图6.png

客户端连接

zkCli.cmd 127.0.0.1:2181

图7.png

3. API代码测试

1). 基础知识

与ZooKeeper集合进行交互的应用程序称为 ZooKeeper客户端或简称客户端。
Znode是ZooKeeper集合的核心组件，ZooKeeper API提供了一小组方法使用ZooKeeper集合来操纵znode的所有细节。
客户端应该遵循以步骤，与ZooKeeper集合进行清晰和干净的交互。

连接到ZooKeeper集合。ZooKeeper集合为客户端分配会话ID。
定期向服务器发送心跳。否则，ZooKeeper集合将过期会话ID，客户端需要重新连接。
只要会话ID处于活动状态，就可以获取/设置znode。
所有任务完成后，断开与ZooKeeper集合的连接。如果客户端长时间不活动，则ZooKeeper集合将自动断开客户端。

2). API

ZooKeeper API的核心部分是ZooKeeper类。它提供了在其构造函数中连接ZooKeeper集合的选项，并具有以下方法：

connect - 连接到ZooKeeper集合
create- 创建znode
exists- 检查znode是否存在及其信息
getData - 从特定的znode获取数据
setData - 在特定的znode中设置数据
getChildren - 获取特定znode中的所有子节点
delete - 删除特定的znode及其所有子项
close - 关闭连接

新建Java工程，并导入/lib/log4j-1.2.16.jar、/lib/slf4j-api-1.6.1.jar、/lib/slf4j-log4j12-1.6.1.jar、zookeeper-3.4.10.jar，将/conf/路径下的log4j.properties文件拷贝到项目的src目录下

图8.png

3). connect(连接到ZooKeeper集合)

I. 方法原型

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

connectionString - ZooKeeper集合主机。
sessionTimeout - 会话超时（以毫秒为单位）。
watcher - 实现“监视器”界面的对象。ZooKeeper集合通过监视器对象返回连接状态。
II. 代码

/**
 * ZooKeeper连接管理
 * @author mazaiting
 */
public class ZooKeeperConnection {

    /**
     * 定义ZooKeeper实例
     */
    private ZooKeeper zooKeeper;
    /**
     * 用户停止(等待)主线程，直到客户端与ZooKeeper集合连接
     */
    private final CountDownLatch connectedSignal = new CountDownLatch(1);
    
    /**
     * 连接
     * @param host 主机地址
     * @return ZooKeeper对象
     * @throws IOException IO异常
     * @throws InterruptedException 中断异常
     */
    public ZooKeeper connect(String host) throws IOException, InterruptedException {
        // 创建ZooKeeper对象
        zooKeeper = new ZooKeeper(host, 5000, new Watcher(){
            /**
             * 
             */
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        });
        // 等待
        connectedSignal.await();
        return zooKeeper;
    }
    
    /**
     * 关机ZooKeeper的连接
     * @throws InterruptedException 中断异常
     */
    public void close() throws InterruptedException {
        zooKeeper.close();
    }
}

4). create(创建znode)

I. 方法原型

create(String path, byte[] data, List<ACL> acl, CreateMode createMode)

path - Znode路径。例如，/myapp1，/myapp2，/myapp1/mydata1，myapp2/mydata1/myanothersubdata
data - 要存储在指定znode路径中的数据
acl - 要创建的节点的访问控制列表。ZooKeeper API提供了一个静态接口 ZooDefs.Ids 来获取一些基本的acl列表。例如，ZooDefs.Ids.OPEN_ACL_UNSAFE返回打开znode的acl列表。
createMode - 节点的类型，即临时，顺序或两者。这是一个枚举。

II. 代码

/**
 * 创建节点
 * @author mazaiting
 */
public class ZKCreate {
    /**
     * 创建静态的ZooKeeper实例
     */
    private static ZooKeeper zooKeeper;
    /**
     * 创建静态的连接实例
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 创建节点
     * @param path Znode路径
     * @param data 存储在Znode路径中的数据
     * @throws KeeperException ZooKeeper异常
     * @throws InterruptedException 中断异常
     */
    public static void create(String path, byte[] data) throws KeeperException, InterruptedException {
        zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
    
    public static void main(String[] args) {
        // znode路径
        String path = "/MyFirst";
        // 二进制数据
        byte[] data = "My First ZooKeeper app".getBytes();
        
        try {
            // 创建连接对象
            conn = new ZooKeeperConnection();
            // 连接， 并返回ZooKeeper
            zooKeeper = conn.connect("127.0.0.1");
            // 创建Znode
            create(path, data);
            // 关闭
            conn.close();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
    
}

III. 测试

执行结果：

图9.png
获取数据：在命令提示符中使用zkCli.cmd开启一个客户端窗口，

图10.png

连接成功后, 输入get /MyFirst(get 路径名)查看数据

图10.png

5). exists(检查znode是否存在及其信息)

I. 方法原型

exists(String path, boolean watcher)

path- Znode路径
watcher - 布尔值，用于指定是否监视指定的znode
II. 代码

/**
 * 判断某个路径是否存在
 * @author mazaiting
 */
public class ZKExists {
    /**
     * 定义静态的ZooKeeper对象
     */
    private static ZooKeeper zooKeeper;
    /**
     * 定义静态的ZooKeeper连接对象
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 判断某个路径是否存在
     * @param path 路径
     * @return 返回状态
     * @throws KeeperException ZooKeeper异常
     * @throws InterruptedException 中断异常
     */
    public static Stat exists(String path) throws KeeperException, InterruptedException {
        return zooKeeper.exists(path, true);
    }
    
    public static void main(String[] args) {
        // znode路径
        String path = "/MyFirst";       
        try {
            // 创建连接对象
            conn = new ZooKeeperConnection();
            // 连接， 并返回ZooKeeper
            zooKeeper = conn.connect("127.0.0.1");
            // 检测路径是否存在
            Stat stat = exists(path);
            // 判断是否为空
            if (null != stat) {
                // 存在打印其版本信息
                System.out.println("Node exists and the node version is " + stat.getVersion());
            } else {
                System.out.println("Node does not exists");
            }
            // 关闭
            conn.close();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}

III. 测试

图11.png

6). getData(从特定的znode获取数据)

I. 方法原型

getData(String path, Watcher watcher, Stat stat)

path - Znode路径。
watcher - 监视器类型的回调函数。当指定的znode的数据改变时，ZooKeeper集合将通过监视器回调进行通知。这是一次性通知。
stat - 返回znode的元数据。

II. 代码

/**
 * 获取数据
 * @author mazaiting
 */
public class ZKGetData {
    /**
     * 定义静态的ZooKeeper对象
     */
    private static ZooKeeper zooKeeper;
    /**
     * 定义静态的ZooKeeper连接对象
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 判断某个路径是否存在
     * @param path 路径
     * @return 返回状态
     * @throws KeeperException ZooKeeper异常
     * @throws InterruptedException 中断异常
     */
    public static Stat exists(String path) throws KeeperException, InterruptedException {
        return zooKeeper.exists(path, true);
    }
    
    public static void main(String[] args) {
        // 路径
        String path = "/MyFirst";
        // 用户停止(等待)主线程，直到客户端知道获取到数据
        final CountDownLatch connectedSignal = new CountDownLatch(1);
        try {
            // 创建连接对象
            conn = new ZooKeeperConnection();
            // 连接
            zooKeeper = conn.connect("127.0.0.1");
            // 检测是否存此路径
            Stat stat = exists(path);
            // 判断是否存在
            if (null != stat) {
                // 获取字节数组
                byte[] bytes = zooKeeper.getData(path, new Watcher() {
                    
                    @Override
                    public void process(WatchedEvent event) {
                        // 检测状态
                        if (Event.EventType.None == event.getType()) {
                            // 判断状态
                            switch (event.getState()) {
                            case Expired:
                                // 关闭
                                connectedSignal.countDown();
                                break;

                            default:
                                break;
                            }
                        } else {
                            try {
                                byte[] bn = zooKeeper.getData(path, false, null);
                                // 将字节数组转换为字符创
                                String data = new String(bn, "UTF-8");
                                // 打印字符串
                                System.out.println(data);
                                // 关闭
                                connectedSignal.countDown();
                            } catch (Exception e) {
                                // 打印错误日志
                                System.out.println(e.getMessage());
                            }
                        }
                    }
                }, null);
                // 将字节数组转换为字符创
                String data = new String(bytes, "UTF-8");
                // 打印字符串
                System.out.println(data);
                // 等待
                connectedSignal.await();
            } else {
                // 节点不存在
                System.out.println("Node does not exists");
            }
        } catch (Exception e) {
            // 打印错误日志
            System.out.println(e.getMessage());
        }
    }
}

III. 测试

图12.png

7). setData(在特定的znode中设置数据)

I. 方法原型

setData(String path, byte[] data, int version)

path- Znode路径。
data - 要存储在指定znode路径中的数据。
version- znode的当前版本。每当数据更改时，ZooKeeper会更新znode的版本号。
II. 代码

/**
 * 设置数据
 * @author mazaiting
 */
public class ZKSetData {

    /**
     * 定义静态的ZooKeeper对象
     */
    private static ZooKeeper zooKeeper;
    /**
     * 定义静态的ZooKeeper连接对象
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 更新数据
     * @param path 路径
     * @param data 二进制数据
     * @throws KeeperException ZooKeeper异常
     * @throws InterruptedException 中断异常
     */
    public static void update(String path, byte[] data) throws KeeperException, InterruptedException {
        zooKeeper.setData(path, data, zooKeeper.exists(path, true).getVersion());
    }
    
    public static void main(String[] args) {
        // 路径
        String path = "/MyFirst";
        // 二进制数据
        byte[] data = "Success".getBytes();
        try {
            // 创建连接对象
            conn = new ZooKeeperConnection();
            // 连接并获取ZooKeeper
            zooKeeper = conn.connect("127.0.0.1");
            // 更新数据
            update(path, data);
        } catch (Exception e) {
            // 打印错误信息
            System.out.println(e.getMessage());
        }
    }
}

III. 测试
运行程序后，在命令提示符中使用zkCli.cmd开启一个客户端窗口，连接成功后, 输入get /MyFirst(get 路径名)查看数据

图13.png

8). getChildren(获取特定znode中的所有子节点)

I. 方法原型

getChildren(String path, Watcher watcher)

path - Znode路径。
watcher - 监视器类型的回调函数。当指定的znode被删除或znode下的子节点被创建/删除时，ZooKeeper集合将进行通知。这是一次性通知。
II. 代码

/**
 * 获取子节点
 * @author mazaiting
 */
public class ZKGetChildren {

    /**
     * 定义静态的ZooKeeper对象
     */
    private static ZooKeeper zooKeeper;
    /**
     * 定义静态的ZooKeeper连接对象
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 判断某个路径是否存在
     * @param path 路径
     * @return 返回状态
     * @throws KeeperException ZooKeeper异常
     * @throws InterruptedException 中断异常
     */
    public static Stat exists(String path) throws KeeperException, InterruptedException {
        return zooKeeper.exists(path, true);
    }
    
    public static void main(String[] args) {
        // 路径
        String path = "/MyFirst";
        try {
            // 创建连接对象
            conn = new ZooKeeperConnection();
            // 连接，并获取ZooKeeper对象
            zooKeeper = conn.connect("127.0.0.1");
            // 判断路径是否存在
            Stat stat = exists(path);
            // 判断是否为空
            if (null != stat) {
                // 获取孩子列表
                List<String> children = zooKeeper.getChildren(path, false);
                // 遍历孩子列表
                for(int i = 0; i < children.size(); i++) {
                    // 打印数据
                    System.out.println(children.get(i));
                }
            } else {
                // 路径不存在
                System.out.println("Node does not exists");
            }
        } catch (Exception e) {
            // 打印错误信息
            System.out.println(e.getMessage());
        }
    }
}

III. 测试

先在命令提示符窗口中添加两个孩子节点

图14.png
运行项目

图15.png

9). delete(删除特定的znode及其所有子项)

I. 方法原型

delete(String path, int version)

path - Znode路径。
version - znode的当前版本。
II. 代码

/**
 * 删除路径
 * @author mazaiting
 */
public class ZKDelete {

    /**
     * 定义静态的ZooKeeper对象
     */
    private static ZooKeeper zooKeeper;
    /**
     * 定义静态的ZooKeeper连接对象
     */
    private static ZooKeeperConnection conn;
    
    /**
     * 删除节点
     * @param path 路径
     * @throws InterruptedException 中断异常
     * @throws KeeperException ZooKeeper异常
     */
    public static void delete(String path) throws InterruptedException, KeeperException {
        zooKeeper.delete(path, zooKeeper.exists(path, true).getVersion());
    }
    
    public static void main(String[] args) {
        // 路径
        String path = "/MyFirst/first";
        try {
            // 创建ZooKeeper连接
            conn = new ZooKeeperConnection();
            // 连接
            zooKeeper = conn.connect("127.0.0.1");
            // 删除
            delete(path);
        } catch (Exception e) {
            // 打印错误信息
            System.out.println(e.getMessage());
        }
    }
}

III. 测试
删除前/MyFirst的numChildren量比删除后多1

图16.png

分布式--ZooKeeper 分布式服务协同

1. ZooKeeper官方

1). 简介

2). ZooKeeper下载

3). 分布式应用

4). 优点及挑战

5). ZooKeeper描述

6). 工作流

7). 节点

2. 安装

1). 修改/conf/zoo_sample.cfg配置文件

2). 启动

3. API代码测试

1). 基础知识

2). API

3). connect(连接到ZooKeeper集合)

4). create(创建znode)

5). exists(检查znode是否存在及其信息)

6). getData(从特定的znode获取数据)

7). setData(在特定的znode中设置数据)

8). getChildren(获取特定znode中的所有子节点)

9). delete(删除特定的znode及其所有子项)

代码下载

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

分布式--ZooKeeper 分布式服务协同

1. ZooKeeper官方

1). 简介

2). ZooKeeper下载

3). 分布式应用

4). 优点及挑战

5). ZooKeeper描述

6). 工作流

7). 节点

2. 安装

1). 修改/conf/zoo_sample.cfg配置文件

2). 启动

3. API代码测试

1). 基础知识

2). API

3). connect(连接到ZooKeeper集合)

4). create(创建znode)

5). exists(检查znode是否存在及其信息)

6). getData(从特定的znode获取数据)

7). setData(在特定的znode中设置数据)

8). getChildren(获取特定znode中的所有子节点)

9). delete(删除特定的znode及其所有子项)

热门文章

最新文章

相关课程

相关电子书

相关实验场景