zookeeper的使用与发现

mac2022-06-30 88

一、概述

ZooKeeper是一个分布式应用所涉及的分布式的、开源的协调服务。

是Google的Chubby的开源实现

Zookeeper最早起源于雅虎的研究院的一个研究小组。在当时，研究人员发现，在雅虎内部很多大型的系统需要依赖一个类似的系统进行分布式协调，但是这些系统往往存在分布式单点问题。所以雅虎的开发人员就试图开发一个通用的无单点问题的分布式协调框架。在立项初期，考虑到很多项目都是用动物的名字来命名的(例如著名的Pig项目)，雅虎的工程师希望给这个项目也取一个动物的名字。时任研究院的首席科学家Raghu Ramakrishnan开玩笑说：再这样下去，我们这儿就变成动物园了。此话一出，大家纷纷表示就叫动物园管理员吧——因为各个以动物命名的分布式组件放在一起，雅虎的整个分布式系统看上去就像一个大型的动物园了，而Zookeeper正好用来进行分布式环境的协调——于是，Zookeeper的名字由此诞生了。

分布式的引用可以建立在同步配置管理、选举、分布式锁、分组和命名等服务的更高级的实现的基础之上。ZooKeeper意欲设计一个易于编程的环境，它的文件系统使用我们所熟悉的目录树结构。ZooKeeper使用Java所编写，但是支持Java和C两种编程语言。

二、ZooKeeper节点

ZooKeeper提供的命名空间与标准的文件系统非常相似。一个名称是由通过斜线分割开来的路径名序列所组成的。ZooKeeper中每一个节点都是通过路径来识别。

ZooKeeper的节点是通过像树一样的结构来进行维护，并且并且每一个节点通过路径标识以及访问。除此之外，每个节点还拥有自身的一些信息，包括：数据、数据长度、创建时间、修改时间等等。从这样一类既含有数据，又可以作为路径表示的节点特点中可以看出，ZooKeeper的节点既可以被看做是一个文件，又可以看做是一个目录，它同时具有二者的特点。

Znode还具有原子性操作的特点：命名空间中。每一个Znode的数据将被原子地读写。读操作将读取与Znode相关的所有数据，写操作将替换掉所有的数据，除此之外，每一个节点都有一个访问控制列表，这个访问控制列表规定了用户操作的权限。

ZooKeeper节点是有生命周期的，这取决于节点的类型。在ZooKeeper中，节点类型可以分为持久节点（PERSISTENT）、临时节点（EPHEMERAL），以及时序节点（SEQUENTIAL），具体在节点创建过程中。一般是组合使用，可以生成以下4中节点类型。

持久节点（PERSISTENT）

持久节点，是指在节点创建后，就一直存在，直到有删除操作来主动清楚这个节点——不会因为创建该节点的客户端端会话失效而消失。

持久顺序节点（PERSISTENT_SEQUENTIAL）

类节点的基本特性和持久节点类型是一致的。额外的特性是，在ZK中，每个父节点会为他的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。

临时节点（EPHEMERAL）

与持久节点不同的是，临时节点的生命周期和客户端会话绑定。也就是说，如果客户端会话失效，那么这个节点就会自动被清除掉。

临时顺序节点（EPHEMERAL_SEQUENTIAL）

类节点的基本特性和临时子节点类型是一致的。额外的特性是，在ZK中，每个父节点会为他的第一级

三、ZooKeeper使用场景

3.1 配置中心（数据发布与订阅）

在分布式应用中为了实现对分布式节点的统一配置，通常将服务中的配置文件集中存储在一个配置服务中，例如 SpringCloud将配置信息存储在Git/SVN中，Solr Cloud 则将配置数据集中存储在Zookeeper中。这典型利用了Zookeeper节点的发布订阅特性。

3.2 命名服务/服务分组（Naming Service）

命名服务也是分布式系统中比较常见的一类场景。在分布式系统中通过使用命名服务，客户端应用能够根据指定名字来获取资源或者服务的地址，提供者等信息。被命名的实体可以是集群中的机器，提供的服务地址，远程对象等等——这些我们都可以统称他们为名字（Name）。其中较为常见的就是一些分布式服务框架中的服务地址列表。通过调用ZK提供的创建节点的API，能够很容易常见一个全局唯一的path，这个path就就可以作为一个名称。继而实现服务弹性部署。

3.3 分布式锁（场景）

Redis分布式锁 VS Zookeeper设计分布式锁的优势

ZK有Watcher机制，实现简单，客户端无需重复读取数据Redis锁不稳定，客户端bug或者超时会导致锁失效，zk客户端挂掉znode直接销毁（临时节点）zk相对公平，顺序节点

四、ZooKeeper安装

ZooKeeper的安装模式分为三种，分别为：单机模式（stand-alone）、集群模式和集群伪分布模式。ZooKeeper单机模式的安装相对比较简单，如果第一次接触ZooKeeper建议安装ZooKeeper单机模式或者伪集群模式。

下载地址：http://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/

4.1 单机模式

zoo.cfg （存在于ZK安装目录下的conf目录下）

tickTime=2000 //session 过期时间为两倍的 tickTime dataDir=/home/zk/data //zk数据文件 clientPort=2181 //外部服务端口

4.2 启动

[root@GuoJiafeng01 zookeeper-3.4.6]# ./bin/zkServer.sh start ./conf/zk.conf JMX enabled by default Using config: ./conf/zk.conf Starting zookeeper ... STARTED 进程名称：QuorumPeerMain

五、ZK基本使用

5.1 启动

Usage: zkServer.sh {start|start-foreground|stop|restart|status|upgrade|print-cmd}

5.2 连接

./bin/zkCli.sh -server 192.168.15.130:2181

5.3 其他指令

connect host:port get path [watch] ls path [watch] set path data [version] rmr path quit printwatches on|off create [-s] [-e] path data acl 默认持久节点 -s 顺序节点 -e 临时节点 stat path [watch] close ls2 path [watch] history setAcl path acl getAcl path addauth scheme auth delete path [version]

5.4 close（关闭Session）

关闭当前连接，在命令行头部会显示CLOSED

5.6 connect （链接ZooKeeper）

connect localhost:2181

5.7 ls / ls2 （查看子节点）

[zk: localhost:2181(CONNECTED) 19] ls / [dubbo, hadoop-ha, zookeeper, baizhi] -------------------------------------- [zk: localhost:2181(CONNECTED) 20] ls2 / [dubbo, hadoop-ha, zookeeper, baizhi] cZxid = 0x0 ctime = Thu Jan 01 08:00:00 CST 1970 mZxid = 0x0 mtime = Thu Jan 01 08:00:00 CST 1970 pZxid = 0x1001 cversion = 4 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 0 numChildren = 4

5.8 create（创建节点）

[zk: localhost:2181(CONNECTED) 15] create /baizhi '1' Created /baizhi

5.9 delete（删除节点）

[zk: localhost:2181(CONNECTED) 21] delete /baizhi [zk: localhost:2181(CONNECTED) 22] ls / [dubbo, hadoop-ha, zookeeper]

5.10 rmr （递归删除节点）

[zk: localhost:2181(CONNECTED) 28] ls /baizhi [gjf] --------------------------------- [zk: localhost:2181(CONNECTED) 44] rmr /baizhi/gjf [zk: localhost:2181(CONNECTED) 46] ls /baizhi []

5.11 quit（退出ZookeeperSHell客户端）

quit

5.12 stat（查看节点状态）

[zk: localhost:2181(CONNECTED) 51] stat /baizhi cZxid = 0x1004 ctime = Tue Mar 12 15:28:00 CST 2019 mZxid = 0x1004 mtime = Tue Mar 12 15:28:00 CST 2019 pZxid = 0x100f cversion = 10 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0

六、Java API操作

6.1 原生 API

1）Maven依赖

<dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.6</version> <type>pom</type> </dependency>

2）相关操作

（1）获得客户端对象

private ZooKeeper zooKeeper; @Before public void getCline() throws Exception{ zooKeeper = new ZooKeeper("192.168.134.99:2181",2000,null); }

（2）获取节点值

@Test public void getData()throws Exception{ /*获取节点值*/ byte[] data = zooKeeper.getData("/baizhi/123", null, null); System.out.println(new String(data)); }

（3）获取节点孩子

@Test public void getChild() throws Exception{ /*获取当前输入节点下的子节点*/ List<String> children = zooKeeper.getChildren("/baizhi/gjf", null); children.forEach((a)->{ System.out.println(a); }); }

（4）创建节点

@Test public void createNode() throws Exception{ /*创建持久节点*/ Object value = null; value = "12312312"; byte[] bytes = ((String) value).getBytes(); String s = zooKeeper.create("/baizhi/123", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println(s); }

（5）更新节点值

@Test public void setData() throws Exception{ Stat stat = zooKeeper.setData("/baizhi/123", new String("123123").getBytes(), -1); System.out.println(stat); }

（6）判断节点是否存在

@Test public void exist() throws Exception{ Stat stat = zooKeeper.exists("/baizhi/13", false); if (stat==null) { System.out.println("该节点不存在！"); }else { System.out.println("该节点存在！"); } }

6.2 zkClinet API

1）Mavne依赖

<dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.8</version> </dependency>

2）核心API

ZkClient client=new ZkClient("ip:port");

3）相关操作

package test; import org.I0Itec.zkclient.ZkClient; import java.util.List; /** * Create by GuoJF on 2019/3/12 */ public class Main { public static void main(String[] args) { ZkClient zkClient = new ZkClient("192.168.134.5:2181"); /* * 获取根目录下所有的node信息 * * */ List<String> children = zkClient.getChildren("/"); children.forEach((a)->{ System.out.println(a); }); } }

（1）创建节点

@Test public void createNode() { String result = zkClient.create("/baizhi/001", "001", CreateMode.PERSISTENT); System.out.println(result); }

（2）获取节点

@Test public void getData() { Object readData = zkClient.readData("/baizhi/001"); System.out.println(readData.toString()); }

（3）更新节点值

@Test public void setData() { zkClient.writeData("/baizhi/001", "123123"); }

（4）删除节点

@Test public void delNode() { boolean delete = zkClient.delete("/baizhi/001"); if (delete) { System.out.println("删除成功！"); } else { System.out.println("删除失败或者节点不存在！"); } }

（5）判断节点是否存在

@Test public void exist(){ boolean exists = zkClient.exists("/baizhi/001"); if (exists){ System.out.println("该节点已经存在！"); }else { System.out.println("该节点不存在！"); } }

（6）获取孩子节点

@Test public void getChild(){ List<String> children = zkClient.getChildren("/baizhi"); for (String child : children) { System.out.println(child); } }

6.3 Curator API

1）Maven依赖

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

2）相关操作

（1）获取客户端对象

private CuratorFramework curatorFramework; @Before public void getClient() { /* * 重连策略四种实现 * ExponentialBackoffRetry、RetryNTimes、RetryOneTimes、RetryUntilElapsed * */ ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(1000, 1000); curatorFramework = CuratorFrameworkFactory.newClient("192.168.134.99:2181", backoffRetry); curatorFramework.start(); }

（2）创建节点

@Test public void createNode() throws Exception { String s = curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/baizhi/002", new String("123").getBytes()); System.out.println(s); }

（3）获取节点

@Test public void getData() throws Exception { byte[] bytes = curatorFramework.getData().forPath("/baizhi/002"); System.out.println(new String(bytes)); }

（4）更新节点值

@Test public void setData() { try { curatorFramework.setData().forPath("/baizhi/001", new String("123123").getBytes()); System.out.println("更新成功！"); } catch (KeeperException.NoNodeException e) { System.out.println("更新失败，该节点不存在！"); } catch (Exception e) { e.printStackTrace(); } }

（5）获取孩子节点

@Test public void getChild() throws Exception { curatorFramework.getChildren().forPath("/baizhi").forEach((child) -> { System.out.println(child); }); }

（6）删除节点

@Test public void delNode() throws Exception { curatorFramework.delete().forPath("/baizhi/002"); }

6.4 Watcher接口

1）原生API实现Watch接口

在这里使用的是原生的Apache的Java API

ZooKeeper Watcher监视使客户端能够接收来自ZooKeeper服务器的通知，并在发生时处理这些事件。 ZooKeeper Java API提供了一个名为Watcher的公共接口，客户端事件处理程序类必须实现该接口才能接收有关来自ZooKeeper服务器的事件通知。以编程方式，使用这种客户端的应用程序通过向客户端注册回调(callback)对象来处理这些事件。

package test; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import java.io.IOException; import java.util.UUID; /** * Create by GuoJF on 2019/3/12 */ public class DataUpdater { private static String hostPort = "192.168.134.5:2181"; private static String zooDataPath = "/baizhi/gjf"; ZooKeeper zk; public DataUpdater() throws IOException { try { zk = new ZooKeeper(hostPort, 2000, null); } catch (IOException e) { e.printStackTrace(); } } public void run() throws InterruptedException, KeeperException { while (true) { String uuid = UUID.randomUUID().toString(); byte zoo_data[] = uuid.getBytes(); zk.setData(zooDataPath, zoo_data, -1); try { Thread.sleep(5000); // Sleep for 5 secs } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } public static void main(String[] args) throws IOException, InterruptedException, KeeperException { DataUpdater dataUpdater = new DataUpdater(); dataUpdater.run(); } } package test; import org.apache.zookeeper.*; /** * Create by GuoJF on 2019/3/13 */ public class DataWatcher implements Watcher,Runnable { ZooKeeper zooKeeper; public DataWatcher() { try { zooKeeper = new ZooKeeper("192.168.134.5:2181",2000,this); if (zooKeeper.exists("/baizhi/gjf",this)==null) { zooKeeper.create("/baizhi/gjf", "get".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } }catch (Exception e){ e.printStackTrace(); } } public static void main(String[] args) { DataWatcher dataWatcher = new DataWatcher(); dataWatcher.run(); } @Override public void process(WatchedEvent event) { if (event.getType() == Event.EventType.NodeDataChanged){ System.out.println("数据发改变了吧"); try { byte[] data = zooKeeper.getData("/baizhi/gjf", this, null); String s = new String(data); System.out.println(s); }catch (Exception e){ } } } public void run() { try { synchronized (this) { while (true) { wait(); } } } catch (InterruptedException e) { e.printStackTrace(); Thread.currentThread().interrupt(); } } }

2）ZkClient实现Watcher接口

@Test public void watcherInterface() throws Exception { zkClient.subscribeDataChanges("/baizhi/gjf", new IZkDataListener() { @Override public void handleDataChange(String dataPath, Object data) throws Exception { System.out.println("节点路径：" + dataPath); System.out.println("节点值：" + data.toString()); } @Override public void handleDataDeleted(String dataPath) throws Exception { } }); Thread.sleep(Integer.MAX_VALUE); }

3）Curator API实现Watcher接口

ZooKeeper原生的API支持通过注册Watcher来进行事件监听，但是Watcher通知是一次性的，因此开发过程中需要反复注册Watcher，比较繁琐。Curator引入了Cache来监听ZooKeeper服务端的事件。Cache对ZooKeeper事件监听进行了封装，能够自动处理反复注册监听，简化了ZooKeeper原生API繁琐的开发过程。

（1）NodeCache

监听指定的数据变化

/* *NodeCache * * */ //在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理 ExecutorService pool = Executors.newFixedThreadPool(2); NodeCache nodeCache = new NodeCache(curatorFramework, "/baizhi/gjf", false); nodeCache.start(true); nodeCache.getListenable().addListener(new NodeCacheListener() { @Override public void nodeChanged() throws Exception { System.out.println(new String(nodeCache.getCurrentData().getData())); } }, pool);

（2）PathChildrenCache

监听指定节点下所有子节点的数据变化

/** * 监听子节点的变化情况 */ PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/baizhi", true); childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT); childrenCache.getListenable().addListener( new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { switch (event.getType()) { case CHILD_ADDED: System.out.println("CHILD_ADDED: " + event.getData().getPath()); break; case CHILD_REMOVED: System.out.println("CHILD_REMOVED: " + event.getData().getPath()); break; case CHILD_UPDATED: System.out.println("CHILD_UPDATED: " + event.getData().getPath()); break; default: break; } } }, pool );

七、Zookeeper ACL

7.1 Shell 操作

zookeeper本身提供了ACL机制，表示为scheme: id:permissions，第一个字段表示采用哪一种机制，第二个id表示用户，permissions表示相关权限（如只读，读写，管理等）。

7.1 .1 scheme ：id 介绍

world: 它下面只有一个id, 叫anyone, world:anyone代表任何人，zookeeper中对所有人有权限的结点就是属于world:anyone的auth: 它不需要id, 只要是通过authentication的user都有权限（zookeeper支持通过kerberos来进行authencation, 也支持username/password形式的authentication)，使用auth来设置权限的时候，需要在zk里注册一个用户才可以digest: 它对应的id为username:BASE64(SHA1(password))，它需要先通过username:password形式的authenticationip: 它对应的id为客户机的IP地址，设置的时候可以设置一个ip段，比如ip:192.168.1.0/16, 表示匹配前16个bit的IP段super: 在这种scheme情况下，对应的id拥有超级权限，可以做任何事情(cdrwa）

7.1 .2 permissions

权限ACL简写描述CREATEc可以创建子节点DELETEd可以删除子节点（仅下一级节点）READr可以读取节点数据及显示子节点列表WRITEw可以设置节点数据ADMINa可以设置节点访问控制列表权限

7.1.3 ACL Shell 命令

命令使用方式描述getAclgetAcl 读取ACL权限setAclsetAcl 设置ACL权限addauthaddauth 添加认证用户

7.1.4 操作

World scheme

其实默认就是Word Scheme

语法

setAcl <path> world:anyone:<acl> #随便创建一个节点 [zk: localhost:2181(CONNECTED) 61] create /baizhiedu 1 Created /baizhiedu [zk: localhost:2181(CONNECTED) 62] getAcl /baizhiedu 'world,'anyone : cdrwa

#在创建完成后相关节点，还可以通过setAcl的方式设置相关权限 [zk: localhost:2181(CONNECTED) 64] setAcl的方式设置相关权限 /baizhiedu world:anyone:cdrw cZxid = 0x1c631 ctime = Tue Jul 09 08:37:06 CST 2019 mZxid = 0x1c631 mtime = Tue Jul 09 08:37:06 CST 2019 pZxid = 0x1c631 cversion = 0 dataVersion = 0 aclVersion = 1 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0 [zk: localhost:2181(CONNECTED) 67] getAcl /baizhiedu 'world,'anyone : cdrw

IP scheme

对于特定IP适用，其他没有设置过的IP没有相关权限

语法

setAcl <path> ip:<ip>:<acl> [zk: localhost:2181(CONNECTED) 73] setAcl /baizhi01 ip:192.168.123.111:cdrwa cZxid = 0x1c635 ctime = Tue Jul 09 08:44:14 CST 2019 mZxid = 0x1c635 mtime = Tue Jul 09 08:44:14 CST 2019 pZxid = 0x1c635 cversion = 0 dataVersion = 0 aclVersion = 1 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0 [zk: localhost:2181(CONNECTED) 78] getAcl /baizhi01 'ip,'192.168.123.111 : cdrwa [zk: localhost:2181(CONNECTED) 79] get /baizhi01 Authentication is not valid : /baizhi01

Auth scheme

语法

addauth digest <user>:<password> #添加认证用户 setAcl <path> auth:<user>:<acl> [zk: localhost:2181(CONNECTED) 81] addauth digest gjf:root [zk: localhost:2181(CONNECTED) 82] setAcl /baizhi03 auth:gjf:root cZxid = 0x1c637 ctime = Tue Jul 09 08:47:00 CST 2019 mZxid = 0x1c637 mtime = Tue Jul 09 08:47:00 CST 2019 pZxid = 0x1c637 cversion = 0 dataVersion = 0 aclVersion = 1 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0 [zk: localhost:2181(CONNECTED) 95] getAcl /baizhi03 'digest,'gjf:bbYGkKPfBgiZDzcwrmVylqDlXnI= : cdrwa

Digest scheme

语法

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

计算密文

echo -n <user>:<password> | openssl dgst -binary -sha1 | openssl base64 [root@GuoJiafeng01 ~]# echo -n gjf:root | openssl dgst -binary -sha1 | openssl base64 bbYGkKPfBgiZDzcwrmVylqDlXnI= [zk: localhost:2181(CONNECTED) 98] create /baizhi04 1 Created /baizhi04 [zk: localhost:2181(CONNECTED) 99] setAcl /baizhi04 digest:gjf:bbYGkKPfBgiZDzcwrmVylqDlXnI=:a cZxid = 0x1c641 ctime = Tue Jul 09 08:59:18 CST 2019 mZxid = 0x1c641 mtime = Tue Jul 09 08:59:18 CST 2019 pZxid = 0x1c641 cversion = 0 dataVersion = 0 aclVersion = 1 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0 [zk: localhost:2181(CONNECTED) 100] getAcl /baizhi04 'digest,'gjf:bbYGkKPfBgiZDzcwrmVylqDlXnI= : a # 当前是没有权限的 [zk: localhost:2181(CONNECTED) 101] get /baizhi04 Authentication is not valid : /baizhi04 # 在当前session中添加认证用户 [zk: localhost:2181(CONNECTED) 102] addauth digest gjf:root #就能获取到相关的权限了 [zk: localhost:2181(CONNECTED) 107] get /baizhi04 1 cZxid = 0x1c641 ctime = Tue Jul 09 08:59:18 CST 2019 mZxid = 0x1c641 mtime = Tue Jul 09 08:59:18 CST 2019 pZxid = 0x1c641 cversion = 0 dataVersion = 0 aclVersion = 2 ephemeralOwner = 0x0 dataLength = 1 numChildren = 0

7.2 Java API

@Before public void getClient() { /* * 重连策略四种实现 * ExponentialBackoffRetry、RetryNTimes、RetryOneTimes、RetryUntilElapsed * */ ACLProvider aclProvider = new ACLProvider() { private List<ACL> acl ; @Override public List<ACL> getDefaultAcl() { if(acl ==null){ ArrayList<ACL> acl = ZooDefs.Ids.CREATOR_ALL_ACL; acl.clear(); acl.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", "admin:123") )); this.acl = acl; } return acl; } @Override public List<ACL> getAclForPath(String path) { return null; } }; ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(1000, 1000); //curatorFramework = CuratorFrameworkFactory.builder().aclProvider(aclProvider).authorization("digest", "admin:123".getBytes()).connectString("192.168.134.99:2181").retryPolicy(backoffRetry).build(); curatorFramework = CuratorFrameworkFactory.newClient("192.168.134.99:2181", backoffRetry); this.curatorFramework.start(); }

八、应用

8.1 分布式高可用（简单）

package test.ha; 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.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.data.Stat; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * Create by GuoJF on 2019/5/12 */ public class App { public static void main(String[] args) throws Exception { Integer port = 1233; ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(1000, 1000); CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("192.168.134.99:2181", backoffRetry); curatorFramework.start(); Stat stat = curatorFramework.checkExists().forPath("/baizhi/test01"); if (stat == null) { curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/baizhi/test01", new String("localhost:" + port).getBytes()); soutFunction(port); } else { System.out.println("当前节点已经存在，主机服务为：" + new String(curatorFramework.getData().forPath("/baizhi/test01"))); System.out.println("当前主机自动切换为StandyBy状态"); } /* *NodeCache * * */ //在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理 ExecutorService pool = Executors.newFixedThreadPool(2); /** * 监听子节点的变化情况 */ PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/baizhi", true); childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT); childrenCache.getListenable().addListener( new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { switch (event.getType()) { case CHILD_ADDED: System.out.println("节点上线: " + event.getData().getPath()); break; case CHILD_REMOVED: System.out.println("节点下线: " + event.getData().getPath()); System.out.println("正在上线其他节点，请稍后。"); if (event.getData().getPath().endsWith("/baizhi/test01")) { try { curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/baizhi/test01", new String("localhost:" + port).getBytes()); System.out.println("当前节点上线成功"); soutFunction(port); } catch (Exception e) { if (e instanceof KeeperException.NodeExistsException) { System.out.println("当前节点上线失败，已有其他节点上线"); } } } break; case CHILD_UPDATED: System.out.println("节点更新: " + event.getData().getPath()); break; default: break; } } }, pool ); Thread.sleep(Integer.MAX_VALUE); } public static void soutFunction(Integer port) { while (true) { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("localhost:" + port + ",正在提供服务"); } } }

8.1 分布式高可用（优雅）

package test.ha; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.framework.api.GetChildrenBuilder; import org.apache.curator.framework.recipes.cache.PathChildrenCache; import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent; import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.data.Stat; import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * Create by GuoJF on 2019/5/12 */ public class AppPro { public static void main(String[] args) throws Exception { final String[] lockip = {null}; final Integer[] port = {1233, 1234, 1235, 1236}; ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(1000, 1000); CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("192.168.123.129:2181", backoffRetry); curatorFramework.start(); List<String> stringList = curatorFramework.getChildren().forPath("/ha"); System.out.println(stringList.size()); if (stringList.size() == 0) { curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/ha/test01" + port[0], new String("localhost:" + port[0]).getBytes()); soutFunction(port[0]); } else { System.out.println("当前节点已经存在，主机服务为：" + lockip); System.out.println("当前主机自动切换为StandyBy状态"); } /* *NodeCache * * */ //在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理 ExecutorService pool = Executors.newFixedThreadPool(2); /** * 监听子节点的变化情况 */ PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/ha", true); childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT); final String[] currentPath = {""}; List<String> childers = curatorFramework.getChildren().forPath("/ha"); Collections.sort(childers); currentPath[0] = childers.get(0); childrenCache.getListenable().addListener( new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { switch (event.getType()) { case CHILD_ADDED: //System.out.println("节点上线: " + event.getData().getPath()); break; case CHILD_REMOVED: if (event.getData().getPath().endsWith(currentPath[0])) { System.out.println("节点下线: " + event.getData().getPath()); System.out.println("正在上线其他节点，请稍后。"); String path = curatorFramework.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/ha/test01", new String("localhost:" + port[0]).getBytes()); List<String> childers = curatorFramework.getChildren().forPath("/ha"); Collections.sort(childers); currentPath[0] = childers.get(0); if (path.endsWith(childers.get(0))) { System.out.println("当前节点上线成功"); soutFunction(port[0]); } else { System.out.println("当前节点上线失败！"); System.out.println("删除当前节点！"); try { curatorFramework.delete().forPath((path)); } catch (Exception e) { e.printStackTrace(); } System.out.println((path) + "节点已经删除！"); } } break; case CHILD_UPDATED: System.out.println("节点更新: " + event.getData().getPath()); break; default: break; } } }, pool ); Thread.sleep(Integer.MAX_VALUE); } public static void soutFunction(Integer port) { while (true) { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("localhost:" + port + ",正在提供服务"); } } }

九、集群搭建

9.1 集群配置

1）添加添加配置文件

tickTime=2000 dataDir=/home/zk/data clientPort=2181 initLimit=5 syncLimit=2 server.1=192.168.134.161:2887:3887 server.2=192.168.134.162:2887:3887 server.3=192.168.134.163:2887:3887

server.id=host:port:port

指出了不同的zk的服务器的自身的标示，作为集群中一部分及其应该知道集群中其他的机器，在datadir目录中创建一个文件名为myid的文件，这个文件仅仅含有一行内容，制定的是自身的id值。比如服务器的id是1就在这个文件写1，第一个port是保持和主机通信，第二个port是做选举的。

2）新建myid文件

在上述配置文件中对的data目录下，创建一个myid文件，在节点1中的myid中写 1 （就一个数字1），在节点2中写数字2，依次列推

3）复制配置启动

9.2 配置文件详解

1）ZK的最小配置

最小配置是指Zookeeper运行所需的最小配置，Zookeeper只需要配置这些项就可以正常的运行Zookeeper。

clientPort

配置ZK监听客户端连接的端口

dataDir

内存数据库快照存放地址，如果没有指定事务日志存放地址(dataLogDir)，默认也是存放在这个路径下，建议两个地址分开存放到不同的设备上。

tickTime

心跳基本时间单位，毫秒级

2）ZK的高级配置(可选）

高级配置是指有的需要直接通过系统属性进行设置)

dataLogDir

将事务日志存储在该路径下，比较重要，这个日志存储的设备效率会影响ZK的写吞吐量。

globalOutstandingLimit

(Java system property: zookeeper.globalOutstandingLimit)默认值是1000，限定了所有连接到服务器上但是还没有返回响应的请求个数(所有客户端请求的总数，不是连接总数)，这个参数是针对单台服务器而言，设定太大可能会导致内存溢出。

preAllocSize

(Java system property: zookeeper.preAllocSize)默认值64M，以KB为单位,预先分配额定空间用于后续transactionlog 写入，每当剩余空间小于4K时，就会又分配64M，如此循环。如果SNAP做得比较频繁(snapCount比较小的时候)，那么请减少这个值。

snapCount

(Java system property: zookeeper.snapCount)默认值100,000，当transaction每达到snapCount/2+rand.nextInt(snapCount/2)时，就做一次SNAPSHOT,默认情况下是50,000~100,000条transactionlog就会做一次，之所以用随机数是为了避免所有服务器可能在同一时间做snapshot.

traceFile (Java system property: requestTraceFile)

maxClientCnxns

默认值是10，一个客户端能够连接到同一个服务器上的最大连接数，根据IP来区分。如果设置为0，表示没有任何限制。设置该值一方面是为了防止DoS攻击。

clientPortAddress

与clientPort匹配，表示某个IP地址，如果服务器有多个网络接口(多个IP地址),如果没有设置这个属性，则clientPort会绑定到所有IP地址上，否则只绑定到该设置的IP地址上。

minSessionTimeout

最小的session time时间，默认值是2个tick time,客户端设置的session time 如果小于这个值，则会被强制协调为这个最小值。

maxSessionTimeout

最大的session time 时间，默认值是20个tick time. ,客户端设置的session time 如果大于这个值，则会被强制协调为这个最大值。

3）ZK的集群配置选项

electionAlg

领导选举算法，默认是3(fast leader election，基于TCP)，0表示leader选举算法(基于UDP)，1表示非授权快速选举算法(基于UDP)，2表示授权快速选举算法(基于UDP),目前1和2算法都没有应用，不建议使用，0算法未来也可能会被干掉，只保留3(fast leader election)算法，因此最好直接使用默认就好。

initLimit

tickTime的个数，表示在leader选举结束后，followers与leader同步需要的时间，如果followers比较多或者说leader的数据非常多时，同步时间相应可能会增加，那么这个值也需要相应增加。当然，这个值也是follower和observer在开始同步leader的数据时的最大等待时间(setSoTimeout)

syncLimit

tickTime的个数，这时间容易和上面的时间混淆，它也表示follower和observer与leader交互时的最大等待时间，只不过是在与leader同步完毕之后，进入正常请求转发或ping等消息交互时的超时时间。

leaderServes

(Java system property: zookeeper.leaderServes) 如果该值不是no，则表示该服务器作为leader时是需要接受客户端连接的。为了获得更高吞吐量，当服务器数三台以上时一般建议设置为no。

cnxTimeout

(Java system property: zookeeper.cnxTimeout) 默认值是5000,单位ms 表示leaderelection时打开连接的超时时间，只用在算法3中。

4）ZK安全配置

skipACL

(Java systemproperty: zookeeper.skipACL) 默认值是no,忽略所有ACL检查，相当于开放了所有数据权限给任何人。

forceSync

(Java systemproperty: zookeeper.forceSync) 默认值是yes, 表示transactionlog在commit时是否立即写到磁盘上，如果关闭这个选项可能会在断电时丢失信息。

jute.maxbuffer

(Java system property: jute.maxbuffer)默认值0xfffff，单位是KB，表示节点数据最多1M。如果要设置这个值，必须要在所有服务器上都需要设置。

授权认证配置项

DigestAuthenticationProvider.superDigest

(Java system property only: zookeeper.DigestAuthenticationProvider.superDigest) 设置这个值是为了确定一个超级用户，它的值格式为

super:<base64encoded(SHA1(idpassword))> ，一旦当前连接addAuthInfo超级用户验证通过，后续所有操作都不会checkACL.

十、一致性算法——Paxos、Raft、ZAB

10.1 CAP理论

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MZ0DmZmu-1569942793685)(assets/1557652887826.png)]

分布式系统的CAP理论：理论首先把分布式系统中的三个特性进行了如下归纳： ● 一致性（C）：在分布式系统中的所有数据备份，在同一时刻是否同样的值。（等同于所有节点访问同一份最新的数据副本）

● 可用性（A）：在集群中一部分节点故障后，集群整体是否还能响应客户端的读写请求。（对数据更新具备高可用性）

● 分区容错性（P）：以实际效果而言，分区相当于对通信的时限要求。系统如果不能在时限内达成数据一致性，就意味着发生了分区的情况，必须就当前操作在C和A之间做出选择。

10.2 什么是一致性？

有两种一致性模型，分别是弱一致性和强一致性

1）弱一致性

（1）最终一致性

DNS 域名系统（英文：Domain Name System，缩写：DNS）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4t8unpG9-1569942793687)(assets/1557653249165.png)]

在域名解析服务提供商上添加域名解析，一般都是10分钟以内之后才能通过域名访问到当前添加的网站，因为你的域名解析ip服务只在当前供应商的服务器上添加，同步到全球或者全国需要一定的时间，在其他地区或者其他DNS服务器就有可能访问不到当前网站，需要等待一段时间。

2 ）强一致性

（1）同步

（2）Paxos

（3）Raft（multi-paxos）

（4）ZAB（multi-paxos）

10.3 Paxos

1）明确问题

一致性算法的为什么会出现，因为数据不能存在单节点上，但是存在集群中有一致性的问题（网络、宕机等原因）。

但是有其他的算法比如说主从，但是主从的可用性极低，一旦有一个节点宕机则无法对外提供服务。

所以需要在保持一致性的同时尽可能的提高可用性。

2）基本思想

多数派：每次写入都要保证写入N/2的节点，每次读保证从何大于N/2个几点中读取多数派加顺序存取：在其基础上，因为有集群的概念，所以还有一个系统正确性需要保证，所以顺序非常重要！

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Y2IDMJqk-1569942793689)(assets/1557654192309.png)]

3）Basic Paxos

Paxos算法是莱斯利·兰伯特（Leslie Lamport，就是 LaTeX 中的"La"，此人现在在微软研究院）于1990年提出的一种基于消息传递的一致性算法。这个算法被认为是类似算法中最有效的。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bxNOl6yK-1569942793690)(assets/1557654273165.png)]

（1）角色介绍：

Client：产生议题者或者是请求发起者，类似于民众

Proposer ：提议者，接收民众的建议并提出议案，在冲突时有调节作用，类似于议员

Acceptor：决策者或者是投票者，国会成员或者议员等

Learner：最终决策记录者备份，对集群一致性没有影响

（2）基本流程：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-otEQsWdA-1569942793692)(assets/1557654744470.png)]

（3）全票通过的场景

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kNdkb59k-1569942793693)(assets/1557654834498.png)]

（4）部分Accepter通过

但是达到了1/2以上的节点，提案通过，反之则不通过

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4GKtIMXs-1569942793696)(assets/1557654922733.png)]

（5）proposer宕机或者失败

当前提案不会通过，但是有其他的proposer处理（客户端需要有重试机制）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vF8XhzwB-1569942793698)(assets/1557655222660.png)]

（6）活锁问题的产生

https://www.cnblogs.com/sunnyCx/p/8108366.html

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GlBqF2Fb-1569942793699)(assets/1557655257127.png)]

timeout解决

4）Multi Paxos

（1）Basix Paxos问题

实现难效率低（2轮RPC，来回验证，效率低）活锁

（2）区别

Leader：唯一propser，选举产生所有的请求都经过leader

（3）基本流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SfADd2Sk-1569942793700)(assets/1557655757195.png)]

（4）简化角色

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5B751Y18-1569942793702)(assets/1557655840660.png)]

10.4 Raft

http://thesecretlivesofdata.com/raft/

1）角色

（1）Leader

（2）Follwer

（3）Candidate

2）三个场景

（1）Leader Election

（2）Log Repliction

（3）Safety

10.5 ZAB

基本与Raft相同叫法的区别：zab将某个一个leader称为epoch，而Raft称之为term 心跳方向为leader向follower。ZAB则相反

1）协议原理

ZAB主要包括消息广播和崩溃恢复两个过程，进一步可以分为三个阶段，分别是发现（Discovery）、同步（Synchronization）、广播（Broadcast）阶段。ZAB的每一个分布式进程会循环执行这三个阶段，称为主进程周期。

· 发现，选举产生PL(prospective leader)，PL收集Follower epoch(cepoch)，根据Follower的反馈，PL产生newepoch(每次选举产生新Leader的同时产生新epoch)。

· 同步，PL补齐相比Follower多数派缺失的状态、之后各Follower再补齐相比PL缺失的状态，PL和Follower完成状态同步后PL变为正式Leader(established leader)。

· 广播，Leader处理客户端的写操作，并将状态变更广播至Follower，Follower多数派通过之后Leader发起将状态变更落地(deliver/commit)。

在正常运行过程中，ZAB协议会一直运行于阶段三来反复进行消息广播流程，如果出现崩溃或其他原因导致Leader缺失，那么此时ZAB协议会再次进入发现阶段，选举新的Leader。

2）运行状态

每个进程都有可能处于如下三种状态之一

· LOOKING：Leader选举阶段。

· FOLLOWING：Follower服务器和Leader服务器保持同步状态。

· LEADING：Leader服务器作为主进程领导状态。

最新回复(0)