rpc系列-ZooKeeper

一.简介

Zookeeper是一个分布式协调服务，就是为用户的分布式应用程序提供协调服务。

Zookeeper本身就是一个分布式程序（只要有半数以上节点存活，zk就能正常服务）。

Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、统一配置管理、分布式共享锁、统一名称服务……

虽然说可以提供各种服务，但是zookeeper在底层其实只提供了两个功能：

管理(存储，读取)用户程序提交的数据；

并为用户程序提供数据节点监听服务；

Zookeeper集群的角色： Leader 和 follower （Observer）

只要集群中有半数以上节点存活，集群就能提供服务

二.结构

特性

1.Zookeeper：一个leader，多个follower组成的集群

2.全局数据一致：每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的

3.分布式读写，更新请求转发，由leader实施

4.更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行

5.数据更新原子性，一次数据更新要么成功，要么失败

6.实时性，在一定时间范围内，client能读到最新数据

结构

1.层次化的目录结构，命名符合常规文件系统规范(见下图)

2.每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识

3.节点Znode可以包含数据和子节点（但是EPHEMERAL类型的节点不能有子节点）

4.客户端应用可以在节点上设置监视器

节点类型

1.Znode有两种类型：

短暂（ephemeral）（断开连接自己删除）

持久（persistent）（断开连接不删除）

2.Znode有四种形式的目录节点（默认是persistent ）

PERSISTENT

PERSISTENT_SEQUENTIAL（持久序列/test0000000019 ）

EPHEMERAL

EPHEMERAL_SEQUENTIAL

3.创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护。

4.在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序。

三.原理

zookeeper的选举机制（集群paxos）

以一个简单的例子来说明整个选举的过程.
假设有五台服务器组成的zookeeper集群,它们的id从1-5,同时它们都是最新启动的,也就是没有历史数据,在存放数据量这一点上,都是一样的.假设这些服务器依序启动,来看看会发生什么.
1. 服务器1启动,此时只有它一台服务器启动了,它发出去的报没有任何响应,所以它的选举状态一直是LOOKING状态。

2.服务器2启动,它与最开始启动的服务器1进行通信,互相交换自己的选举结果,由于两者都没有历史数据,所以id值较大的服务器2胜出,但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3),所以服务器1,2还是继续保持LOOKING状态。

3.服务器3启动,根据前面的理论分析,服务器3成为服务器1,2,3中的老大,而与上面不同的是,此时有三台服务器选举了它,所以它成为了这次选举的leader。

4. 服务器4启动,根据前面的分析,理论上服务器4应该是服务器1,2,3,4中最大的,但是由于前面已经有半数以上的服务器选举了服务器3,所以它只能接收当小弟的命了。
5.服务器5启动,同4一样,当小弟.

非全新集群的选举机制(数据恢复)

那么，初始化的时候，是按照上述的说明进行选举的，但是当zookeeper运行了一段时间之后，有机器down掉，重新选举时，选举过程就相对复杂了。

需要加入数据id、leader id和逻辑时钟。

数据id：数据新的id就大，数据每次更新都会更新id。

Leader id：就是我们配置的myid中的值，每个机器一个。

逻辑时钟：这个值从0开始递增,每次选举对应一个值,也就是说: 如果在同一次选举中,那么这个值应该是一致的 ; 逻辑时钟值越大,说明这一次选举leader的进程更新。

选举的标准就变成：

1.逻辑时钟小的选举结果被忽略，重新投票

2.统一逻辑时钟后，数据id大的胜出

3.数据id相同的情况下，leader id大的胜出

根据这个规则选出leader。

四.示例

Zookeeper的监听器工作机制

监听器是一个接口，我们的代码中可以实现Wather这个接口，实现其中的process方法，方法中即我们自己的业务逻辑

监听器的注册是在获取数据的操作中实现：

getData(path,watch)监听的事件是：节点数据变化事件

getChildren(path,watch)监听的事件是：节点下的子节点增减变化事件

服务端

public class DistributedServer {private static final String host = "localhost:2181";private static final int sessionTimeout = 2000;private static final String parentNode = "/servers/";private ZooKeeper zk = null;/*** 创建到zk的客户端连接** @throws Exception*/public void getConnect() throws Exception {zk = new ZooKeeper(host, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent event) {// 收到事件通知后的回调函数System.out.println(event.getType() + "__" + event.getPath());try {zk.getChildren("/", true);} catch (Exception e) {}}});}/*** 向zk集群注册服务器信息* ZooDefs.Ids.OPEN_ACL_UNSAFE   默认匿名权限，权限scheme id：'world,'anyone，权限位：31（adcwr）* ZooDefs.Ids.READ_ACL_UNSAFE   只读权限，权限scheme id：'world,'anyone，权限位：1（r）** CreateMode* 节点类型，类型定义在枚举CreateMode中：* （1）PERSISTENT：持久；* （2）PERSISTENT_SEQUENTIAL：持久顺序；* （3）EPHEMERAL：临时；* （4）EPHEMERAL_SEQUENTIAL：临时顺序。* @param data  创建节点初始化内容* @throws Exception*/public void registerServer(String data) throws Exception {String create = zk.create(parentNode + "test", data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println(data + " 注册节点 " + create);}/*** 业务功能** @throws InterruptedException*/public void handleBussiness(String data) throws InterruptedException {System.out.println(data + "开始handleBussiness");Thread.sleep(Long.MAX_VALUE);}public static void main(String[] args) throws Exception {// 获取zk连接DistributedServer server = new DistributedServer();server.getConnect();// 利用zk连接注册服务器信息server.registerServer("test01");// 启动业务功能server.handleBussiness("test01");}}

分布式锁

/****分布式锁：几种实现方式，示例用临时顺序节点实现共享锁的一般做法** 逻辑* 1.zk上注册一个"临时+序号"的znode,并监听父节点* 2.获取父节点下所有程序子节点，比较序号大小* 3.序号最小的获取到"锁"，去访问资源，访问完后，删除自己的节点，释放锁，重新注册一个新的子节点* 4.其他程序节点会收到事件通知，可以去zk上获取锁*/
public class DistributedClientLock {// 会话超时private static final int SESSION_TIMEOUT = 2000;// zookeeper集群地址private String hosts = "localhost:2181";private String groupNode = "servers";private String subNode = "test";private boolean haveLock = false;private ZooKeeper zk;/*** 记录自己创建的子节点路径* volatile 不是线程安全的，具有可见性，在一个子内存操作完后，立即刷新回到主内存。* 如果不加Volatile，每次调用thisPath，会有副本，修改会有延迟，比如其它线程抢到没有修改完的数据，就在新的线程继续执行，造成最后数据值有误* 比如：一个线程写，其它线程去读的时候，用的Volatile，比如监听新节点插入。*/private volatile String thisPath;/*** 连接zookeeper*/public void connectZookeeper() throws Exception {zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {public void process(WatchedEvent event) {try {System.out.println(event.getType()+"____"+event.getPath());/*** 判断事件类型，此处只处理子节点变化事件*                          event For “/path”                           event For “/path/child”* create(“/path”)          EventType.NodeCreated                       无* delete(“/path”)          EventType.NodeDeleted       	            无* setData(“/path”)         EventType.NodeDataChanged                   无* create(“/path/child”)    EventType.NodeChildrenChanged（getChild)     EventType.NodeCreated* delete(“/path/child”)    EventType.NodeChildrenChanged（getChild)     EventType.NodeDeleted* setData(“/path/child”)   无                                           EventType.NodeDataChanged*/if (event.getType() == Event.EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {//获取子节点，并对父节点进行监听List<String> childrenNodes = zk.getChildren("/" + groupNode, true);String thisNode = thisPath.substring(("/" + groupNode + "/").length());// 去比较是否自己是最小idCollections.sort(childrenNodes);if (childrenNodes.indexOf(thisNode) == 0) {//访问共享资源处理业务，并且在处理完成之后删除锁doSomething();//重新注册一把新的锁thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);}}} catch (Exception e) {e.printStackTrace();}}});// 程序一进来就先注册一把锁到zk上thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);// wait一小会，便于观察Thread.sleep(new Random().nextInt(1000));// 从zk的锁父目录下，获取所有子节点，并且注册对父节点的监听List<String> childrenNodes = zk.getChildren("/" + groupNode, true);//如果争抢资源的程序就只有自己，则可以直接去访问共享资源if (childrenNodes.size() == 1) {doSomething();thisPath = zk.create("/" + groupNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);}}/*** 处理业务逻辑，并且在最后释放锁*/private void doSomething() throws Exception {try {System.out.println("锁: " + thisPath);Thread.sleep(2000);} finally {System.out.println("完成: " + thisPath);//删除当前节点zk.delete(this.thisPath, -1);}}public static void main(String[] args) throws Exception {DistributedClientLock dl = new DistributedClientLock();dl.connectZookeeper();Thread.sleep(Long.MAX_VALUE);}
}