《Zookeeper 分布式过程协同技术详解》读书笔记-2

zk的一些内部原理和应用
- 请求，事务和标识
- - 读写操作
  - 事务标识（zxid）
- 群首选举
- Zab协议（ZooKeeper Atomic Broadcast protocol）
- 文件系统和监听通知机制
- - 分布式配置中心, 简单Demo
  - - java code
  - 集群管理
  - - code
  - 分布式锁

zk的一些内部原理和应用

请求，事务和标识

读写操作

只读请求（exists, getData,getChildren）会在zk服务其本地处理，然后返回给客户端（所以zk处理以只读请求为主要负载时，性能会很高，可以增加更多的服务器到zk集群，这样能处理更多的读请求）

因为ZooKeeper集群中所有的server节点都拥有相同的数据，所以读的时候可以在任意一台server节点上，客户端连接到集群中某一节点，读请求，然后直接返回。当然因为ZooKeeper协议的原因（一半以上的server节点都成功写入了数据，这次写请求便算是成功），读数据的时候可能会读到数据不是最新的server节点，所以比较推荐使用watch机制，在数据改变时，及时感知到

写操作(create, delete, setData) 将会被转发给群首（群首会执行相应请求，并形成状态更新，我们称为事务(Transaction)）

在这里插入图片描述

Client向Zookeeper的server1发送一个写请求，客户端写数据到服务器1上；
如果server1不是Leader，那么server1会把接收到的写请求转发给Leader；然后Leader会将写请求转发给每个server；
- server1和server2负责写数据，并且两个Follower的写入数据是一致的，保存相同的数据副本；
- server1和server2写数据成功后，通知Leader；
当Leader收到集群半数以上的节点写成功的消息后，说明该写操作执行成功；
- 这里是3台服务器，只要2台Follower服务器写成功就ok
- 因为client访问的是server1，所以Leader会告知server1集群中数据写成功；
被访问的server1进一步通知client数据写成功，这时，客户端就知道整个写操作成功了

事务标识（zxid）

当群首产生了一个事务，就会为该事务分配一个标识符，我们称之为ZooKeeper会话ID（zxid），通过Zxid对事务进行标识，就可以按照群首所指定的顺序在各个服务器中按序执行

zxid为一个long型（64位）整数，分为两部分：时间戳（epoch）部分和计数器（counter）部分。每个部分为32位，在我们讨论zab(Zookeeper Atomic Broadcast )协议 时，我们就会发现时间戳（epoch）和计数器（counter）的具体作用，我们通过该协议来广播各个服务器的状态变更信息。

eg:

setData 加上事务（版本和新数据值），一个事务为一个单位，以原子方式执行；ZooKeeper集群以事务方式运行，并确保所有的变更操作以原子方式被执行，同时不会被其他事务所干扰；并不存在传统的关系数据库中所涉及的回滚机制，而是确保事务的每一步操作都互不干扰

同时一个事务还具有幂等性，也就是说，我们可以对同一个事务执行两次，我们得到的结果还是一样的，我们甚至还可以对多个事务执行多次，同样也会得到一样的结果，前提是我们确保多个事务的执行顺序每次都是一样的。事务的幂等性可以让我们在进行恢复处理时更加简单。

实际上，ZooKeeper的每个节点维护者两个Zxid值，为别为：cZxid、mZxid。

（1）cZxid：是节点的创建时间所对应的Zxid格式时间戳。

（2）mZxid：是节点的修改时间所对应的Zxid格式时间戳。

高32位是epoch用来标识Leader关系是否改变，每次一个Leader被选出来，它都会有一个新的epoch。低32位是个递增计数。

群首选举

群首为集群中的服务器选择出来的一个服务器，并会一直被集群所认可。设置群首的目的是为了对客户端所发起的ZooKeeper状态变更请求进行排序，包括：create、setData和delete操作。群首将每一个请求转换为一个事务，将这些事务发送给追随者，确保集群按照群首确定的顺序接受并处理这些事务。

每个服务器启动后进入LOOKING状态，开始选举一个新的群首或查找已经存在的群首，如果群首已经存在，其他服务器就会通知这个新启动的服务器，告知哪个服务器是群首，与此同时，新的服务器会与群首建立连接，以确保自己的状态与群首一致。
如果集群中所有的服务器均处于LOOKING状态，这些服务器之间就会进行通信来选举一个群首，通过信息交换对群首选举达成共识的选择。在本次选举过程中胜出的服务器将进入LEADING状态，而集群中其他服务器将会进入FOLLOWING状态。

当一个服务器进入LOOKING状态，就会发送向集群中每个服务器发送一个通知消息,如下

投票<服务器标识，最近执行的事务的zxid信息>
（vote<sid, zxid>）

在这里插入图片描述

快速群首选举的快速指的是什么？

Zab协议（ZooKeeper Atomic Broadcast protocol）

在接收到一个写请求操作后，追随者会将请求转发给群首，群首将探索性地执行该请求，并将执行结果以事务的方式对状态更新进行广播。一个事务中包含服务器需要执行变更的确切操作，当事务提交时，服务器就会将这些变更反馈到数据树上，其中数据树为ZooKeeper用于保存状态信息的数据结构（请参考DataTree类）。

之后我们需要面对的问题便是服务器如何确认一个事务是否已经提交，由此引入了我们所采用的协议：Zab：ZooKeeper原子广播协议（ZooKeeper Atomic Broadcast protocol）。

propose: leader 广播给 followers
accept: followers收到广播信息给leader发送确认消息
commit: leader收到确认仲裁数量后发送消息给follower进行提交操作

具体如下：

两个保障（依靠了zxid）

一个被选举的群首确保在提交完所有之前的时间戳内需要提交的事务，之后才开始广播新的事务。
在任何时间点，都不会出现两个被仲裁支持的群首。

为了实现第一个需求，群首并不会马上处于活动状态，直到确保仲裁数量的服务器认可这个群首新的时间戳值。一个时间戳的最初状态必须包含所有的之前已经提交的事务，或者某些已经被其他服务器接受，但尚未提交完成的事务。这一点非常重要，在群首进行时间戳e的任何新的提案前，必须保证自时间戳开始值到时间戳e－1内的所有提案被提交。如果一个提案消息处于时间戳e’<e，在群首处理时间戳e的第一个提案消息前没有提交之前的这个提案，那么旧的提案将永远不会被提交。

文件系统和监听通知机制

分布式配置中心, 简单Demo

在zookeeper里增加一个目录节点，并把配置信息存储在里面(作为配置中心存储)

在这里插入图片描述

多个客户端能够读取到
服务端配置改变，客户端(所有注册监听的客户端)都能监听到事件

java code

package demo;import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;import java.util.concurrent.CountDownLatch;/*** @Author mubi* @Date 2020/3/27 22:36*/
public class ZooKeeperProSync implements Watcher {private static CountDownLatch connectedSemaphore = new CountDownLatch(1);private static ZooKeeper zk = null;private static Stat stat = new Stat();public static void main(String[] args) throws Exception {// zookeeper配置数据的存放路径String path = "/myconfig";// 连接zookeeper并且注册一个监听器zk = new ZooKeeper("127.0.0.1:2281", 5000, new ZooKeeperProSync());// 等待zk连接成功的通知（等待connectedSemaphore.countDown()减少为0）connectedSemaphore.await();// 获取path目录节点的配置数据，并注册对节点的监听System.out.println(new String(zk.getData(path, true, stat)));Thread.sleep(Integer.MAX_VALUE);}@Overridepublic void process(WatchedEvent event) {// zk连接成功通知事件if (KeeperState.SyncConnected == event.getState()) {if (EventType.None == event.getType() && null == event.getPath()) {connectedSemaphore.countDown();}else if (event.getType() == EventType.NodeDataChanged) {// zk目录节点数据变化通知事件try {System.out.println("配置已修改，新值为：" + new String(zk.getData(event.getPath(), true, stat)));// TODO 具体业务} catch (Exception e) {}}}}
}

集群管理

集群管理原理：机器的加入/退出，选举leader节点

在这里插入图片描述

持久节点 / 临时节点
有序节点 / 无序节点

可以临时+有序构成server，然后最小编号节点作为leader节点

创建集群节点目录（持久节点）
启动服务，并加入和减少机器观察

code

AppMaster

package demo;import java.util.ArrayList;
import java.util.List;import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;/*** @Author mubi* @Date 2020/3/27 23:43*/
public class AppMaster {private String clusterNode = "mycluster";private ZooKeeper zk;private volatile List<String> serverList;public void connectZookeeper() throws Exception{// 注册全局默认watcherzk = new ZooKeeper("127.0.0.1:2281", 5000, new Watcher(){@Overridepublic void process(WatchedEvent event){if (event.getType() == EventType.NodeChildrenChanged&& ("/" + clusterNode).equals(event.getPath())){try{updateServerList();}catch (Exception e){e.printStackTrace();}}}});updateServerList();}private void updateServerList() throws Exception{List<String> newServerList = new ArrayList<String>();// watcher注册后，只能监听事件一次，参数true表示继续使用默认watcher监听事件List<String> subList = zk.getChildren("/" + clusterNode, true);for (String subNode : subList){// 获取节点数据byte[] data = zk.getData("/" + clusterNode + "/" + subNode, false, null);newServerList.add(new String(data, "utf-8"));}serverList = newServerList;System.out.println("server list updated: " + serverList);}public static void main(String[] args) throws Exception{AppMaster ac = new AppMaster();ac.connectZookeeper();Thread.sleep(Long.MAX_VALUE);}
}

AppServer

package demo;import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;/*** @Author mubi* @Date 2020/3/27 23:44*/
public class AppServer extends Thread
{private String clusterNode = "mycluster";private String serverNode = "server_address";private String serverName;@Overridepublic void run(){try {connectZookeeper(serverName);} catch (Exception e) {e.printStackTrace();}}public void connectZookeeper(String address) throws Exception{ZooKeeper zk = new ZooKeeper("127.0.0.1:2281", 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {}});// 关键方法，创建包含自增长id名称的目录，这个方法支持了分布式锁的实现// 四个参数：// 1、目录名称 2、目录文本信息// 3、文件夹权限，Ids.OPEN_ACL_UNSAFE表示所有权限// 4、目录类型，CreateMode.EPHEMERAL_SEQUENTIAL表示创建临时目录，session断开连接则目录自动删除String createdPath = zk.create("/" + clusterNode + "/" + serverNode,address.getBytes("utf-8"),Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println("create: " + createdPath);Thread.sleep(Integer.MAX_VALUE);}public AppServer(String serverName){this.serverName = serverName;}
}

Server1

package demo;/*** @Author mubi* @Date 2020/3/27 23:46*/public class Server1
{public static void main(String[] args) throws Exception{AppServer server1 = new AppServer("Server1");server1.start();}
}