devbox
喵喵爱编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Zookeeper-基础_zk连接的properties

作者：喵喵爱编程 | 2024-08-05 19:23:47

赞

踩

zk连接的properties

1. 简介

zookeeper是一个开源的分布式协调服务, 提供分布式数据一致性解决方案,分布式应用程序可以实现数据统一配置管理、统一命名服务、分布式锁、集群管理等功能.

ZooKeeper主要服务于分布式系统，使用分布式系统就无法避免对节点管理的问题(需要实时感知节点的状态、对节点进行统一管理等等)，而由于这些问题处理起来可能相对麻烦和提高了系统的复杂性，ZooKeeper作为一个能够通用解决这些问题的中间件就应运而生了。

2. 数据模型

2.1 模型结构

ZooKeeper 提供的命名空间很像标准文件系统的命名空间。名称是由斜杠 (/) 分隔的一系列路径元素。ZooKeeper 命名空间中的每个节点都由路径标识。

与典型的为存储而设计的文件系统不同，ZooKeeper 数据保存在内存中，这意味着 ZooKeeper 可以实现高吞吐量和低延迟。

2.2 模型的特点

  • 每个子目录如/app1都被称作一个znode（节点）。这个znode是被它所在的路径唯一标识。
  • znode可以有子节点目录，并且每个znode可以存储数据。
  • znode是有版本的，每个znode中存储的数据可以有多个版本，也就是一个访问路径中可以存储多份数据。每次 znode 的数据更改时，版本号都会增加。例如，每当客户端检索数据时，它也会收到数据的版本。
  • 每个znode都有一个访问控制列表 (ACL)，它限制了谁可以做什么。
  • znode可以被监控，包括这个目录中存储的数据的修改，子节点目录变化等，一旦变化可以通知设置监控的客户端。

⚠️注意：当使用zkCli.sh 创建会话时，节点的监听事件只能被触发一次。

2.3 节点分类

2.3.1 Persistent

持久节点：节点被创建后，就一直存在，除非客户端主动删除这个节点。

2.3.2 Persistent Sequential

持久顺序节点：有序的持久节点。在zk中，每个父节点会为它的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。例如：

 

 
 
$ create -s /app1 # 创建 /app1 节点
 
 
Created /app10000000000 # 创建成功后的节点名称为 /app10000000000
 

 

2.3.3 Ephemeral
 

临时节点：和持久节点不同的是，临时节点的生命周期和客户端会话绑定且临时节点下面不能创建子节点。如果客户端会话失效，那么这个节点就会自动被清除掉。
 

 
 
注意：客户端失效临时节点会被清除，但如果是断开链接，临时节点并不会立马被清除。
 
 
“立马”：在会话超时持续时间内未从客户端收到任何心跳信号之后，zk服务器将删除该会话的临时节点。但如果正常关闭会话，临时节点会立马被清除。
 

 

 

  1. # 1. 创建会话 
  2. $ ./zkCli.sh 
  3. # 2. 创建临时节点 /app3 
  4. $ create -e /app3 
  5. Created /app3 
  6. # 3. ctrl + c 关闭会话 
  7. # 4. 紧接着再次创建会话 
  8. $ ./zkCli.sh 
  9. # 5. 查看节点内容 发现临时节点依旧存在 
  10. $ ls / 
  11. [app10000000000, app3, zookeeper] 
  12. # 6. 等待几秒，再次查看 发现临时节点消失 
  13. $ ls / 
  14. [app10000000000, zookeeper] 
  15. # 7. 再次创建临时节点 /app3 
  16. $ create -e /app3 
  17. Created /app3 
  18. # 8. 正常关闭会话 
  19. $ quit 
  20. # 9. 再次创建会话 临时节点消失 
  21. $ ./zkCli.sh 
  22. $ ls / 
  23. [app10000000000, zookeeper]
 

2.3.4 Ephemeral Sequential
 

临时顺序节点：有序的临时节点。创建es节点时，zk会维护一份时序，会记录每个节点的顺序。例如：
 

  1. # 1. 创建会话
  2. $ ./zkCli.sh
  3. # 2. 创建临时节点 /app3
  4. $ create -e /app3
  5. Created /app3
  6. # 3. ctrl + c 关闭会话
  7. # 4. 紧接着再次创建会话 
  8. $ ./zkCli.sh
  9. # 5. 查看节点内容 发现临时节点依旧存在
  10. $ ls /
  11. [app10000000000, app3, zookeeper]
  12. # 6. 等待几秒，再次查看 发现临时节点消失
  13. $ ls /
  14. [app10000000000, zookeeper]
  15.  
  16. # 7. 再次创建临时节点 /app3
  17. $ create -e /app3
  18. Created /app3
  19. # 8. 正常关闭会话
  20. $ quit
  21. # 9. 再次创建会话 临时节点消失
  22. $ ./zkCli.sh
  23. $ ls /
  24. [app10000000000, zookeeper]
 

 
 
$ create -s -e /app2 # 创建临时有序节点 /app2
 
 
Created /app20000000001 # 创建成功后的节点名称为 app20000000001
 

 

3. 安装
 

3.1 官方
 

官方地址：Apache ZooKeeper
 

3.2 docker
 

 
 
$ docker run --name zookeeper --restart always -d zookeeper

 

 

3.2 docker-compose
 

 
 
⚠️注意：执行脚本前，需先将配置文件挂载到宿主机上。
 

 

  1. version: '3.1'
  2. services:
  3.   zk:
  4.     image: zookeeper
  5.     restart: always
  6.     container_name: zookeeper
  7.     ports:
  8.       - 2181:2181
  9.     volumes:
  10.       - ./data:/data
  11.       - ./logs:/datalog
  12.       - ./conf/zoo.cfg:/conf/zoo.cfg
 

 

3.3 配置信息
 

  1. # 数据存储目录
  2. dataDir=/data
  3. # 日志存储目录
  4. dataLogDir=/datalog
  5. # 集群模式下 节点之间的心跳时间2s（每2s进行一次心跳检测）
  6. tickTime=2000
  7. # 集群初始化时 节点之间同步次数。（5 * 2s = 10s 10s内未初始化成功则初始化失败）
  8. initLimit=5
  9. # 集群模式下 同步数据次数。 （2 * 2s = 4s 4s内未同步则超时失败）
  10. syncLimit=2
  11. # 数据快照保留个数
  12. autopurge.snapRetainCount=3
  13. # 数据快照清除时间间隔（单位为小时） 0：不清除 ，1：1小时 
  14. autopurge.purgeInterval=0
  15. # 最大的客户端链接数
  16. maxClientCnxns=60
 

4. 基础命令
 

不同版本之前的命令会有所差异，本章是基于zk:3.7版本。官方地址
 

4.1 创建会话
 

首先执行命令，创建新的会话，进入终端。
 

  1. # 进入到zk的安装目录的bin目录下执行
  2. $ ./zkCli.sh
  3. # 如果zk不是在本机 则可以使用server参数指定链接地址
  4. $ ./zkCli.sh -server 127.0.0.1:2181
 

 

4.2 ls
 

ls [-s] [-w] [-R] path：ls 命令用于查看某个路径下目录列表。
 

  • -s：查看此znode状态信息。
  • -w：监听此znode目录变化。
  • -R：递归查看。
 

  1. $ ls / # 查看根目录
  2. [app, app10000000000, zookeeper]
  3. $ ls /app
  4. []
  5. $ ls -s /app
  6. [app01, app02]
  7. cZxid = 0x1b
  8. ctime = Sun Aug 29 13:07:24 UTC 2021
  9. mZxid = 0x1b
  10. mtime = Sun Aug 29 13:07:24 UTC 2021
  11. pZxid = 0x2e
  12. cversion = 2
  13. dataVersion = 0
  14. aclVersion = 0
  15. ephemeralOwner = 0x0
  16. dataLength = 0
  17. numChildren = 2
 

 

监听znode目录变化：
 

 

 

4.3 create
 

create [-s] [-e] [-c] [-t ttl] path [data] [acl]：create 用于创建znode。
 

  •  
    -s：创建顺序节点。
     
  •  
    -e：创建临时节点。
     
  •  
    -c：创建一个容器节点，当容器中的最后一个子节点被删除时，容器也随之消失。
     
  •  
    -t：设置节点存活时间（毫秒）。
     
  
     
   
    ⚠️注意：如果要设置超时时间需在配置文件中激活配置：zookeeper.extendedTypesEnabled=true
     
  
     
  •  
    data：设置的数据。
     
  •  
    acl：访问控制配置。
     
 

  1. $ ls /
  2. [app, app10000000000, zookeeper]
  3. # 创建持久节点 /app3
  4. $ create /app3 zhangtieniu
  5. Created /app3
  6. # 创建有序的持久节点 /app4
  7. $ create -s /app4 zhangsan
  8. Created /app40000000011
  9. # 创建临时节点 /app5
  10. $ create -e /app5 linshi
  11. Created /app5
  12. # 创建临时有序节点 /app6
  13. $ create -s -e /app6 linshiyouxu
  14. Created /app60000000013
  15. $ ls /
  16. [app, app10000000000, app3, app40000000011, app5, app60000000013, zookeeper]
 

 

4.4 get
 

get [-s] [-w] path： 命令用于获取节点数据和状态信息。
 

  • -s：查看此znode状态信息。
  • -w：监听此znode值变化。
 

 
 
$ get /app3 # 查看/app3 znode内容 
zhangtieniu

 

 

 

监听znod值变化：
 

 

4.5 stat
 

stat [-w] path：命令用于查看节点状态信息。
 

  • -w：监听节点状态变化。
 

  1. $ stat /app3
  2. cZxid = 0x34 # 创建节点时的事务id
  3. ctime = Sun Aug 29 14:31:52 UTC 2021 # 创建时间
  4. mZxid = 0x34 # 创建的版本号
  5. mtime = Sun Aug 29 14:31:52 UTC 2021 # 修改时间
  6. pZxid = 0x34 # 子节点列表最后一次被修改的事务id
  7. cversion = 0 # 节点版本号
  8. dataVersion = 0 # 数据版本号
  9. aclVersion = 0 # 权限版本号 
  10. ephemeralOwner = 0x0 # 临时节点sessionid，当前节点为持久节点时，默认为“0x0”
  11. dataLength = 11 # 节点值数据长度
  12. numChildren = 0 # 子节点数量
 

 

4.6 set
 

set [-s] [-v version] path data： 命令用于修改节点存储的数据。
 

  • -s：查看设置成功后的状态信息。
  • -v：指定设置值的版本号，该值只能为该节点的最新版本。可以使用其实现乐观锁。
 

  1. $ set /app wangmazi
  2. $ set -s /app lisi
  3. cZxid = 0x1b
  4. ctime = Sun Aug 29 13:07:24 UTC 2021
  5. mZxid = 0x43
  6. mtime = Sun Aug 29 14:50:52 UTC 2021
  7. pZxid = 0x3f
  8. cversion = 4
  9. dataVersion = 6
  10. aclVersion = 0
  11. ephemeralOwner = 0x0
  12. dataLength = 4
  13. numChildren = 4
 

 

4.7 delete
 

delete [-v version] path：删除指定节点。
 

 
 
⚠️注意：当删除的节点有子节点时，不能使用该命令。需使用deleteall命令删除。
 

 

  • -v：删除指定版本。与set同理
 

 
 
$ delete /app3

 

 

 

4.8 quit
 

关闭会话。
 

5. 节点的监听机制
 

zk的监听机制，可以使客户端可以监听znode节点的变化，znode节点的变化出发相应的事件，然后清除对该节点的检测。
 

 
 
⚠️注意：在这里再强调一次，当使用zkCli.sh 创建会话时，对znode的监听只能触发一次。但在使用java客户端链接时，可以一直触发。
 

 

 
 
$ ls -w /path # 监听节点目录的变化 
$ get -w /path # 监听节点数据的变化

 

 

6. quick start
 

6.1 项目结构
 

 

 

 

  1. .
  2. ├── pom.xml
  3. └── src
  4.     └── main
  5.         ├── java
  6.         │   └── com
  7.         │       └── ldx
  8.         │           └── zookeeper
  9.         │               ├── ZookeeperApplication.java # 启动类
  10.         │               ├── config
  11.         │               │   ├── CuratorClientConfig.java # zk配置类
  12.         │               │   └── CuratorClientProperties.java # zk 配置属性文件
  13.         │               ├── controller
  14.         │               │   └── AppController.java # zk 测试文件
  15.         │               └── util
  16.         │                   └── CuratorClient.java # zk工具类
  17.         └── resources
  18.             └── application.yaml # 服务配置文件
 

6.2 引入依赖
 

 
 
Curator 是 Netflix 公司开源的一套 zookeeper 客户端框架，解决了很多 Zookeeper 客户端非常底层的细节开发工作，包括连接重连、反复注册 Watcher 和 NodeExistsException 异常等。
 
 
curator-recipes：封装了一些高级特性，如：Cache 事件监听、选举、分布式锁、分布式计数器、分布式 Barrier 等。
 

 

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.5.4</version>
  9.         <relativePath/> <!-- lookup parent from repository -->
  10.     </parent>
  11.     <groupId>com.ldx</groupId>
  12.     <artifactId>zookeeper</artifactId>
  13.     <version>0.0.1-SNAPSHOT</version>
  14.     <name>zookeeper</name>
  15.     <description>Demo project for Spring Boot</description>
  16.     <properties>
  17.         <java.version>1.8</java.version>
  18.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-web</artifactId>
  23.         </dependency>
  24.        <!-- client 操作工具包 -->
  25.         <dependency>
  26.             <groupId>org.apache.curator</groupId>
  27.             <artifactId>curator-recipes</artifactId>
  28.             <version>5.2.0</version>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-configuration-processor</artifactId>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>commons-lang</groupId>
  36.             <artifactId>commons-lang</artifactId>
  37.             <version>2.6</version>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.projectlombok</groupId>
  41.             <artifactId>lombok</artifactId>
  42.             <optional>true</optional>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.springframework.boot</groupId>
  46.             <artifactId>spring-boot-starter-test</artifactId>
  47.             <scope>test</scope>
  48.         </dependency>
  49.     </dependencies>
  50.  
  51.     <build>
  52.         <plugins>
  53.             <plugin>
  54.                 <groupId>org.springframework.boot</groupId>
  55.                 <artifactId>spring-boot-maven-plugin</artifactId>
  56.                 <configuration>
  57.                     <excludes>
  58.                         <exclude>
  59.                             <groupId>org.projectlombok</groupId>
  60.                             <artifactId>lombok</artifactId>
  61.                         </exclude>
  62.                     </excludes>
  63.                 </configuration>
  64.             </plugin>
  65.         </plugins>
  66.     </build>
  67.  
  68. </project>
 

 

6.3 application.yaml
 

  1. server:
  2.   port: 8080
  3. # curator配置
  4. curator-client:
  5.   # 连接字符串
  6.   connection-string: localhost:2181
  7.   # 根节点
  8.   namespace: ldx
  9.   # 节点数据编码
  10.   charset: utf8
  11.   # session超时时间
  12.   session-timeout-ms: 60000
  13.   # 连接超时时间
  14.   connection-timeout-ms: 15000
  15.   # 关闭连接超时时间
  16.   max-close-wait-ms: 1000
  17.   # 默认数据
  18.   default-data: ""
  19.   # 当半数以上zookeeper服务出现故障仍然提供读服务
  20.   can-be-read-only: false
  21.   # 自动创建父节点
  22.   use-container-parents-if-available: true
  23.   # 重试策略，默认使用BoundedExponentialBackoffRetry
  24.   retry:
  25.     max-sleep-time-ms: 10000
  26.     base-sleep-time-ms: 1000
  27.     max-retries: 3
  28.   # 认证信息
  29.   #auth:
  30.     #scheme: digest
  31.     # auth: username:password
 

 

6.4 CuratorClientProperties
 

  1. package com.ldx.zookeeper.config;
  2.  
  3. import lombok.Data;
  4. import org.springframework.boot.context.properties.ConfigurationProperties;
  5.  
  6. /**
  7.  * zk 属性配置类
  8.  *
  9.  * @author ludangxin
  10.  * @date 2021/8/31
  11.  */
  12. @Data
  13. @ConfigurationProperties(prefix = "curator-client")
  14. public class CuratorClientProperties {
  15.     /**
  16.      * 连接地址
  17.      */
  18.     private String connectionString;
  19.     /**
  20.      * 命名空间
  21.      */
  22.     private String namespace;
  23.     /**
  24.      * 字符集
  25.      */
  26.     private String charset = "utf8";
  27.     /**
  28.      * 会话超时时间 毫秒
  29.      */
  30.     private int sessionTimeoutMs = 60000;
  31.     /**
  32.      * 连接超时时间 毫秒
  33.      */
  34.     private int connectionTimeoutMs = 15000;
  35.     /**
  36.      * 最大关闭等待时间 毫秒
  37.      */
  38.     private int maxCloseWaitMs = 1000;
  39.     /**
  40.      * 默认数据
  41.      */
  42.     private String defaultData = "";
  43.     /**
  44.      * 当半数以上zookeeper服务出现故障仍然提供读服务
  45.      */
  46.     private boolean canBeReadOnly = false;
  47.     /**
  48.      * 自动创建父节点
  49.      */
  50.     private boolean useContainerParentsIfAvailable = true;
  51.     /**
  52.      * 线程池名称
  53.      */
  54.     private String threadFactoryClassName;
  55.     private Retry retry = new Retry();
  56.     private Auth auth = new Auth();
  57.  
  58.     @Data
  59.     public static class Retry {
  60.         private int maxSleepTimeMs = 10000;
  61.         private int baseSleepTimeMs = 1000;
  62.         private int maxRetries = 3;
  63.     }
  64.  
  65.     @Data
  66.     public static class Auth {
  67.         private String scheme = "digest";
  68.         private String auth;
  69.     }
  70.  
  71. }
 

 

6.5 CuratorClientConfig
 

  1. package com.ldx.zookeeper.config;
  2.  
  3. import com.ldx.zookeeper.util.CuratorClient;
  4. import lombok.SneakyThrows;
  5. import lombok.extern.slf4j.Slf4j;
  6. import org.apache.commons.lang.StringUtils;
  7. import org.apache.curator.RetryPolicy;
  8. import org.apache.curator.ensemble.EnsembleProvider;
  9. import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
  10. import org.apache.curator.framework.CuratorFrameworkFactory;
  11. import org.apache.curator.framework.api.ACLProvider;
  12. import org.apache.curator.framework.api.CompressionProvider;
  13. import org.apache.curator.framework.imps.GzipCompressionProvider;
  14. import org.apache.curator.retry.BoundedExponentialBackoffRetry;
  15. import org.apache.curator.utils.DefaultZookeeperFactory;
  16. import org.apache.curator.utils.ZookeeperFactory;
  17. import org.apache.zookeeper.ZooDefs;
  18. import org.apache.zookeeper.data.ACL;
  19. import org.springframework.boot.context.properties.EnableConfigurationProperties;
  20. import org.springframework.context.annotation.Bean;
  21. import org.springframework.context.annotation.Configuration;
  22. import java.nio.charset.Charset;
  23. import java.util.List;
  24. import java.util.concurrent.ThreadFactory;
  25.  
  26. /**
  27.  * zk 配置类
  28.  *
  29.  * @author ludangxin
  30.  * @date 2021/8/31
  31.  */
  32. @Slf4j
  33. @Configuration
  34. @EnableConfigurationProperties(CuratorClientProperties.class)
  35. public class CuratorClientConfig {
  36.     @Bean
  37.     public EnsembleProvider ensembleProvider(CuratorClientProperties curatorClientProperties) {
  38.         return new FixedEnsembleProvider(curatorClientProperties.getConnectionString());
  39.     }
  40.  
  41.     @Bean
  42.     public RetryPolicy retryPolicy(CuratorClientProperties curatorClientProperties) {
  43.         CuratorClientProperties.Retry retry = curatorClientProperties.getRetry();
  44.         return new BoundedExponentialBackoffRetry(retry.getBaseSleepTimeMs(), retry.getMaxSleepTimeMs(), retry.getMaxRetries());
  45.     }
  46.  
  47.     @Bean
  48.     public CompressionProvider compressionProvider() {
  49.         return new GzipCompressionProvider();
  50.     }
  51.  
  52.     @Bean
  53.     public ZookeeperFactory zookeeperFactory() {
  54.         return new DefaultZookeeperFactory();
  55.     }
  56.  
  57.     @Bean
  58.     public ACLProvider aclProvider() {
  59.         return new ACLProvider() {
  60.             @Override
  61.             public List<ACL> getDefaultAcl() {
  62.                 return ZooDefs.Ids.CREATOR_ALL_ACL;
  63.             }
  64.             @Override
  65.             public List<ACL> getAclForPath(String path) {
  66.                 return ZooDefs.Ids.CREATOR_ALL_ACL;
  67.             }
  68.         };
  69.     }
  70.  
  71.     @Bean
  72.     @SneakyThrows
  73.     public CuratorFrameworkFactory.Builder builder(EnsembleProvider ensembleProvider,
  74.                                                    RetryPolicy retryPolicy,
  75.                                                    CompressionProvider compressionProvider,
  76.                                                    ZookeeperFactory zookeeperFactory,
  77.                                                    ACLProvider aclProvider,
  78.                                                    CuratorClientProperties curatorClientProperties) {
  79.         String charset = curatorClientProperties.getCharset();
  80.         CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
  81.                 .ensembleProvider(ensembleProvider)
  82.                 .retryPolicy(retryPolicy)
  83.                 .compressionProvider(compressionProvider)
  84.                 .zookeeperFactory(zookeeperFactory)
  85.                 .namespace(curatorClientProperties.getNamespace())
  86.                 .sessionTimeoutMs(curatorClientProperties.getSessionTimeoutMs())
  87.                 .connectionTimeoutMs(curatorClientProperties.getConnectionTimeoutMs())
  88.                 .maxCloseWaitMs(curatorClientProperties.getMaxCloseWaitMs())
  89.                 .defaultData(curatorClientProperties.getDefaultData().getBytes(Charset.forName(charset)))
  90.                 .canBeReadOnly(curatorClientProperties.isCanBeReadOnly());
  91.         if (!curatorClientProperties.isUseContainerParentsIfAvailable()) {
  92.             builder.dontUseContainerParents();
  93.         }
  94.         CuratorClientProperties.Auth auth = curatorClientProperties.getAuth();
  95.         if (StringUtils.isNotBlank(auth.getAuth())) {
  96.             builder.authorization(auth.getScheme(), auth.getAuth().getBytes(Charset.forName(charset)));
  97.             builder.aclProvider(aclProvider);
  98.         }
  99.         String threadFactoryClassName = curatorClientProperties.getThreadFactoryClassName();
  100.         if (StringUtils.isNotBlank(threadFactoryClassName)) {
  101.             try {
  102.                 Class<?> cls = Class.forName(threadFactoryClassName);
  103.                 ThreadFactory threadFactory = (ThreadFactory) cls.newInstance();
  104.                 builder.threadFactory(threadFactory);
  105.             } catch (Exception e) {
  106.                 log.error("init CuratorClient error", e);
  107.             }
  108.         }
  109.         return builder;
  110.     }
  111.  
  112.     @Bean(initMethod = "init", destroyMethod = "stop")
  113.     public CuratorClient curatorClient(CuratorFrameworkFactory.Builder builder) {
  114.         return new CuratorClient(builder);
  115.     }
  116.  
  117. }
  118.  
 

 

6.6 CuratorClient
 

  1. package com.ldx.zookeeper.util;
  2.  
  3. import lombok.SneakyThrows;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.curator.framework.CuratorFramework;
  6. import org.apache.curator.framework.CuratorFrameworkFactory;
  7. import org.apache.curator.framework.recipes.cache.CuratorCache;
  8. import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
  9. import org.apache.curator.framework.recipes.locks.*;
  10. import org.apache.curator.framework.state.ConnectionState;
  11. import org.apache.zookeeper.CreateMode;
  12. import java.nio.charset.Charset;
  13. import java.util.List;
  14. import java.util.Objects;
  15. import java.util.concurrent.Executor;
  16. import java.util.concurrent.TimeUnit;
  17.  
  18. /**
  19.  * zookeeper工具类
  20.  *
  21.  * @author ludangxin
  22.  * @date 2021/8/30
  23.  */
  24. @Slf4j
  25. public class CuratorClient {
  26.     /**
  27.      * 默认的字符编码集
  28.      */
  29.     private static final String DEFAULT_CHARSET = "utf8";
  30.     /**
  31.      * 客户端
  32.      */
  33.     private final CuratorFramework client;
  34.     /**
  35.      * 字符集
  36.      */
  37.     private String charset = DEFAULT_CHARSET;
  38.  
  39.     @SneakyThrows
  40.     public CuratorClient(CuratorFrameworkFactory.Builder builder) {
  41.         client = builder.build();
  42.     }
  43.  
  44.     @SneakyThrows
  45.     public CuratorClient(CuratorFrameworkFactory.Builder builder, String charset) {
  46.         client = builder.build();
  47.         this.charset = charset;
  48.     }
  49.  
  50.     public void init() {
  51.         client.start();
  52.         client.getConnectionStateListenable().addListener((client, state) -> {
  53.             if (state==ConnectionState.LOST) {
  54.                 // 连接丢失
  55.                 log.info("lost session with zookeeper");
  56.             } else if (state==ConnectionState.CONNECTED) {
  57.                 // 连接新建
  58.                 log.info("connected with zookeeper");
  59.             } else if (state==ConnectionState.RECONNECTED) {
  60.                 // 重新连接
  61.                 log.info("reconnected with zookeeper");
  62.             }
  63.         });
  64.     }
  65.  
  66.     /**
  67.      * 关闭会话
  68.      */
  69.     public void stop() {
  70.         log.info("zookeeper session close");
  71.         client.close();
  72.     }
  73.  
  74.     /**
  75.      * 创建节点
  76.      *
  77.      * @param mode     节点类型
  78.      *   1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
  79.      *   2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
  80.      *   3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
  81.      *   4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已经存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
  82.      * @param path     节点名称
  83.      * @param nodeData 节点数据
  84.      */
  85.     @SneakyThrows
  86.     public void createNode(CreateMode mode, String path, String nodeData) {
  87.        // 使用creatingParentContainersIfNeeded()之后Curator能够自动递归创建所有所需的父节点
  88.        client.create().creatingParentsIfNeeded().withMode(mode).forPath(path, nodeData.getBytes(Charset.forName(charset)));
  89.     }
  90.  
  91.     /**
  92.      * 创建节点
  93.      *
  94.      * @param mode 节点类型
  95.      *   1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
  96.      *   2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
  97.      *   3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
  98.      *   4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已经存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
  99.      * @param path 节点名称
  100.      */
  101.     @SneakyThrows
  102.     public void createNode(CreateMode mode, String path) {
  103.        // 使用creatingParentContainersIfNeeded()之后Curator能够自动递归创建所有所需的父节点
  104.        client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
  105.     }
  106.  
  107.     /**
  108.      * 删除节点数据
  109.      *
  110.      * @param path 节点名称
  111.      */
  112.     @SneakyThrows
  113.     public void deleteNode(final String path) {
  114.        deleteNode(path, true);
  115.     }
  116.  
  117.     /**
  118.      * 删除节点数据
  119.      *
  120.      * @param path 节点名称
  121.      * @param deleteChildre 是否删除子节点
  122.      */
  123.     @SneakyThrows
  124.     public void deleteNode(final String path, Boolean deleteChildre) {
  125.        if (deleteChildre) {
  126.            // guaranteed()删除一个节点，强制保证删除,
  127.            // 只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到删除节点成功
  128.            client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
  129.        } else {
  130.            client.delete().guaranteed().forPath(path);
  131.        }
  132.     }
  133.  
  134.     /**
  135.      * 设置指定节点的数据
  136.      *
  137.      * @param path 节点名称
  138.      * @param data 节点数据
  139.      */
  140.     @SneakyThrows
  141.     public void setNodeData(String path, String data) {
  142.        client.setData().forPath(path, data.getBytes(Charset.forName(charset)));
  143.     }
  144.  
  145.     /**
  146.      * 获取指定节点的数据
  147.      *
  148.      * @param path 节点名称
  149.      * @return 节点数据
  150.      */
  151.     @SneakyThrows
  152.     public String getNodeData(String path) {
  153.        return new String(client.getData().forPath(path), Charset.forName(charset));
  154.     }
  155.  
  156.     /**
  157.      * 获取数据时先同步
  158.      *
  159.      * @param path 节点名称
  160.      * @return 节点数据
  161.      */
  162.     public String synNodeData(String path) {
  163.         client.sync();
  164.         return getNodeData(path);
  165.     }
  166.  
  167.     /**
  168.      * 判断节点是否存在
  169.      *
  170.      * @param path 节点名称
  171.      * @return true 节点存在，false 节点不存在
  172.      */
  173.     @SneakyThrows
  174.     public boolean isExistNode(final String path) {
  175.         client.sync();
  176.         return Objects.nonNull(client.checkExists().forPath(path));
  177.     }
  178.  
  179.     /**
  180.      * 获取节点的子节点
  181.      *
  182.      * @param path 节点名称
  183.      * @return 子节点集合
  184.      */
  185.     @SneakyThrows
  186.     public List<String> getChildren(String path) {
  187.         return client.getChildren().forPath(path);
  188.     }
  189.  
  190.     /**
  191.      * 创建排他锁
  192.      *
  193.      * @param path 节点名称
  194.      * @return 排他锁
  195.      */
  196.     public InterProcessSemaphoreMutex getSemaphoreMutexLock(String path) {
  197.        return new InterProcessSemaphoreMutex(client, path);
  198.     }
  199.  
  200.     /**
  201.      * 创建可重入排他锁
  202.      *
  203.      * @param path 节点名称
  204.      * @return 可重入排他锁
  205.      */
  206.     public InterProcessMutex getMutexLock(String path) {
  207.        return new InterProcessMutex(client, path);
  208.     }
  209.  
  210.     /**
  211.      * 创建一组可重入排他锁
  212.      *
  213.      * @param paths 节点名称集合
  214.      * @return 锁容器
  215.      */
  216.     public InterProcessMultiLock getMultiMutexLock(List<String> paths) {
  217.         return new InterProcessMultiLock(client, paths);
  218.     }
  219.  
  220.     /**
  221.      * 创建一组任意类型的锁
  222.      *
  223.      * @param locks 锁集合
  224.      * @return 锁容器
  225.      */
  226.     public InterProcessMultiLock getMultiLock(List<InterProcessLock> locks) {
  227.         return new InterProcessMultiLock(locks);
  228.     }
  229.  
  230.     /**
  231.      * 加锁
  232.      *
  233.      * @param lock 分布式锁对象
  234.      */
  235.     @SneakyThrows
  236.     public void acquire(InterProcessLock lock) {
  237.        lock.acquire();
  238.     }
  239.  
  240.     /**
  241.      * 加锁
  242.      *
  243.      * @param lock 分布式锁对象
  244.      * @param time 等待时间
  245.      * @param unit 时间单位
  246.      */
  247.     @SneakyThrows
  248.     public void acquire(InterProcessLock lock, long time, TimeUnit unit) {
  249.        lock.acquire(time, unit);
  250.     }
  251.  
  252.     /**
  253.      * 释放锁
  254.      *
  255.      * @param lock 分布式锁对象
  256.      */
  257.     @SneakyThrows
  258.     public void release(InterProcessLock lock) {
  259.        lock.release();
  260.     }
  261.  
  262.     /**
  263.      * 检查是否当前jvm的线程获取了锁
  264.      *
  265.      * @param lock 分布式锁对象
  266.      * @return true/false
  267.      */
  268.     public boolean isAcquiredInThisProcess(InterProcessLock lock) {
  269.         return lock.isAcquiredInThisProcess();
  270.     }
  271.  
  272.     /**
  273.      * 获取读写锁
  274.      *
  275.      * @param path 节点名称
  276.      * @return 读写锁
  277.      */
  278.     public InterProcessReadWriteLock getReadWriteLock(String path) {
  279.         return new InterProcessReadWriteLock(client, path);
  280.     }
  281.  
  282.     /**
  283.      * 监听数据节点的变化情况
  284.      *
  285.      * @param path 节点名称
  286.      * @param listener 监听器
  287.      * @return 监听节点的TreeCache实例
  288.      */
  289.     @SneakyThrows
  290.     public CuratorCache watch(String path, CuratorCacheListener listener) {
  291.         CuratorCache curatorCache = CuratorCache.builder(client, path).build();
  292.         curatorCache.listenable().addListener(listener);
  293.         curatorCache.start();
  294.         return curatorCache;
  295.     }
  296.  
  297.     /**
  298.      * 监听数据节点的变化情况
  299.      *
  300.      * @param path 节点名称
  301.      * @param listener 监听器
  302.      * @return 监听节点的TreeCache实例
  303.      */
  304.     public CuratorCache watch(String path, CuratorCacheListener listener, Executor executor) {
  305.         CuratorCache curatorCache = CuratorCache.builder(client, path).build();
  306.         curatorCache.listenable().addListener(listener, executor);
  307.         curatorCache.start();
  308.         return curatorCache;
  309.     }
  310.  
  311.     /**
  312.      * 取消监听节点
  313.      *
  314.      * @param path 节点名称
  315.      * @param listener 监听器
  316.      */
  317.     public void unwatch(String path, CuratorCacheListener listener) {
  318.         CuratorCache curatorCache = CuratorCache.builder(client, path).build();
  319.         curatorCache.listenable().removeListener(listener);
  320.     }
  321.  
  322. }
 

 

6.7 AppController
 

  1. package com.ldx.zookeeper.controller;
  2.  
  3. import com.fasterxml.jackson.core.JsonProcessingException;
  4. import com.fasterxml.jackson.databind.ObjectMapper;
  5. import com.ldx.zookeeper.util.CuratorClient;
  6. import lombok.RequiredArgsConstructor;
  7. import lombok.extern.slf4j.Slf4j;
  8. import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
  9. import org.apache.zookeeper.CreateMode;
  10. import org.apache.zookeeper.data.Stat;
  11. import org.springframework.web.bind.annotation.*;
  12.  
  13. /**
  14.  * demo
  15.  *
  16.  * @author ludangxin
  17.  * @date 2021/8/30
  18.  */
  19. @Slf4j
  20. @RestController
  21. @RequestMapping("app")
  22. @RequiredArgsConstructor
  23. public class AppController {
  24.  
  25.    private final CuratorClient curatorClient;
  26.  
  27.    @GetMapping("{appName}")
  28.    public String getData(@PathVariable String appName) {
  29.       return curatorClient.getNodeData(setPrefix(appName));
  30.    }
  31.  
  32.    @PostMapping("{appName}")
  33.    public String addApp(@PathVariable String appName, @RequestParam String data) {
  34.       curatorClient.createNode(CreateMode.PERSISTENT, setPrefix(appName), data);
  35.       return "ok";
  36.    }
  37.  
  38.    @PostMapping("{appName}/{childName}")
  39.    public String addAppChild(@PathVariable String appName, @PathVariable String childName, @RequestParam String data) {
  40.       curatorClient.createNode(CreateMode.PERSISTENT, setPrefix(appName).concat(setPrefix(childName)), data);
  41.       return "ok";
  42.    }
  43.  
  44.    @PutMapping("{appName}")
  45.    public String setData(@PathVariable String appName, String data) {
  46.       curatorClient.setNodeData(setPrefix(appName), data);
  47.       return "ok";
  48.    }
  49.  
  50.    @PutMapping("{appName}/{childName}")
  51.    public String setData(@PathVariable String appName, @PathVariable String childName, String data) {
  52.       curatorClient.setNodeData(setPrefix(appName).concat(setPrefix(childName)), data);
  53.       return "ok";
  54.    }
  55.  
  56.    @DeleteMapping("{appName}")
  57.    public String delApp(@PathVariable String appName) {
  58.       curatorClient.deleteNode(setPrefix(appName));
  59.       return "ok";
  60.    }
  61.  
  62.    @PostMapping("{appName}/watch/dir")
  63.    public String watchAppDir(@PathVariable String appName) {
  64.       CuratorCacheListener listener = CuratorCacheListener.builder().forDeletes(obj -> {
  65.          String path = obj.getPath();
  66.          String data = new String(obj.getData());
  67.          Stat stat = obj.getStat();
  68.          log.info("节点：{} 被删除，节点数据：{}，节点状态：\\{version：{}，createTime:{}\\}", path, data,
  69.             stat.getVersion(), stat.getCtime());
  70.       }).build();
  71.       curatorClient.watch(setPrefix(appName), listener);
  72.       return "ok";
  73.    }
  74.  
  75.    @PostMapping("{appName}/watch/data")
  76.    public String watchAppData(@PathVariable String appName) {
  77.       ObjectMapper mapper = new ObjectMapper();
  78.       CuratorCacheListener listener = CuratorCacheListener.builder().forChanges((oldNode, newNode) -> {
  79.          try {
  80.             String path = oldNode.getPath();
  81.             log.info("节点：{} 被修改，修改前：{} ", path, mapper.writeValueAsString(oldNode));
  82.             log.info("节点：{} 被修改，修改后：{} ", path, mapper.writeValueAsString(newNode));
  83.          } catch(JsonProcessingException e) {
  84.             e.printStackTrace();
  85.          }
  86.       }).build();
  87.       curatorClient.watch(setPrefix(appName), listener);
  88.       return "ok";
  89.    }
  90.  
  91.    private String setPrefix(String appName) {
  92.       String prefix = "/";
  93.       if(!appName.startsWith(prefix)) {
  94.          appName = prefix.concat(appName);
  95.       }
  96.       return appName;
  97.    }
  98. }
 

 

7. ZAB 协议介绍
 

ZAB（ZooKeeper Atomic Broadcast 原子广播） 协议是为分布式协调服务 ZooKeeper 专门设计的一种支持崩溃恢复的原子广播协议。 在 ZooKeeper 中，主要依赖 ZAB 协议来实现分布式数据一致性，基于该协议，ZooKeeper 实现了一种主备模式的系统架构来保持集群中各个副本之间的数据一致性。
 

ZAB 协议两种基本的模式：崩溃恢复和消息广播
 

ZAB协议包括两种基本的模式，分别是 崩溃恢复和消息广播。当整个服务框架在启动过程中，或是当 Leader 服务器出现网络中断、崩溃退出与重启等异常情况时，ZAB 协议就会进人恢复模式并选举产生新的Leader服务器。当选举产生了新的 Leader 服务器，同时集群中已经有过半的机器与该Leader服务器完成了状态同步之后，ZAB协议就会退出恢复模式。其中，所谓的状态同步是指数据同步，用来保证集群中存在过半的机器能够和Leader服务器的数据状态保持一致。
 

当集群中已经有过半的Follower服务器完成了和Leader服务器的状态同步，那么整个服务框架就可以进人消息广播模式了。 当一台同样遵守ZAB协议的服务器启动后加人到集群中时，如果此时集群中已经存在一个Leader服务器在负责进行消息广播，那么新加人的服务器就会自觉地进人数据恢复模式：找到Leader所在的服务器，并与其进行数据同步，然后一起参与到消息广播流程中去。正如上文介绍中所说的，ZooKeeper设计成只允许唯一的一个Leader服务器来进行事务请求的处理。Leader服务器在接收到客户端的事务请求后，会生成对应的事务提案并发起一轮广播协议；而如果集群中的其他机器接收到客户端的事务请求，那么这些非Leader服务器会首先将这个事务请求转发给Leader服务器。
 

 

 

代码：Zookeeper-基础-Java文档类资源-CSDN下载
 

 

转载自：Zookeeper-基础 - 张铁牛 - 博客园

                

        
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/喵喵爱编程/article/detail/933832
推荐阅读
相关标签
Copyright ©  2003-2013 www.wpsshop.cn  版权所有，并保留所有权利。

  
闽ICP备14008679号