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

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,118元/月
云原生网关 MSE Higress,422元/月
日志服务 SLS,月写入数据量 50GB 1个月
简介: 1. ZooKeeper官方1). 简介ZooKeeper是一个分布式的,开放源码的分布式应用程序协调服务,是Google的Chubby一个开源的实现,是Hadoop和Hbase的重要组件。

1. ZooKeeper官方

1). 简介

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

2). ZooKeeper下载
img_70cdaa5fc3ecb19a227dda8330675987.png
图1.png
3). 分布式应用
img_a709f133892efde6f4766031c3638e4c.png
图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。


    img_19e01b83c25d0d2145d99d0fb399f179.png
    图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.


img_5232f0c44202c690a635fce8876126bb.png
图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
img_ad9af82ef59758b84f19418fcb0fed62.png
图5.png
  • 查看进程
jps -l -v
img_ec12da74970c031c9fd955f47b956143.png
图6.png
  • 客户端连接
zkCli.cmd 127.0.0.1:2181
img_d2fb34e3b3e054921dd0eef3be899c99.png
图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目录下


img_aa0e5a7c669951f8d210deb19273fe4a.png
图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. 测试

  • 执行结果:


    img_217e085cb70430fe6207d473ee98720e.png
    图9.png
  • 获取数据:在命令提示符中使用zkCli.cmd开启一个客户端窗口,
    img_8648da2f23e0979d233d483f1f6bad1b.png
    图10.png

    连接成功后, 输入get /MyFirst(get 路径名)查看数据
    img_53f44df2f0bf8e8d1a87020ee94d7784.png
    图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. 测试


img_ab6f0b3e8b2c8252af3a288e6e1e3d8a.png
图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. 测试


img_9ec077c06a51a21738051baf99bc8299.png
图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 路径名)查看数据

img_c334fb281dd9ddefbc149a2db2520585.png
图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. 测试

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


    img_1348de2bf16e4c4add50e02946031b61.png
    图14.png
  • 运行项目


    img_d125e7bd957185cd67d3f2388027c175.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


img_0468bc7bc46e504281b210e91ee9047d.png
图16.png

代码下载

相关实践学习
基于MSE实现微服务的全链路灰度
通过本场景的实验操作,您将了解并实现在线业务的微服务全链路灰度能力。
目录
相关文章
|
9天前
|
安全 应用服务中间件 API
微服务分布式系统架构之zookeeper与dubbo-2
微服务分布式系统架构之zookeeper与dubbo-2
|
9天前
|
负载均衡 Java 应用服务中间件
微服务分布式系统架构之zookeeper与dubbor-1
微服务分布式系统架构之zookeeper与dubbor-1
|
10天前
|
存储 负载均衡 Dubbo
分布式-Zookeeper(一)
分布式-Zookeeper(一)
|
7天前
|
数据采集 分布式计算 MaxCompute
MaxCompute 分布式计算框架 MaxFrame 服务正式商业化公告
MaxCompute 分布式计算框架 MaxFrame 服务于北京时间2024年09月27日正式商业化!
27 3
|
10天前
分布式-Zookeeper-数据订阅
分布式-Zookeeper-数据订阅
|
10天前
|
监控
分布式-Zookeeper-Zab协议
分布式-Zookeeper-Zab协议
|
10天前
|
Java
分布式-Zookeeper-分布式锁
分布式-Zookeeper-分布式锁
|
10天前
|
存储 负载均衡 算法
分布式-Zookeeper-Master选举
分布式-Zookeeper-Master选举
|
10天前
|
NoSQL 容灾 关系型数据库
分布式协调服务-Zookeeper
分布式协调服务-Zookeeper
|
10天前
|
存储 NoSQL Redis
分布式-Zookeeper(二)
分布式-Zookeeper(二)
下一篇
无影云桌面