作者: wangkai 2021-07-29 07:48:36
大数据
分布式 想起很久以前在某个客户现场,微服务 B 突然无法调用到微服务 A,为了使服务尽快正常恢复,重启了微服务 B 。虽然故障原因找到了,但对于 Zookeeper 的理解还是不够深刻,于是重新学习了 Zookeeper 的核心设计,并记录于此文共勉。
创新互联是一家专注于成都网站设计、成都网站建设与策划设计,浮梁网站建设哪家好?创新互联做网站,专注于网站建设十年,网设计领域的专业建站公司;建站业务涵盖:浮梁等地区。浮梁做网站价格咨询:18980820575
本文转载自微信公众号「KK架构」,作者wangkai 。转载本文请联系KK架构公众号。
想起很久以前在某个客户现场,微服务 B 突然无法调用到微服务 A,为了使服务尽快正常恢复,重启了微服务 B 。
但客户不依不饶询问这个问题出现的原因,于是我还大老远从杭州飞到深圳,现场排查问题。
最后的结论是,zk 在某时刻出现主备切换,此时微服务 A(基于 dubbo)需要重新往 zk上注册,但是端口号变了。
但是微服务 B 本地有微服务 A rpc 接口的缓存,缓存里面还是旧的端口,所以调用不到。
解决方法就是,把微服务的 rpc 端口号改成固定的。
虽说原因找到了,但对于 Zookeeper 的理解还是不够深刻,于是重新学习了 Zookeeper 的核心设计,并记录于此文共勉。
(1)Zookeeper 是一个分布式协调服务,是为了解决多个节点状态不一致的问题,充当中间机构来调停。如果出现了不一致,则把这个不一致的情况写入到 Zookeeper 中,Zookeeper 会返回响应,响应成功,则表示帮你达成了一致。
比如,A、B、C 节点在集群启动时,需要推举出一个主节点,这个时候,A、B、C 只要同时往 Zookeeper 上注册临时节点,谁先注册成功,谁就是主节点。
(2)Zookeeper 虽然是一个集群,但是数据并不是分散存储在各个节点上的,而是每个节点都保存了集群所有的数据。
其中一个节点作为主节点,提供分布式事务的写服务,其他节点和这个节点同步数据,保持和主节点状态一致。
(3)Zookeeper 所有节点的数据状态通过 Zab 协议保持一致。当集群中没有 Leader 节点时,内部会执行选举,选举结束,Follower 和 Leader 执行状态同步;当有 Leader 节点时,Leader 通过 ZAB 协议主导分布式事务的执行,并且所有的事务都是串行执行的。
(4)Zookeeper 的节点个数是不能线性扩展的,节点越多,同步数据的压力越大,执行分布式事务性能越差。推荐3、5、7 这样的数目。
Zookeeper 并没有沿用 Master/Slave 概念,而是引入了 Leader,Follower,Observer 三种角色。
通过 Leader 选举算法来选定一台服务器充当 Leader 节点,Leader 服务器为客户端提供读、写服务。
Follower 节点可以参加选举,也可以接受客户端的读请求,但是接受到客户端的写请求时,会转发到 Leader 服务器去处理。
Observer 角色只能提供读服务,不能选举和被选举,所以它存在的意义是在不影响写性能的前提下,提升集群的读性能。
答案是否,CAP 只能同时满足其二。
Zookeeper 是有取舍的,它实现了 A 可用性、P 分区容错性、C 的写入一致性,牺牲的是 C的读一致性。
也就是说,Zookeeper 并不保证读取的一定是最新的数据。如果一定要最新,需要使用 sync 回调处理。
Zookeeper 的 ZNode 模型其实可以理解为类文件系统,如下图:
为什么是类文件系统呢?因为 ZNode 模型没有文件和文件夹的概念,每个节点既可以有子节点,也可以存储数据。
那么既然每个节点可以存储数据,是不是可以任意存储无限制的数据呢?答案是否定的。在 Zookeeper 中,限制了每个节点只能存储小于 1 M 的数据,实际应用中,最好不要超过 1kb。
原因有以下四点:
(1)按生命周期分类
按照声明周期,ZNode 可分为永久节点和临时节点。
很好理解,永久节点就是要显示的删除,否则会一直存在;临时节点,是和会话绑定的,会话创建的所有节点,会在会话断开连接时,全部被 Zookeeper 系统删除。
(2)按照是否带序列号分类
带序列号的话,比如在代码中创建 /a 节点,创建之后其实是 /a000000000000001,再创建的话,就是 /a000000000000002,依次递增
不带序号,就是创建什么就是什么
(3)所以,一共有四种 ZNode:
(4)注意的点
临时节点下面不能挂载子节点,只能作为其他节点的叶子节点。
ZNode 的数据模型其实很简单,只有这么多知识。下面用代码来巩固一下。
这里我们使用 curator 框架来做 demo。(当然,你可以选择使用 Zookeeper 官方自带的 Api)
引入 pom 坐标:
org.apache.curator curator-framework 4.2.0 org.apache.curator curator-recipes 4.2.0
代码:
- public class ZkTest {
- // 会话超时
- private final int SESSION_TIMEOUT = 30 * 1000;
- // 连接超时 、 有啥区别
- private static final int CONNECTION_TIMEOUT = 3 * 1000;
- private static final String CONNECT_ADDR = "localhost:2181";
- private CuratorFramework client = null;
- public static void main(String[] args) throws Exception {
- // 创建客户端
- RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
- CuratorFramework client = CuratorFrameworkFactory.builder()
- .connectString(CONNECT_ADDR)
- .connectionTimeoutMs(CONNECTION_TIMEOUT)
- .retryPolicy(retryPolicy)
- .build();
- client.start();
- System.out.println(ZooKeeper.States.CONNECTED);
- System.out.println(client.getState());
- // 创建节点 /test1
- client.create()
- .forPath("/test1", "curator data".getBytes(StandardCharsets.UTF_8));
- System.out.println(client.getChildren().forPath("/"));
- // 临时节点
- client.create().withMode(CreateMode.EPHEMERAL)
- .forPath("/secondPath", "hello world".getBytes(StandardCharsets.UTF_8));
- System.out.println(new String(client.getData().forPath("/secondPath")));
- client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL)
- .forPath("/abc", "hello".getBytes(StandardCharsets.UTF_8));
- // 递归创建
- client.create()
- .creatingParentContainersIfNeeded()
- .forPath("/secondPath1/sencond2/sencond3");
- Thread.sleep(10000);
- }
Watcher 监听机制是 Zookeeper 解决各种分布式不一致疑难杂症的独家法门,也是学习 Zookeeper 必学的知识点。
Zookeeper 提供了数据的发布与订阅的功能,多个订阅者可以同时监听某一个对象,当这个对象自身状态发生变化时(例如节点数据或者节点的子节点个数变化),Zookeeper 系统会通知这些订阅者。
对于发布和订阅这个概念的理解,我们可以用这个场景来理解:
比如前两天的台风,老板想发一个通知给员工:明天在家办公。
于是老板会在钉钉群上 Ding 一个消息,员工自己打开钉钉查看。
在这个场景中,老板是发布者,员工是订阅者,钉钉群就是 Zookeeper 系统。
老板并不一一给员工发消息,而是把消息发到群里,员工就可以感知到消息的变化。
订阅者 | 员工 | 客户端1 |
---|---|---|
系统 | 钉钉群 | Zookeeper系统 |
发布者 | 老板 | 客户端2 |
客户端首先将 Watcher 注册到服务器上,同时将 Watcher 对象保存在客户端的 Watcher 管理器中。当 Zookeeper 服务端监听到数据状态发生变化时,服务端会首先主动通知客户端,接着客户端的 Watcher 管理器会触发相关的 Watcher 来回调响应的逻辑,从而完成整体的发布/订阅流程。
监听器 Watcher 的定义:
- public interface Watcher {
- // WatchedEvent 对象中有下面三个属性,Zookeeper状态,事件类型,路径
- // final private KeeperState keeperState;
- // final private EventType eventType;
- // private String path;
- abstract public void process(WatchedEvent event);
- }
下面是监听的大致流程图:
稍稍解释一下:
1、Client1 和 Client2 都关心 /app2 节点的数据状态变化,于是注册一个对于 /app2 的监听器到 Zookeeper 上;
2、当 Client3 修改 /app2 的值后,Zookeeper 会主动通知 Client1 和 Client2 ,并且回调监听器的方法。
当然这里的数据状态变化有下面这些类型:
我们还是用 Curator 框架来验证一下这个监听器。
代码很简单,这里我们使用 TreeCache 表示对于 /app2 的监听,并且注册了监听的方法。
- public class CuratorWatcher {
- public static void main(String[] args) throws Exception {
- CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181")
- .connectionTimeoutMs(10000)
- .retryPolicy(new ExponentialBackoffRetry(1000, 10))
- .build();
- client.start();
- String path = "/app2";
- TreeCache treeCache = new TreeCache(client, path);
- treeCache.start();
- treeCache.getListenable().addListener((client1, event) -> {
- System.out.println("event.getData()," + event.getData());
- System.out.println("event.getType()," + event.getType());
- });
- Thread.sleep(Integer.MAX_VALUE);
- }
- }
当 /app2 的状态发生变化时,就会调用监听的方法。
Curator 是对原生的 Zookeeper Api 有封装的,原生的 Zookeeper 提供的 Api ,注册监听后,当数据发生改变时,监听就被服务端删除了,要重复注册监听。
Curator 则对这个做了相应的封装和改进。
这里我们主要想实现的功能是:
代码还是用 Curator 框架实现的:
- package com.kkarch.zookeeper;
- import cn.hutool.core.util.StrUtil;
- import lombok.extern.slf4j.Slf4j;
- import org.apache.curator.framework.CuratorFramework;
- import org.apache.curator.framework.recipes.cache.TreeCache;
- import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
- import org.apache.zookeeper.CreateMode;
- import java.nio.charset.StandardCharsets;
- /**
- * 分布式选举
- *
- * @Author wangkai
- * @Time 2021/7/25 20:12
- */
- @Slf4j
- public class ElectorTest {
- private static final String PARENT = "/cluster_ha";
- private static final String ACTIVE = PARENT + "/ActiveMaster";
- private static final String STANDBY = PARENT + "/StandbyMaster";
- private static final String LOCK = PARENT + "/ElectorLock";
- private static final String HOSTNAME = "bigdata05";
- private static final String activeMasterPath = ACTIVE + "/" + HOSTNAME;
- private static final String standByMasterPath = STANDBY + "/" + HOSTNAME;
- public static void main(String[] args) throws Exception {
- CuratorFramework zk = ZkUtil.createZkClient("localhost:2181");
- zk.start();
- // 注册好监听
- TreeCache treeCache = new TreeCache(zk, PARENT);
- treeCache.start();
- treeCache.getListenable().addListener((client, event) -> {
- if (event.getType().equals(TreeCacheEvent.Type.INITIALIZED) || event.getType().equals(TreeCacheEvent.Type.CONNECTION_LOST)
- || event.getType().equals(TreeCacheEvent.Type.CONNECTION_RECONNECTED) || event.getType().equals(TreeCacheEvent.Type.CONNECTION_SUSPENDED)) {
- return;
- }
- System.out.println(event.getData());
- // 如果 Active 下有节点被移除了,没有节点,则应该去竞选成为 Active
- if (StrUtil.startWith(event.getData().getPath(), ACTIVE) && event.getType().equals(TreeCacheEvent.Type.NODE_REMOVED)) {
- if (getChildrenNumber(zk, ACTIVE) == 0) {
- createZNode(client, LOCK, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL);
- System.out.println(HOSTNAME + "争抢到了锁");
- }
- }
- // 如果有锁节点被创建,则判断是不是自己创建的,如果是,则切换自己的状态为 ACTIVE
- else if (StrUtil.equals(event.getData().getPath(), LOCK) && event.getType().equals(TreeCacheEvent.Type.NODE_ADDED)) {
- if (StrUtil.equals(new String(event.getData().getData()), HOSTNAME)) {
- createZNode(zk, activeMasterPath, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL);
- if (checkExists(client, standByMasterPath)) {
- deleteZNode(client, standByMasterPath);
- }
- }
- }
- });
- // 先创建 ACTIVE 和 STANDBY 节点
- if (zk.checkExists().forPath(ACTIVE) == null) {
- zk.create().creatingParentContainersIfNeeded().forPath(ACTIVE);
- }
- if (zk.checkExists().forPath(STANDBY) == null) {
- zk.create().creatingParentContainersIfNeeded().forPath(STANDBY);
- }
- // 判断 ACTIVE 下是否有子节点,如果没有则去争抢一把锁
- if (getChildrenNumber(zk, ACTIVE) == 0) {
- createZNode(zk, LOCK, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL);
- }
- // 如果有,则自己成为 STANDBY 状态
- else {
- createZNode(zk, standByMasterPath, HOSTNAME.getBytes(StandardCharsets.UTF_8), CreateMode.EPHEMERAL);
- }
- Thread.sleep(1000000000);
- }
- public static int getChildrenNumber(CuratorFramework client, String path) throws Exception {
- return client.getChildren().forPath(path).size();
- }
- public static void createZNode(CuratorFramework client, String path, byte[] data, CreateMode mode) {
- try {
- client.create().withMode(mode).forPath(path, data);
- } catch (Exception e) {
- log.error("创建节点失败", e);
- System.out.println("创建节点失败了");
- }
- }
- public static boolean checkExists(CuratorFramework client, String path) throws Exception {
- return client.checkExists().forPath(path) != null;
- }
- public static void deleteZNode(CuratorFramework client, String path) {
- try {
- if (checkExists(client, path)) {
- client.delete().forPath(path);
- }
- } catch (Exception e) {
- log.error("删除节点失败", e);
- }
- }
- }
分享名称:分布式协调框架Zookeeper核心设计理解与实战
网站网址:http://www.shufengxianlan.com/qtweb/news45/276245.html
网站建设、网络推广公司-创新互联,是专注品牌与效果的网站制作,网络营销seo公司;服务项目有等
声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联